"""A7 · Agent Tools v1.1 demo. 离线工具边界实验：注册表、授权、审批、参数校验、调用去重/冲突、预算。 标准库 only；不是真实 LLM / MCP / 远程 API。 运行：python3 A7-Agent-Tools-v1.1-demo.py """ from __future__ import annotations import json from dataclasses import dataclass, field from typing import Any, Callable, Dict, FrozenSet, List, Optional class ToolError(Exception): def __init__(self, code: str, message: str = ""): self.code = code super().__init__(message or code) @dataclass(frozen=True) class ToolSpec: name: str allowed_fields: FrozenSet[str] required_fields: FrozenSet[str] effect_class: str # read_only | side_effect requires_approval: bool = False handler: Optional[Callable[[Dict[str, Any], "Context"], Dict[str, Any]]] = None @dataclass(frozen=True) class ToolCall: call_id: str name: str arguments_json: str @dataclass class Context: grants: FrozenSet[str] = frozenset() approvals: FrozenSet[str] = frozenset() # approved call_ids budget: int = 5 spent: int = 0 published: List[str] = field(default_factory=list) audit: List[Dict[str, Any]] = field(default_factory=list) class Registry: def __init__(self): self.tools: Dict[str, ToolSpec] = {} self.ledger: Dict[str, Dict[str, Any]] = {} # call_id -> result snapshot def register(self, spec: ToolSpec) -> None: self.tools[spec.name] = spec def execute(self, call: ToolCall, ctx: Context) -> Dict[str, Any]: # 1) call-id consistency / dedup if call.call_id in self.ledger: prev = self.ledger[call.call_id] if prev["name"] != call.name or prev["arguments_json"] != call.arguments_json: raise ToolError("CALL_ID_CONFLICT", "same call_id with different payload") return prev["result"] # 2) budget if ctx.spent >= ctx.budget: raise ToolError("BUDGET_EXCEEDED", "call budget exhausted") # 3) registration spec = self.tools.get(call.name) if spec is None: raise ToolError("UNKNOWN_TOOL", call.name) # 4) authorization if call.name not in ctx.grants: raise ToolError("PERMISSION_DENIED", call.name) # 5) approval for high-risk if spec.requires_approval and call.call_id not in ctx.approvals: raise ToolError("APPROVAL_REQUIRED", call.call_id) # 6) JSON parse try: args = json.loads(call.arguments_json) except json.JSONDecodeError as exc: raise ToolError("INVALID_JSON", str(exc)) from exc if not isinstance(args, dict): raise ToolError("INVALID_ARGUMENTS", "arguments must be an object") # 7) field schema (whitelist + required + string types) unknown = set(args) - set(spec.allowed_fields) if unknown: raise ToolError("UNKNOWN_FIELD", ",".join(sorted(unknown))) missing = set(spec.required_fields) - set(args) if missing: raise ToolError("MISSING_FIELD", ",".join(sorted(missing))) for key, value in args.items(): if not isinstance(value, str): raise ToolError("INVALID_FIELD_TYPE", key) assert spec.handler is not None result = spec.handler(args, ctx) ctx.spent += 1 event = { "call_id": call.call_id, "name": call.name, "status": "SUCCEEDED" if result.get("ok") else "FAILED", } ctx.audit.append(event) self.ledger[call.call_id] = { "name": call.name, "arguments_json": call.arguments_json, "result": result, } return result def make_registry() -> Registry: reg = Registry() def lookup(args: Dict[str, Any], ctx: Context) -> Dict[str, Any]: catalog = { "react": "ReAct interleaves reasoning and acting with observations.", "toolformer": "Toolformer learns when and how to call APIs.", } q = args["query"].strip().lower() hit = next((v for k, v in catalog.items() if k in q), None) if hit is None: return {"ok": True, "data": {"items": []}} return {"ok": True, "data": {"items": [{"title": q, "snippet": hit}]}} def publish(args: Dict[str, Any], ctx: Context) -> Dict[str, Any]: # 模拟外部副作用：写入内存发布列表 ctx.published.append(args["document_id"]) return {"ok": True, "effect_ref": f"pub:{args['document_id']}"} reg.register( ToolSpec( name="lookup", allowed_fields=frozenset({"query"}), required_fields=frozenset({"query"}), effect_class="read_only", requires_approval=False, handler=lookup, ) ) reg.register( ToolSpec( name="publish", allowed_fields=frozenset({"document_id"}), required_fields=frozenset({"document_id"}), effect_class="side_effect", requires_approval=True, handler=publish, ) ) return reg if __name__ == "__main__": reg = make_registry() ctx = Context(grants=frozenset({"lookup", "publish"}), budget=5) # 1) 正常查询 result = reg.execute(ToolCall("c1", "lookup", '{"query":"react"}'), ctx) assert result["ok"] assert result["data"]["items"] # 2) 相同 call_id 去重（不增加 spent） spent_after_first = ctx.spent again = reg.execute(ToolCall("c1", "lookup", '{"query":"react"}'), ctx) assert again == result assert ctx.spent == spent_after_first # 3) call_id 冲突 try: reg.execute(ToolCall("c1", "lookup", '{"query":"toolformer"}'), ctx) raise AssertionError("expected CALL_ID_CONFLICT") except ToolError as err: assert err.code == "CALL_ID_CONFLICT" # 4) 未审批拒绝（无副作用） try: reg.execute(ToolCall("p1", "publish", '{"document_id":"draft-9"}'), ctx) raise AssertionError("expected APPROVAL_REQUIRED") except ToolError as err: assert err.code == "APPROVAL_REQUIRED" assert ctx.published == [] # 5) 非法 JSON try: reg.execute(ToolCall("c2", "lookup", "{not-json"), ctx) raise AssertionError("expected INVALID_JSON") except ToolError as err: assert err.code == "INVALID_JSON" # 6) 未知字段 try: reg.execute(ToolCall("c3", "lookup", '{"query":"react","extra":"x"}'), ctx) raise AssertionError("expected UNKNOWN_FIELD") except ToolError as err: assert err.code == "UNKNOWN_FIELD" # 7) 批准后发布一次 + 预算阻断 ctx.approvals = frozenset({"p1"}) published = reg.execute(ToolCall("p1", "publish", '{"document_id":"draft-9"}'), ctx) assert published["ok"] assert ctx.published == ["draft-9"] # 去重不会二次发布 reg.execute(ToolCall("p1", "publish", '{"document_id":"draft-9"}'), ctx) assert ctx.published == ["draft-9"] # 耗尽预算：已 spent 含 c1 + p1；再补足到预算上限 while ctx.spent < ctx.budget: cid = f"fill-{ctx.spent}" reg.execute(ToolCall(cid, "lookup", '{"query":"react"}'), ctx) try: reg.execute(ToolCall("over", "lookup", '{"query":"react"}'), ctx) raise AssertionError("expected BUDGET_EXCEEDED") except ToolError as err: assert err.code == "BUDGET_EXCEEDED" print("A7 tool boundary: 7 checks passed")