🔥个人主页：爱和冰阔乐
📚专栏传送门：《数据结构与算法》 、C++
🐶学习方向：C++方向学习爱好者
⭐人生格言：得知坦然 ，失之淡然

---

🏠博主简介

---

前言

Agent 能输出一段工具参数，离“工具可以放心执行”还差很远。

最简单的工具调用通常长这样：

{
  "tool": "write_file",
  "arguments": {
    "path": "config/app.yaml",
    "content": "port: 8080"
  }
}

演示时看不出问题，接入真实工作区以后，麻烦会一起出现：

• 参数字段拼错，执行器拿到空值；
• 路径越过工作目录，写到了不该写的位置；
• 模型重复调用，订单、通知或文件被创建两次；
• 工具执行到一半失败，数据库和文件状态对不上；
• 日志只写了一句“调用失败”，无法还原当时参数；
• 测试依赖在线模型，同一个用例每次输出都不同；
• 高风险工具和只读工具走了同一套放行逻辑；
• 工具返回内容过大，又被原样塞回下一轮上下文。

所以工程里真正需要的不是一个巨大的 if tool_name == ...，而是一层独立的工具执行边界：

模型提出调用
    ↓
解析与契约校验
    ↓
权限和风险判断
    ↓
幂等检查
    ↓
工具执行
    ↓
结果裁剪与结构化
    ↓
审计落库
    ↓
返回 Agent

这层边界不负责“让模型变聪明”，它负责把不稳定的模型输出，转换成可判断、可拒绝、可追踪的系统行为。

本文用 Python 写一个小型执行器，包含下面几部分：

• 工具注册表；
• JSON Schema 参数校验；
• 工作区路径限制；
• 风险等级与能力清单；
• 幂等键；
• SQLite 审计日志；
• 超时和错误归一化；
• 录制与回放测试；
• 故障注入和验收清单。

---

一、先把模型输出当作不可信输入

1.1 工具调用不是函数调用

程序内部调用函数时，参数一般已经经过类型系统或调用方约束：

result = read_text(path="README.md", max_chars=4000)

模型给出的内容则更接近网络请求：

{
  "tool": "read_text",
  "arguments": {
    "path": 123,
    "max_chars": "全部"
  }
}

字段可能缺失、类型可能错误、字符串可能包含越界路径，甚至工具名本身都不存在。

因此入口处至少要区分四类失败：

阶段	示例	是否执行工具
请求解析失败	不是合法 JSON	否
契约校验失败	max_chars 是字符串	否
策略拒绝	路径越过工作区	否
工具执行失败	文件不存在	已进入执行

把它们都写成“工具失败”，后面无法判断到底该修改提示词、修改参数，还是修工具本身。

1.2 执行器的输入输出

先定义统一请求：

from dataclasses import dataclass, field
from typing import Any

@dataclass(slots=True)
class ToolCall:
    call_id: str
    tool_name: str
    arguments: dict[str, Any]
    actor: str
    session_id: str
    idempotency_key: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

统一结果：

@dataclass(slots=True)
class ToolResult:
    call_id: str
    ok: bool
    code: str
    message: str
    data: Any = None
    retryable: bool = False
    duration_ms: int = 0

ok 只表达成功与否，code 才用于机器判断。

例如：

INVALID_ARGUMENT
TOOL_NOT_FOUND
PERMISSION_DENIED
PATH_OUTSIDE_WORKSPACE
DUPLICATE_CALL
TIMEOUT
EXECUTION_ERROR
OK

如果下一轮需要模型自动修正参数，稳定的错误码比一大段异常堆栈更有用。

二、工具注册表不要只保存函数

2.1 一个工具需要哪些描述

工具注册表至少要记录：

from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, Literal

RiskLevel = Literal["read", "write", "external", "dangerous"]

@dataclass(slots=True)
class ToolSpec:
    name: str
    description: str
    schema: dict[str, Any]
    handler: Callable[..., Any]
    risk: RiskLevel
    timeout_seconds: float = 10.0
    result_limit: int = 8000
    idempotent: bool = False

这里几个字段不是为了文档好看：

• schema：决定参数能不能进入执行阶段；
• risk：决定是否需要额外权限或人工确认；
• timeout_seconds：防止一个工具拖住整个会话；
• result_limit：避免把几十万字符重新送给模型；
• idempotent：决定重复调用能否安全复用结果。

2.2 注册器

class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, ToolSpec] = {}

    def register(self, spec: ToolSpec) -> None:
        if spec.name in self._tools:
            raise ValueError(f"duplicate tool: {spec.name}")
        self._tools[spec.name] = spec

    def get(self, name: str) -> ToolSpec | None:
        return self._tools.get(name)

    def list_for_model(self) -> list[dict[str, Any]]:
        return [
            {
                "name": item.name,
                "description": item.description,
                "parameters": item.schema,
            }
            for item in self._tools.values()
        ]

注册和暴露给模型是两件事。

执行器可以注册十几个内部工具，但某个会话只向模型开放其中三个。这样做比在提示词里写“请不要调用危险工具”可靠得多。

三、用 JSON Schema 拦住参数漂移

3.1 读文件工具的参数契约

READ_TEXT_SCHEMA = {
    "type": "object",
    "properties": {
        "path": {
            "type": "string",
            "minLength": 1,
            "maxLength": 300
        },
        "max_chars": {
            "type": "integer",
            "minimum": 1,
            "maximum": 20000,
            "default": 4000
        }
    },
    "required": ["path"],
    "additionalProperties": False
}

additionalProperties: false 很重要。

假设模型输出：

{
  "path": "README.md",
  "max_char": 4000
}

如果允许额外字段，拼错的 max_char 会被静默忽略，工具继续使用默认值。问题表面上不报错，行为却和调用意图不一致。

3.2 校验代码

安装依赖：

python -m pip install jsonschema

校验：

from jsonschema import Draft202012Validator

def validate_arguments(
    schema: dict[str, Any],
    arguments: dict[str, Any]
) -> list[str]:
    validator = Draft202012Validator(schema)
    errors = sorted(
        validator.iter_errors(arguments),
        key=lambda item: list(item.path)
    )

    result: list[str] = []
    for error in errors:
        location = ".".join(str(part) for part in error.path) or "$"
        result.append(f"{location}: {error.message}")
    return result

测试：

bad_arguments = {
    "path": "README.md",
    "max_char": "4000"
}

print(validate_arguments(READ_TEXT_SCHEMA, bad_arguments))

输出类似：

["$: Additional properties are not allowed ('max_char' was unexpected)"]

契约校验解决的是“形状是否正确”，不解决“值是否安全”。合法字符串仍然可能是：

../../../../Windows/System32/drivers/etc/hosts

所以校验后还要进入策略层。

图里要看的不是工具数量，而是拒绝发生的位置。格式错误、参数错误和权限错误都应在真正执行前结束；只有通过前面几层的请求，才进入带超时的执行区。

四、路径安全不能依赖字符串前缀

4.1 常见错误写法

下面这种判断看起来合理：

if not user_path.startswith(str(workspace)):
    raise PermissionError("outside workspace")

但调用参数通常是相对路径，字符串前缀本身也不能处理 ..、符号链接和大小写差异。

另一个常见错误：

target = workspace / user_path

拼接不等于限制。workspace / "../secret.txt" 仍然会指向工作区外。

4.2 解析后的父子关系

from pathlib import Path

class Workspace:
    def __init__(self, root: Path) -> None:
        self.root = root.resolve(strict=True)

    def resolve(self, value: str, must_exist: bool = False) -> Path:
        candidate = (self.root / value).resolve(strict=must_exist)

        try:
            candidate.relative_to(self.root)
        except ValueError as error:
            raise PermissionError(
                f"path outside workspace: {value}"
            ) from error

        return candidate

读取工具：

def make_read_text(workspace: Workspace):
    def read_text(path: str, max_chars: int = 4000) -> dict[str, Any]:
        target = workspace.resolve(path, must_exist=True)
        if not target.is_file():
            raise IsADirectoryError(path)

        content = target.read_text(encoding="utf-8")
        truncated = len(content) > max_chars
        return {
            "path": target.relative_to(workspace.root).as_posix(),
            "content": content[:max_chars],
            "truncated": truncated,
            "total_chars": len(content)
        }

    return read_text

这里返回相对路径，不把机器的绝对目录泄露给模型。

4.3 符号链接边界

Linux 工作区里可能有：

ln -s /etc workspace/system-config

如果只检查拼接前的路径：

workspace/system-config/passwd

看起来仍在工作区，解析符号链接后却指向 /etc/passwd。因此安全检查必须对最终解析路径执行。

写文件时目标可能还不存在，处理方式是先解析父目录：

def resolve_new_file(workspace: Workspace, value: str) -> Path:
    raw = workspace.root / value
    parent = raw.parent.resolve(strict=True)
    parent.relative_to(workspace.root)
    return parent / raw.name

文件名还应额外限制空名称、保留名称和长度。不要把所有平台差异都塞进一条正则，路径解析和业务命名规则分开更容易维护。

五、能力清单比“允许所有再拉黑”稳

5.1 按会话发放能力

定义会话权限：

@dataclass(slots=True)
class CapabilitySet:
    allowed_tools: set[str]
    allowed_risks: set[RiskLevel]
    writable_prefixes: tuple[str, ...] = ()
    network_hosts: tuple[str, ...] = ()

只读代码审查会话：

review_caps = CapabilitySet(
    allowed_tools={"list_files", "read_text", "search_text"},
    allowed_risks={"read"}
)

代码修改会话：

edit_caps = CapabilitySet(
    allowed_tools={
        "list_files",
        "read_text",
        "search_text",
        "write_file"
    },
    allowed_risks={"read", "write"},
    writable_prefixes=("src/", "tests/")
)

模型即使知道 run_shell 这个名字，只要当前能力集中没有，就不能执行。

5.2 风险不是只有“安全”和“危险”

可以先用四级分类：

等级	示例	默认策略
read	搜索、读取、列目录	会话授权后直接执行
write	修改文件、写数据库	限路径、审计、支持回滚
external	发消息、提交表单、调用第三方 API	明确目标与数据后确认
dangerous	删除、改权限、执行任意命令	默认不开放或逐次确认

分类不必追求一次定死。关键是不要让“读取 README”和“删除目录”共用一个布尔开关。

5.3 策略判断

def authorize(spec: ToolSpec, caps: CapabilitySet) -> None:
    if spec.name not in caps.allowed_tools:
        raise PermissionError(f"tool not allowed: {spec.name}")

    if spec.risk not in caps.allowed_risks:
        raise PermissionError(f"risk not allowed: {spec.risk}")

写文件工具还要判断具体目标：

def check_write_prefix(
    relative_path: str,
    prefixes: tuple[str, ...]
) -> None:
    normalized = relative_path.replace("\\", "/").lstrip("/")
    if not any(normalized.startswith(prefix) for prefix in prefixes):
        raise PermissionError(
            f"write target not allowed: {relative_path}"
        )

实际项目里可以把策略结果写成：

{
  "decision": "deny",
  "policy": "workspace.write_prefix",
  "reason": "docs/private.md is outside allowed prefixes"
}

这比只抛一个 PermissionError 更利于审计和统计。

六、幂等不只是“请求 ID 不重复”

6.1 为什么 Agent 更容易重复调用

重复可能来自：

• 模型没有看到上一轮工具结果；
• 网络超时后上层重试；
• 执行成功但响应写回失败；
• 用户刷新页面；
• 工作流恢复时从错误节点重新开始；
• 模型根据相似错误再次生成同一调用。

只读工具重复通常只是浪费时间，外部副作用工具可能重复发送消息、重复创建工单或重复扣减库存。

6.2 幂等键的组成

不要直接使用模型生成的 call_id 作为业务幂等键。call_id 表示一次调用记录，幂等键表示“这些调用在业务上是不是同一件事”。

import hashlib
import json

def make_idempotency_key(
    tool_name: str,
    arguments: dict[str, Any],
    scope: str
) -> str:
    canonical = json.dumps(
        arguments,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":")
    )
    raw = f"{scope}\n{tool_name}\n{canonical}".encode("utf-8")
    return hashlib.sha256(raw).hexdigest()

scope 可以是任务 ID、工单 ID 或业务操作 ID。不能永远用整个会话 ID，否则用户在同一会话里两次合法修改同一文件，也可能被错误拦截。

6.3 哪些工具可以缓存结果

工具	能否直接复用旧结果	原因
读取固定版本文件	可以考虑	输入版本不变时结果稳定
搜索当前工作区	谨慎	文件状态可能已变化
写入同样内容	可以设计为幂等	需要校验目标当前状态
发送通知	必须使用业务幂等键	重复发送有副作用
执行任意 Shell	通常不能	副作用不可推断

幂等是工具契约的一部分，不是执行器可以替所有工具自动猜出来的属性。

七、审计日志要能还原一次调用

7.1 表结构

CREATE TABLE IF NOT EXISTS tool_call_audit (
    call_id TEXT PRIMARY KEY,
    session_id TEXT NOT NULL,
    actor TEXT NOT NULL,
    tool_name TEXT NOT NULL,
    risk TEXT NOT NULL,
    arguments_json TEXT NOT NULL,
    arguments_hash TEXT NOT NULL,
    idempotency_key TEXT,
    decision TEXT NOT NULL,
    result_code TEXT,
    result_preview TEXT,
    error_message TEXT,
    started_at TEXT NOT NULL,
    finished_at TEXT,
    duration_ms INTEGER,
    replay_of TEXT
);

CREATE INDEX IF NOT EXISTS idx_audit_session
ON tool_call_audit(session_id, started_at);

CREATE INDEX IF NOT EXISTS idx_audit_idempotency
ON tool_call_audit(idempotency_key);

不要默认把完整返回值全部写入审计表。返回值可能包含大文件、隐私数据或第三方响应。

比较稳妥的做法：

参数：按工具做字段脱敏后保存
结果：保存状态、长度、摘要和有限预览
大结果：单独放对象存储，审计表只保存引用
密钥：绝不进入普通审计日志

7.2 字段脱敏

SENSITIVE_KEYS = {
    "password",
    "token",
    "api_key",
    "authorization",
    "secret"
}

def redact(value: Any) -> Any:
    if isinstance(value, dict):
        return {
            key: "***" if key.lower() in SENSITIVE_KEYS else redact(item)
            for key, item in value.items()
        }
    if isinstance(value, list):
        return [redact(item) for item in value]
    return value

单纯按键名脱敏不可能覆盖所有情况，但至少能挡住最明显的配置字段。对于发送邮件、数据库查询等工具，应在工具定义中指定自己的脱敏函数。

7.3 审计不是调试日志

调试日志可能随版本调整，审计记录应保持稳定字段。

一次调用最好能回答：

谁提出的调用？
属于哪个会话和任务？
调用了什么工具？
参数经过什么规则处理？
为什么允许或拒绝？
实际执行了多长时间？
结果码是什么？
是否命中过幂等记录？
能否找到对应的重放用例？

图里的秤代表策略判断，抽屉代表审计记录。工具调用不是从模型直接掉进系统，而是先被称重、编号，再进入对应的执行通道。

八、组合一个最小执行器

import json
import time
import uuid

class ToolRuntime:
    def __init__(
        self,
        registry: ToolRegistry,
        audit_store,
        capabilities: CapabilitySet
    ) -> None:
        self.registry = registry
        self.audit_store = audit_store
        self.capabilities = capabilities

    def execute(self, call: ToolCall) -> ToolResult:
        started = time.perf_counter()
        spec = self.registry.get(call.tool_name)

        if spec is None:
            return ToolResult(
                call_id=call.call_id,
                ok=False,
                code="TOOL_NOT_FOUND",
                message=f"unknown tool: {call.tool_name}"
            )

        errors = validate_arguments(spec.schema, call.arguments)
        if errors:
            return ToolResult(
                call_id=call.call_id,
                ok=False,
                code="INVALID_ARGUMENT",
                message="; ".join(errors)
            )

        try:
            authorize(spec, self.capabilities)

            if call.idempotency_key:
                previous = self.audit_store.find_success(
                    call.idempotency_key
                )
                if previous is not None:
                    return ToolResult(
                        call_id=call.call_id,
                        ok=True,
                        code="IDEMPOTENT_REPLAY",
                        message="reused previous result",
                        data=previous
                    )

            self.audit_store.begin(call, spec)
            data = run_with_timeout(
                spec.handler,
                call.arguments,
                spec.timeout_seconds
            )

            duration = int(
                (time.perf_counter() - started) * 1000
            )
            result = ToolResult(
                call_id=call.call_id,
                ok=True,
                code="OK",
                message="tool completed",
                data=data,
                duration_ms=duration
            )
            self.audit_store.finish(call.call_id, result)
            return result

        except Exception as error:
            code, retryable = map_exception(error)
            duration = int(
                (time.perf_counter() - started) * 1000
            )
            result = ToolResult(
                call_id=call.call_id,
                ok=False,
                code=code,
                message=str(error),
                retryable=retryable,
                duration_ms=duration
            )
            self.audit_store.finish(call.call_id, result)
            return result

这里为了突出边界省略了确认流程、异步任务和结果对象存储。真实实现里还要注意一个问题：begin() 自己失败时，后面的 finish() 不能假设审计记录已经存在。

可以为调用状态定义明确状态机：

RECEIVED
VALIDATED
AUTHORIZED
RUNNING
SUCCEEDED
FAILED
DENIED
TIMED_OUT

状态转换比随意写几列布尔值更容易排查“调用到底停在哪一步”。

九、录制和回放让测试不依赖在线模型

9.1 测试对象是执行层

工具执行器测试不需要每次请求模型。保存一组规范化调用即可：

{
  "name": "reject_parent_path",
  "call": {
    "tool_name": "read_text",
    "arguments": {
      "path": "../secret.txt",
      "max_chars": 100
    }
  },
  "expected": {
    "ok": false,
    "code": "PERMISSION_DENIED"
  }
}

回放器：

def replay_case(runtime: ToolRuntime, case: dict[str, Any]) -> None:
    call = ToolCall(
        call_id=str(uuid.uuid4()),
        tool_name=case["call"]["tool_name"],
        arguments=case["call"]["arguments"],
        actor="replay-test",
        session_id="offline-suite"
    )
    result = runtime.execute(call)

    expected = case["expected"]
    assert result.ok == expected["ok"]
    assert result.code == expected["code"]

9.2 不要对完整文本做脆弱断言

下面的断言容易因措辞调整失效：

assert result.message == "path outside workspace: ../secret.txt"

更稳的是：

assert result.code == "PERMISSION_DENIED"
assert "outside workspace" in result.message

对于文件写入，验证最终状态：

assert target.read_text(encoding="utf-8") == expected_content

对于审计，验证关键字段：

record = audit_store.get(call.call_id)
assert record["decision"] == "allow"
assert record["result_code"] == "OK"
assert record["arguments_json"] != ""

9.3 录制内容也要脱敏

不要把线上调用日志直接复制到测试仓库。

回放样本应：

• 替换用户名和真实路径；
• 删除令牌、Cookie 和请求头；
• 缩小文件内容；
• 固定时间和随机数；
• 用本地假服务替代真实外部 API；
• 保留能触发问题的最小参数。

十、故障注入比只测成功更有价值

建议至少覆盖下面这些用例：

用例	预期结果
未注册工具	TOOL_NOT_FOUND
缺少必填字段	INVALID_ARGUMENT
多余字段	INVALID_ARGUMENT
路径包含 .. 越界	拒绝
符号链接指向工作区外	拒绝
只读会话请求写工具	拒绝
同一幂等键重复提交	不产生第二次副作用
工具内部抛异常	转为稳定错误码
外部服务超时	返回 TIMEOUT
工具返回超大文本	裁剪并标记
审计数据库短暂不可用	不静默丢记录
执行成功但响应写回失败	可通过幂等记录恢复

模拟慢工具：

import time

def slow_tool(seconds: int) -> dict[str, int]:
    time.sleep(seconds)
    return {"slept": seconds}

模拟部分失败时，不要只抛异常：

def write_then_fail(path: str, content: str) -> None:
    Path(path).write_text(content, encoding="utf-8")
    raise RuntimeError("simulated response failure")

这个用例可以检查重试是否再次写入，以及审计里能不能区分“副作用已发生”和“响应失败”。

十一、上线前检查清单

11.1 工具定义

[ ] 每个工具有唯一名称和明确描述
[ ] 参数禁止未声明字段
[ ] 数字、字符串和数组设置合理范围
[ ] 工具有风险等级
[ ] 工具有内部超时
[ ] 返回值结构稳定
[ ] 大结果明确标记截断

11.2 权限

[ ] 默认能力集是最小集合
[ ] 路径在解析符号链接后仍位于工作区
[ ] 写操作限制目标前缀
[ ] 外部发送动作明确目标和数据
[ ] 高风险工具不会仅靠提示词约束

11.3 一致性

[ ] 有副作用工具定义业务幂等键
[ ] 重试次数和退避规则有限
[ ] 部分成功状态能够被识别
[ ] 执行成功但响应失败时可以恢复

11.4 审计与测试

[ ] 参数落库前脱敏
[ ] 结果只保存必要预览或引用
[ ] 每次调用能关联会话和任务
[ ] 拒绝调用也留下策略记录
[ ] 回放测试不依赖在线模型
[ ] 越界、超时、重复和部分失败都有用例

总结

Agent 的工具层如果只负责“根据名字找到函数”，系统很快会被参数漂移、重复副作用和不可追踪失败拖住。

更稳的分工是：

模型负责提出意图
Schema 负责检查参数形状
策略层负责决定能不能做
工具负责完成一个边界明确的动作
幂等层负责控制重复副作用
审计层负责还原过程
回放测试负责固定已知行为

这套结构不会消除模型的不确定性，但能把不确定性关在执行边界之外。工具调用出错时，也不再只剩一句“Agent 又乱调用了”，而是能定位到解析、契约、权限、执行、超时、结果或恢复中的具体环节。

参考资料

1. JSON Schema：https://json-schema.org/
2. JSON Schema Object Reference：https://json-schema.org/understanding-json-schema/reference/object
3. SQLite Write-Ahead Logging：https://www.sqlite.org/wal.html
4. Python subprocess：https://docs.python.org/3/library/subprocess.html
5. Python pathlib：https://docs.python.org/3/library/pathlib.html