Agent工具集必须提交到Schema并做哈希校验

如果模型能调用一个不在代码库里的工具，那你就不是一个 Agent。你只是一个拥有 root 权限的、糊里糊涂的实习生。把工具列表钉死、做哈希、拒绝每一个不匹配的调用。

这就是全文。剩下的部分是一个从零开始的实操演示，在一台笔记本上就能跑。想用远程机器？没问题。代理依然跑在代码旁边。

为什么从这里开始？因为每一个花哨的 Agent 演示都藏着同样的漏洞。模型会凭空发明一个工具、一个路径、一个"有益的"副作用。你的循环耸耸肩就调用了它。我不耸耸肩。你呢？

一个极简的三文件循环：

tools.schema.json — 唯一存在的工具列表。

proxy.py — 对该文件做哈希、校验每次调用、然后分发。

test_proxy.py — 四个检查项，在任何模型看到 prompt 之前必须通过。

模型的职责很窄。它只输出一个 JSON 对象。代理决定这个对象是否真实存在。如果 schema 文件变了，哈希就变了，昨天的调用记录就不再可信。这就是重点。

我不会去接一套完整的 MCP 协议。也不会去解释二十个 Agent 术语。我只是在钉死一份你可以 grep 的契约。

Stage 0：建一个用完即弃的临时目录

在一个空文件夹里做这个。别在你主项目仓库里做，也别放在 .env 旁边。

mkdir agent-tool-pin && cd agent-tool-pin
python3 -m venv .venv
. .venv/bin/activate

验证：which python 指向 .venv。如果没有，停下来。你接下来不会安装任何东西，但习惯还是要养成。

Stage 1：把 schema 写成数据，而不是 prompt 段落

Prompt 会腐坏。文件会被 review。把工具写在 JSON 里。

{
  "schema_version": 1,
  "tools": {
    "list_dir": {
      "args": {"path": "string"},
      "side_effect": "read",
      "roots": ["workspace"]
    },
    "read_file": {
      "args": {"path": "string", "max_bytes": "int"},
      "side_effect": "read",
      "roots": ["workspace"]
    },
    "write_file": {
      "args": {"path": "string", "content": "string"},
      "side_effect": "write",
      "roots": ["workspace"]
    }
  }
}

保存为 tools.schema.json。三个工具。已经很大方了。大多数任务两个就够。

python3 -c "import json; json.load(open('tools.schema.json')); print('schema ok')"

如果输出的不是 schema ok，你就根本没有 schema。你只有一个拼写错误。

Stage 2：对你要发布的文件做哈希

不要对内存里重建的 Python 字典做哈希。要对磁盘上的原始字节做哈希。CI 看到的是同一个文件。代理加载的也是同一个文件。

# hash_schema.py
from pathlib import Path
import hashlib

p = Path("tools.schema.json")
digest = hashlib.sha256(p.read_bytes()).hexdigest()
print(digest)
Path("tools.schema.sha256").write_text(digest + "\n")

python3 hash_schema.py
cat tools.schema.sha256

验证：文件包含 64 个十六进制字符和一个换行符。把两个文件一起提交。如果有人"顺手加了一个工具"却忘了更新哈希，代理必须拒绝启动。这份拒绝就是它的价值。

Stage 3：那个Fail Closed的代理

这才是我真正想放在循环里的分发器。它不跟模型通信、不猜测。它加载 schema、校验哈希、校验名称、校验参数、校验路径根目录，然后才执行一个函数。

# proxy.py
from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parent
WORKSPACE = ROOT / "workspace"
SCHEMA_PATH = ROOT / "tools.schema.json"
HASH_PATH = ROOT / "tools.schema.sha256"

class ToolRefused(Exception):
    pass

def load_schema() -> dict[str, Any]:
    expected = HASH_PATH.read_text().strip()
    actual = hashlib.sha256(SCHEMA_PATH.read_bytes()).hexdigest()
    if actual != expected:
        raise ToolRefused(f"schema hash mismatch: {actual} != {expected}")
    return json.loads(SCHEMA_PATH.read_text())

def _safe_path(raw: str) -> Path:
    candidate = (WORKSPACE / raw).resolve()
    workspace = WORKSPACE.resolve()
    if workspace not in candidate.parents and candidate != workspace:
        raise ToolRefused(f"path escapes workspace: {raw}")
    return candidate

def list_dir(path: str) -> dict[str, Any]:
    target = _safe_path(path)
    if not target.exists():
        raise ToolRefused(f"missing: {path}")
    names = sorted(p.name for p in target.iterdir())
    return {"path": path, "entries": names}

def read_file(path: str, max_bytes: int) -> dict[str, Any]:
    if not isinstance(max_bytes, int) or max_bytes < 1 or max_bytes > 65536:
        raise ToolRefused("max_bytes out of range")
    target = _safe_path(path)
    data = target.read_bytes()[:max_bytes]
    return {"path": path, "bytes": len(data), "text": data.decode("utf-8", "replace")}

def write_file(path: str, content: str) -> dict[str, Any]:
    if not isinstance(content, str):
        raise ToolRefused("content must be a string")
    if len(content.encode("utf-8")) > 65536:
        raise ToolRefused("write too large")
    target = _safe_path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(content)
    return {"path": path, "wrote": True}

DISPATCH = {
    "list_dir": list_dir,
    "read_file": read_file,
    "write_file": write_file,
}

def run_tool_call(call: dict[str, Any]) -> dict[str, Any]:
    schema = load_schema()
    if set(call.keys()) != {"tool", "args"}:
        raise ToolRefused("call must be {tool, args} only")
    name = call["tool"]
    args = call["args"]
    if name not in schema["tools"]:
        raise ToolRefused(f"unknown tool: {name}")
    spec = schema["tools"][name]
    expected_args = spec["args"]
    if set(args.keys()) != set(expected_args.keys()):
        raise ToolRefused(f"args mismatch for {name}")
    # Type tags are labels, not a full validator. Still cheaper than hope.
    for key, kind in expected_args.items():
        value = args[key]
        if kind == "string" and not isinstance(value, str):
            raise ToolRefused(f"{key} must be string")
        if kind == "int" and not isinstance(value, int):
            raise ToolRefused(f"{key} must be int")
    fn = DISPATCH[name]
    return {"ok": True, "tool": name, "result": fn(**args)}

看 DISPATCH。如果一个名称出现在 schema 里却不在这里，run_tool_call 会在查找时爆炸。好。不要从 schema 自动生成这张表。两份必须保持一致的定义列表，比一份会说谎的列表要好。

mkdir -p workspace
echo 'hello' > workspace/note.txt
python3 - <<'PY'
from proxy import run_tool_call
print(run_tool_call({"tool": "list_dir", "args": {"path": "."}}))
PY

你应该在 entries 里看到 note.txt。如果没有，workspace 路径有问题。在邀请模型之前先把这个修好。

Stage 4：先证明拒绝机制有效，然后也许再调用模型

测试先行。始终如此。模型是你这场派对最后邀请的客人。你为什么要反过来？

# test_proxy.py
import json
from pathlib import Path

import pytest
from proxy import ToolRefused, run_tool_call

def test_happy_read():
    Path("workspace/note.txt").write_text("hello\n")
    out = run_tool_call({"tool": "read_file", "args": {"path": "note.txt", "max_bytes": 32}})
    assert out["ok"] is True
    assert "hello" in out["result"]["text"]

def test_unknown_tool_is_dead():
    with pytest.raises(ToolRefused, match="unknown tool"):
        run_tool_call({"tool": "run_shell", "args": {"cmd": "id"}})

def test_extra_arg_is_dead():
    with pytest.raises(ToolRefused, match="args mismatch"):
        run_tool_call({"tool": "list_dir", "args": {"path": ".", "follow_symlinks": True}})

def test_path_escape_is_dead():
    with pytest.raises(ToolRefused, match="escapes workspace"):
        run_tool_call({"tool": "read_file", "args": {"path": "../proxy.py", "max_bytes": 16}})

def test_hash_mismatch_is_dead(tmp_path, monkeypatch):
    # This test is a reminder: if you edit the schema, update the hash in the same commit.
    schema = Path("tools.schema.json")
    original = schema.read_text()
    try:
        data = json.loads(original)
        data["tools"]["run_shell"] = {"args": {"cmd": "string"}, "side_effect": "exec", "roots": []}
        schema.write_text(json.dumps(data))
        with pytest.raises(ToolRefused, match="schema hash mismatch"):
            run_tool_call({"tool": "list_dir", "args": {"path": "."}})
    finally:
        schema.write_text(original)

在虚拟环境里安装 pytest 并运行：

pip install pytest
pytest -q test_proxy.py

验证：四个失败意味着你不发版。一个 skip 意味着你无聊了。零 skip、全绿，意味着你有了一道门。

run_shell 那个 case 就是全文的核心论点。模型会请求它。你的代理不会长出嘴巴。

Stage 5：给模型一种形状，而不是一个人格

现在轮到提议者了。保持无聊。模型返回 JSON。你解析它。你不用正则和祈祷从一段文字里"提取"它。

# propose.py
import json
from pathlib import Path

SCHEMA = Path("tools.schema.json").read_text()

SYSTEM = """You propose exactly one tool call.
Return a JSON object with keys tool and args.
No markdown. No commentary.
You may only use tools listed in the schema.
"""

def build_user_prompt(task: str) -> str:
    return (
        "schema:\n"
        + SCHEMA
        + "\n\ntask:\n"
        + task
        + "\n"
    )

def parse_call(raw: str) -> dict:
    data = json.loads(raw)
    if not isinstance(data, dict):
        raise ValueError("not an object")
    return data

用你习惯的方式接模型。一个本地 stub 就足够完成这个教程了：

# stub_model.py
def complete(_system: str, user: str) -> str:
    # Deterministic stand-in so the loop is testable offline.
    if "list the workspace" in user:
        return '{"tool": "list_dir", "args": {"path": "."}}'
    return '{"tool": "run_shell", "args": {"cmd": "echo pwned"}}'

# loop.py
from propose import SYSTEM, build_user_prompt, parse_call
from proxy import ToolRefused, run_tool_call
from stub_model import complete

def one_step(task: str) -> dict:
    raw = complete(SYSTEM, build_user_prompt(task))
    call = parse_call(raw)
    try:
        return run_tool_call(call)
    except ToolRefused as exc:
        return {"ok": False, "error": str(exc), "raw": call}

if __name__ == "__main__":
    print(one_step("list the workspace"))
    print(one_step("open a shell and look around"))

python3 loop.py

验证：第一次打印有 "ok": true。第二次打印有 "ok": false 和 unknown tool。如果第二个跑了一个 shell，你不是在按这篇文章做。你是在按一个 demo 做。

Stage 6：什么时候提议者不应该坐在你笔记本上

代理在本地是有原因的。模型调用不一定非要在本地。如果你想让提议者离开你的机器，把那一步停在一个免费的远程盒子上，把 proxy.py 留在你的文件旁边。

披露：这篇文章是 MonkeyCode 产品推广的一部分。我用 MonkeyCode 的免费模型访问和免费服务器选项来做提议者那一跳，当我不想在笔记本上再多一个 API key 的时候。schema 文件、哈希值和分发器依然留在仓库里。远程那端只看到 schema 文本和任务字符串。它永远拿不到 shell。

把 tools.schema.json 拷贝到远程的 prompt 构建器里。别拷贝 workspace/。别拷贝 secrets。拿回来一个 JSON 对象。在本地跑 run_tool_call。一样的测试。一样的拒绝。

一个草图，标注为草图，不是厂商 SDK：

[laptop] pytest test_proxy.py          # must be green
[laptop] send schema + task  ----->   [remote proposer]
[laptop] recv JSON object    <-----   [remote proposer]
[laptop] run_tool_call(call)          # fail closed

如果远程盒子明天没了，代理依然能用 stub_model.py 工作。这就是设计。免费服务器是便利，不是你的信任边界。

这套方案不做什么

它不会阻止模型在 workspace 里写垃圾。它不替代 code review。它不校验语义，只校验名称、类型和路径根目录。write_file 依然可以覆盖 workspace/note.txt。那是允许的。那就是为什么 workspace 是一个临时目录。

它也不会造出一个"Agent"。三个工具和一个 JSON blob 是一个带额外延迟的函数调用。如果你需要多步骤计划，加一个显式的步骤计数器加一个人工检查点。别加一个能凭空制造工具的隐藏规划器。

如果你在驱动一个生产发布机器人、任何带云凭证的东西、或者一个非一次性的支付流程，跳过这套方案。如果你的 schema 包含了 exec、http_request 或 sql，跳过它。那些不是用一个周末代理就能钉死的东西。那些是需要审计的产品。

在我信任一个变更之前运行的检查

Schema 和哈希在同一 commit 里。

pytest -q test_proxy.py 全绿。

虚构的工具依然被拒绝。

路径 ../ 依然被拒绝。

模型输出解析为 JSON，从不用 markdown。

远程提议者（如果有的话）永不挂载 workspace。

如果第二步是红的，我不会"就试试模型算了"。这句话就是你拿到 shell 的方式。你早就知道了。哈希文件在那里，就是为了让你没法假装自己忘了。