《AI Agent 工程实战》4.3 工具权限与审批：高危工具二次确认

# minagent/permissions.py
from minagent import Tool, ToolCall


class ToolTier:
    READONLY = "readonly"
    WRITE = "write"
    DANGEROUS = "dangerous"


class ApprovalRequired(Exception):
    def __init__(self, tc: ToolCall) -> None:
        super().__init__(tc.name)
        self.tc = tc


class PermissionedRegistry:
    """包一层 ToolRegistry：每个工具挂一个权限档，危险档必须拿到审批才放行。"""

    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}
        self._tier: dict[str, str] = {}

    def add(self, tool: Tool, tier: str) -> None:
        self._tools[tool.name] = tool
        self._tier[tool.name] = tier

    def tier_of(self, name: str) -> str:
        return self._tier[name]

    def schemas(self):
        return [{"name": t.name, "description": t.description,
                 "parameters": t.parameters} for t in self._tools.values()]

    def call(self, tc: ToolCall, approved: frozenset[str] = frozenset()) -> str:
        if self._tier[tc.name] == ToolTier.DANGEROUS and tc.id not in approved:
            raise ApprovalRequired(tc)          # 拦截点：没有审批就不执行
        return str(self._tools[tc.name].func(**tc.arguments))

# minagent/permissions.py（续）
import json
from minagent import Message, RunResult, Step


def run_with_approval(model, registry, user_input, approver,
                      system="", max_steps=8):
    messages: list[Message] = []
    if system:
        messages.append(Message(role="system", content=system))
    messages.append(Message(role="user", content=user_input))

    steps: list[Step] = []
    approved: set[str] = set()

    for i in range(max_steps):
        resp = model.complete(messages, registry.schemas())
        msg = resp.message
        messages.append(msg)
        if not msg.tool_calls:
            steps.append(Step(i, "final", msg.content))
            return RunResult(messages, steps, msg.content)

        for tc in msg.tool_calls:
            try:
                result = registry.call(tc, frozenset(approved))
            except ApprovalRequired as ar:
                steps.append(Step(i, "approval_required",
                                  f"{tc.name}({tc.arguments})"))
                if not approver(ar.tc):          # 挂起等人工；被拒走错误回灌
                    denied = json.dumps({"error": "approval_denied",
                                         "tool": tc.name})
                    messages.append(Message(role="tool", content=denied,
                                            tool_call_id=tc.id))
                    steps.append(Step(i, "approval_denied", tc.name))
                    continue
                approved.add(tc.id)
                steps.append(Step(i, "approval_granted", tc.name))
                result = registry.call(tc, frozenset(approved))
            messages.append(Message(role="tool", content=result,
                                    tool_call_id=tc.id))
            steps.append(Step(i, "tool_call",
                              f"{tc.name}({tc.arguments}) -> {result}"))

    steps.append(Step(max_steps, "error", "达到 max_steps 仍未收敛"))
    return RunResult(messages, steps, "")

# demo_43.py（节选）
reg = PermissionedRegistry()
reg.add(Tool("search", "检索本地语料", {...},
             func=lambda q: f"命中 3 条与『{q}』相关的片段"), ToolTier.READONLY)
reg.add(Tool("delete_run", "永久删除一次运行及其产物", {...},
             func=lambda run_id: f"已删除 {run_id}"), ToolTier.DANGEROUS)

model = FakeModel([
    [ToolCall("c1", "search", {"q": "run_42 摘要"})],
    [ToolCall("c2", "delete_run", {"run_id": "run_42"})],
    "已检索并清理 run_42，报告如下：……",
])

def approver(tc):
    print(f"  [人工审批] 工具 {tc.name}({tc.arguments}) 需要确认 -> 通过")
    return True

result = run_with_approval(model, reg, "清理 run_42 并给我报告",
                           approver, system="你是研究助理。")
for s in result.steps:
    print(f"[{s.index}] {s.kind:<17} {s.detail}")
print("最终:", result.final)
print("模型调用次数:", len(model.calls))

  [人工审批] 工具 delete_run({'run_id': 'run_42'}) 需要确认 -> 通过
[0] tool_call         search({'q': 'run_42 摘要'}) -> 命中 3 条与『run_42 摘要』相关的片段
[1] approval_required delete_run({'run_id': 'run_42'})
[1] approval_granted  delete_run
[1] tool_call         delete_run({'run_id': 'run_42'}) -> 已删除 run_42
[2] final             已检索并清理 run_42，报告如下：……
最终: 已检索并清理 run_42，报告如下：……
模型调用次数: 3

{
  "tool_call_id": "c2",
  "tool": "delete_run",
  "arguments": {"run_id": "run_42"},
  "workspace_id": "ws_acme",
  "status": "pending",
  "expires_at": "2026-09-26T14:15:00+08:00"
}