跳转至主要内容

AI 智能体与 MCP 工具的授权

AI 智能体在运行时决定调用哪些工具、使用哪些参数。模型可能出错，也可能被它在工单、网页或文档中读到的文字诱导。授权必须放在模型之外、在工具调用实际执行的地方，并回答两个问题：

  1. 这个智能体能否用这些参数调用这个工具？ 客服助手可以读取工单、发起小额退款，但绝不应该执行 shell 命令。
  2. 它所代表的用户能否做这件事？ 代表某人行事的智能体，权限不能超过这个人本身。

Casbin 用一个模型和一份策略同时解决这两个问题，方式和处理用户时一样。本页先介绍规则，再把它们接入一个普通的工具调用循环和一个 MCP 服务器。代码使用 Python。模型和策略在其他 Casbin 实现中同样适用；在 Go 中，把参数作为结构体传入，并在条件中使用导出的字段名，例如 r.args.Amount。

1. 安装​

pip install pycasbin

2. 编写模型​

保存为 model.conf：

[request_definition]
r = sub, tool, args

[policy_definition]
p = sub, tool, cond, eft

[role_definition]
g = _, _

[policy_effect]
e = some(where (p.eft == allow)) && !some(where (p.eft == deny))

[matchers]
m = g(r.sub, p.sub) && globMatch(r.tool, p.tool) && (p.cond == '' || eval(p.cond))
  • r = sub, tool, args：调用方（智能体或用户）、工具名，以及作为对象的工具参数。
  • p = sub, tool, cond, eft：每条规则指定一个角色、一个工具模式、一个可选的参数条件，以及允许还是拒绝。
  • g = _, _：智能体和用户都有角色，角色可以继承其他角色。
  • globMatch 用通配符匹配工具名，例如 tickets.* 或 *.delete。
  • 策略效果规定：只要有规则允许 并且没有规则拒绝，调用就被允许，所以拒绝规则总是优先。

3. 编写策略​

保存为 policy.csv：

p, support-agent, tickets.*, , allow
p, support-agent, orders.read, , allow
p, support-agent, orders.refund, r.args.amount <= 100, allow
p, any-agent, shell.*, , deny
p, any-agent, *.delete, , deny

p, support-staff, tickets.*, , allow
p, support-staff, orders.*, , allow

g, support-agent, any-agent
g, triage-agent, any-agent
p, triage-agent, tickets.read, , allow
p, triage-agent, tickets.label, , allow

g, alice, support-staff

从上往下看：

  • support-agent 可以使用所有工单工具、读取订单，每次最多退款 100。条件 r.args.amount \<= 100 针对模型生成的参数求值。
  • 所有继承自 any-agent 的角色都禁止使用 shell 工具和任何以 .delete 结尾的工具，无论还被授予了什么。把适用于所有智能体的护栏规则放在这里。
  • support-staff 是人类角色，alice 拥有它。
  • triage-agent 是第二个权限更窄的智能体，只能读取工单和打标签。
提示

条件中的字符串字面量用单引号，例如 r.args.region == 'eu'。如果条件中需要逗号，用双引号把整个字段括起来。

4. 检查每次工具调用​

authorize_tool_call 做两项检查。智能体自身的权限限制它能做的全部事情，用户的权限限制它能为这个用户做的事情。只有两项都通过的调用才会执行。

tools.py
from types import SimpleNamespace

import casbin

enforcer = casbin.Enforcer("model.conf", "policy.csv")


class ToolDenied(Exception):
    pass


def authorize_tool_call(agent: str, user: str, tool: str, args: dict) -> None:
    """Allow the call only if both the agent and the user it acts for may make it."""
    request_args = SimpleNamespace(**args)
    if not enforcer.enforce(agent, tool, request_args):
        raise ToolDenied(f"{agent} may not call {tool} with {args}")
    if not enforcer.enforce(user, tool, request_args):
        raise ToolDenied(f"{user} may not call {tool}, so an agent acting for them may not either")


# The tools your agent can see. In a real application they call your services.
TOOLS = {
    "tickets.read": lambda ticket_id: {"id": ticket_id, "subject": "Refund request", "order_id": "A-1001"},
    "orders.read": lambda order_id: {"id": order_id, "total": 240},
    "orders.refund": lambda order_id, amount: {"order_id": order_id, "refunded": amount},
    "shell.exec": lambda command: {"output": "..."},
}


def call_tool(agent: str, user: str, tool: str, args: dict) -> dict:
    """Run a tool call proposed by the model, or return the denial as the tool result."""
    try:
        authorize_tool_call(agent, user, tool, args)
    except ToolDenied as e:
        # Returning the reason lets the model explain it or try something else.
        return {"error": "permission_denied", "reason": str(e)}
    return TOOLS[tool](**args)


if __name__ == "__main__":
    # Tool calls as a model might propose them while helping alice with a ticket.
    proposed = [
        ("tickets.read", {"ticket_id": "T-42"}),
        ("orders.read", {"order_id": "A-1001"}),
        ("orders.refund", {"order_id": "A-1001", "amount": 240}),
        ("orders.refund", {"order_id": "A-1001", "amount": 80}),
        ("shell.exec", {"command": "rm -rf /"}),
    ]
    for tool, args in proposed:
        print(tool, args, "->", call_tool("support-agent", "alice", tool, args))

    print()
    print("bob:", call_tool("support-agent", "bob", "tickets.read", {"ticket_id": "T-42"}))

运行后输出：

tickets.read {'ticket_id': 'T-42'} -> {'id': 'T-42', 'subject': 'Refund request', 'order_id': 'A-1001'}
orders.read {'order_id': 'A-1001'} -> {'id': 'A-1001', 'total': 240}
orders.refund {'order_id': 'A-1001', 'amount': 240} -> {'error': 'permission_denied', 'reason': "support-agent may not call orders.refund with {'order_id': 'A-1001', 'amount': 240}"}
orders.refund {'order_id': 'A-1001', 'amount': 80} -> {'order_id': 'A-1001', 'refunded': 80}
shell.exec {'command': 'rm -rf /'} -> {'error': 'permission_denied', 'reason': "support-agent may not call shell.exec with {'command': 'rm -rf /'}"}

bob: {'error': 'permission_denied', 'reason': 'bob may not call tickets.read, so an agent acting for them may not either'}

240 的退款被拒绝了：alice 本人可以退任意金额，但智能体不可以。 Bob 没有任何角色，所以智能体不能为他做任何事。把拒绝作为工具结果返回，而不是直接抛给用户，模型就可以解释这个限制，或把事情转交给人工。

5. 在 MCP 服务器中使用​

在使用 OAuth 的 MCP 服务器中，访问令牌已经给出了两个身份：OAuth 客户端是智能体，令牌的主体是授权它的用户。使用官方 Python SDK（第 2 版）时，用 get_access_token() 读取它们，并检查每次工具调用：

server.py
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

from tools import ToolDenied, authorize_tool_call

mcp = MCPServer("support-tools")


def authorize(tool: str, args: dict) -> None:
    """Check the tool call against Casbin for the OAuth client (the agent) and the user it acts for."""
    token = get_access_token()
    if token is None or token.subject is None:
        raise ToolError("not authenticated")
    try:
        authorize_tool_call(token.client_id, token.subject, tool, args)
    except ToolDenied as e:
        raise ToolError(str(e)) from e


@mcp.tool()
def refund_order(order_id: str, amount: float) -> dict:
    """Refund part or all of an order."""
    authorize("orders.refund", {"order_id": order_id, "amount": amount})
    return {"order_id": order_id, "refunded": amount}


@mcp.tool()
def read_ticket(ticket_id: str) -> dict:
    """Read a support ticket."""
    authorize("tickets.read", {"ticket_id": ticket_id})
    return {"id": ticket_id, "subject": "Refund request"}

抛出 ToolError 会把拒绝变成一个设置了 isError 的工具结果，MCP 客户端会把它传回给模型。令牌的 client_id 必须与策略中的智能体名称一致；把每个智能体注册为单独的 OAuth 客户端，这样可以单独给它授权，也可以单独撤销。

设计要点​

  • 默认拒绝。 没有任何规则提到的工具不能被调用，所以新工具在你授权之前对智能体不可用。
  • 每个智能体最小权限。 给每个智能体单独的角色，而不是共用一个权限很大的角色。上面的 triage-agent 即使被提示词要求，也碰不到订单。
  • 检查参数，而不只是名称。 对金额、收件人、路径或地区的限制应写在条件中。在检查之前校验参数类型：条件比较的是模型发来的任何东西，把字符串 "500" 和数字比较会抛出错误，而不是返回一个判定。
  • 不重新部署就能修改权限。 规则就是数据：enforcer.add_policy(...)、enforcer.add_role_for_user(...) 或通过 适配器 的修改，下一次调用就会生效。多个进程为智能体提供服务时，使用 watcher。
  • 审计。 enforcer.enforce_ex(...) 会返回允许这次调用的规则；把它和工具调用一起记录到日志中。
  • 过滤工具列表。 列出工具时也做同样的检查，这样模型根本不会看到它不能用的工具。这也能节省 token。

下一步​

Comments are powered by giscus and GitHub Discussions. They load from giscus.app and GitHub only after you click the button below.