工具描述与输入输出 Schema：让模型建议变成可校验的请求

{
  "name": "read_note",
  "description": "读取当前用户的一篇笔记，返回笔记编号和正文。",
  "inputSchema": {
    "type": "object",
    "properties": {
      "note_id": {"type": "string", "minLength": 1}
    },
    "required": ["note_id"],
    "additionalProperties": false
  }
}

{"note_id": "n1", "text": "仅供 u1 读取的示例笔记"}

{
  "type": "object",
  "properties": {
    "note_id": {"type": "string", "minLength": 1},
    "text": {"type": "string"}
  },
  "required": ["note_id", "text"],
  "additionalProperties": false
}

mkdir -Force "F:\学习\agentdepth-l2-tool-schema"
cd "F:\学习\agentdepth-l2-tool-schema"

"""用内存笔记观察参数校验、资源归属和重复写入。"""

from copy import deepcopy


class ToolError(Exception):
    def __init__(self, code, message):
        super().__init__(message)
        self.code = code


class NoteStore:
    def __init__(self):
        # 两位用户各有一篇笔记，owner 保存归属。
        self.notes = {
            "n1": {"owner": "u1", "text": "仅供 u1 读取的示例笔记"},
            "n2": {"owner": "u2", "text": "u2 的示例笔记"},
        }
        # reads 记录成功读取次数，effects 记录实际创建次数。
        self.reads = 0
        self.effects = 0
        self.seen_keys = {}

    def read(self, identity, note_id):
        note = self.notes.get(note_id)
        if note is None or note["owner"] != identity:
            raise ToolError("forbidden", "resource_not_owned")
        self.reads += 1
        return {"note_id": note_id, "text": note["text"]}

    def create_tag(self, identity, note_id, tag, idem_key):
        if identity != "u1" or note_id != "n1":
            raise ToolError("forbidden", "resource_not_owned")
        # 同一个业务键对应同一次创建动作，保存输入和结果用于去重。
        payload = {"identity": identity, "note_id": note_id, "tag": tag}
        if idem_key in self.seen_keys:
            if self.seen_keys[idem_key]["payload"] != payload:
                raise ToolError("idempotency_conflict", "same_key_different_payload")
            return deepcopy(self.seen_keys[idem_key]["result"])
        self.effects += 1
        result = {"tag_id": "tag-%d" % self.effects, "status": "created"}
        self.seen_keys[idem_key] = {"payload": deepcopy(payload), "result": result}
        return deepcopy(result)


def call_tool(request, identity, store):
    """依次检查工具名、输入字段、资源归属及返回值。"""
    if not isinstance(request, dict):
        raise ToolError("invalid_request", "object_required")
    if request.get("name") not in ("read_note", "create_tag"):
        raise ToolError("unknown_tool", "tool_not_allowlisted")
    args = request.get("arguments")
    if not isinstance(args, dict):
        raise ToolError("invalid_arguments", "object_required")
    if request["name"] == "read_note":
        if set(args) != {"note_id"} or not isinstance(args["note_id"], str) or not args["note_id"].strip():
            raise ToolError("invalid_arguments", "read_note_schema")
        # 先核对笔记归属，再执行读取；读取成功后计数增加。
        note = store.notes.get(args["note_id"])
        if note is None or note["owner"] != identity:
            raise ToolError("forbidden", "resource_not_owned")
        result = store.read(identity, args["note_id"])
        if (not isinstance(result, dict) or set(result) != {"note_id", "text"}
                or not isinstance(result["note_id"], str) or not result["note_id"]
                or not isinstance(result["text"], str)):
            raise ToolError("invalid_tool_result", "read_note_output_schema")
        return result
    if set(args) != {"note_id", "tag", "idempotency_key"}:
        raise ToolError("invalid_arguments", "create_tag_schema")
    if not all(isinstance(args[k], str) and args[k].strip() for k in args):
        raise ToolError("invalid_arguments", "non_empty_strings_required")
    return store.create_tag(identity, args["note_id"], args["tag"], args["idempotency_key"])


def demo():
    store = NoteStore()
    cases = [
        ("own_read", {"name": "read_note", "arguments": {"note_id": "n1"}}, "u1"),
        ("cross_user_read", {"name": "read_note", "arguments": {"note_id": "n1"}}, "u2"),
        ("unknown_tool", {"name": "delete_all", "arguments": {}}, "u1"),
    ]
    # 收集每个请求的结果，最后查看实际读取和创建的次数。
    result = []
    for label, request, identity in cases:
        try:
            value = call_tool(request, identity, store)
            result.append({"case": label, "status": "ok", "value": value})
        except ToolError as error:
            result.append({"case": label, "status": "error", "code": error.code})
    return {"cases": result, "body_reads": store.reads, "effects": store.effects}


if __name__ == "__main__":
    print(demo())

python -X utf8 tool_gate.py

python -X utf8 -m unittest -v test_tool_gate.py