#!/usr/bin/env python3 """tool_loop.py — 工具使用完整回路模拟（无需 API key，本地可跑） 用「脚本化 FakeClaude」+ 真实的消息结构，完整走一遍官方回路： User -> Claude(tool_selection) -> tool_use -> 应用执行(权限门/错误处理) -> tool_result -> Claude -> 最终回答 演示三条关键路径（模块 04 README）： A. 正常路径：tool_use -> 执行 -> tool_result -> 最终回答 B. 错误回传：工具 500 -> is_error=true 带指导性消息 -> Claude 改道(换工具) C. 权限门：不可逆动作 -> 人审拒绝 -> is_error 回传 -> Claude 告知用户 消息结构严格对照官方文档（访问于 2026-08-30）: https://platform.claude.com/docs/en/agents-and-tools/tool-use/handle-tool-calls https://platform.claude.com/docs/en/agents-and-tools/tool-use/overview 运行: python3 tool_loop.py """ from __future__ import annotations import json TOOLS = [ { "name": "get_weather", "description": "Get the current weather for a given location. 用于用户询问某地实时天气。", "input_schema": { "type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"], }, }, { "name": "get_weather_forecast", "description": "Get the weather forecast. 当实时天气服务不可用时,可用预报数据兜底。", "input_schema": { "type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"], }, }, { "name": "delete_order_record", "description": "Permanently delete an order record. 不可逆操作。", "input_schema": { "type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"], }, }, ] # 应用侧权限策略:哪些工具需要人工审批(不可逆/高代价) HUMAN_APPROVAL_REQUIRED = {"delete_order_record"} READ_ONLY = {"get_weather", "get_weather_forecast"} def assistant_turn(content_blocks: list[dict], stop_reason: str) -> dict: """构造 assistant 消息(回传历史时原样携带 tool_use)。""" return {"role": "assistant", "content": content_blocks, "stop_reason": stop_reason} def tool_result_message(results: list[dict], extra_text: str | None = None) -> dict: """构造回传消息。官方硬约束:tool_result 必须在最前,text 只能跟在后面。""" blocks = [{"type": "tool_result", **r} for r in results] if extra_text: blocks.append({"type": "text", "text": extra_text}) return {"role": "user", "content": blocks} def permission_gate(tool_name: str) -> tuple[bool, str]: """执行前门(应用层强制):LLM 的指令不是授权机制。""" if tool_name in READ_ONLY: return True, "" if tool_name in HUMAN_APPROVAL_REQUIRED: # 演示:人审队列返回"拒绝" return False, "人工审批被拒绝:删除订单记录属于不可逆操作,需要部门主管批准(本演示自动拒绝)。" return False, f"权限被拒:工具 {tool_name} 不在白名单内。" def execute_tool(name: str, tool_input: dict, fail: bool = False) -> dict: """确定性执行器。fail=True 模拟上游服务 500(路径 B)。""" if name == "get_weather" and not fail: return {"tool_use_id": tool_input["_id"], "content": "15 degrees Celsius, partly cloudy", "is_error": False} if name == "get_weather" and fail: return { "tool_use_id": tool_input["_id"], "content": "ConnectionError: the weather service API is not available (HTTP 500). " "建议:get_weather_forecast 不依赖该服务,可改用它。", "is_error": True, # 官方建议:错误消息要有指导性,而非只回 "failed" } if name == "get_weather_forecast": return {"tool_use_id": tool_input["_id"], "content": "Forecast: 14-17 degrees, cloudy with sunny spells (data source: forecast-db)", "is_error": False} raise AssertionError(f"未实现: {name}") def fake_claude(messages: list[dict]) -> dict: """脚本化模型:按剧情返回 tool_use / 最终回答。真实系统这里是 POST /v1/messages。""" last = messages[-1] if last["role"] == "user" and isinstance(last["content"], str): if "删" in last["content"]: blocks = [{"type": "text", "text": "我来删除这条订单记录。"}, {"type": "tool_use", "id": "toolu_del_1", "name": "delete_order_record", "input": {"order_id": "A-1024"}}] return assistant_turn(blocks, "tool_use") blocks = [{"type": "text", "text": "I'll check the current weather."}, {"type": "tool_use", "id": "toolu_wx_1", "name": "get_weather", "input": {"location": "San Francisco, CA"}}] return assistant_turn(blocks, "tool_use") # 收到 tool_result 之后的回合 results = [b for b in last["content"] if b["type"] == "tool_result"] for r in results: if r.get("is_error"): if "HTTP 500" in r["content"]: # 自我修正:换兜底工具(官方:Claude 会带着错误上下文改道) blocks = [{"type": "tool_use", "id": "toolu_wx_2", "name": "get_weather_forecast", "input": {"location": "San Francisco, CA"}}] return assistant_turn(blocks, "tool_use") # 权限拒绝 -> 如实告知用户 return assistant_turn( [{"type": "text", "text": "抱歉,该操作需要人工审批且已被拒绝,我无法删除这条订单记录。" "您可以联系部门主管走线下审批流程。"}], "end_turn") if "Forecast" in r["content"]: return assistant_turn( [{"type": "text", "text": "实时天气服务暂时不可用。根据预报数据:旧金山 14-17 度,多云间晴。"}], "end_turn") return assistant_turn( [{"type": "text", "text": "旧金山当前 15 摄氏度,局部多云。"}], "end_turn") raise AssertionError("剧情走飞了") def run_agent(user_question: str, tool_fails: bool = False) -> None: print(f"\n===== 用户: {user_question} =====") tools_by_id = {} messages: list[dict] = [{"role": "user", "content": user_question}] for step in range(4): # 防御性最大步数(agent loop 停止条件的最简形态) resp = fake_claude(messages) messages.append({"role": "assistant", "content": resp["content"]}) print(f"[Claude] stop_reason={resp['stop_reason']}") if resp["stop_reason"] != "tool_use": text = next(b["text"] for b in resp["content"] if b["type"] == "text") print(f"[Claude] 最终回答: {text}") return # 应用层:权限门 -> 执行 -> 组装 tool_result results = [] for block in resp["content"]: if block["type"] != "tool_use": continue print(f"[应用] 收到 tool_use: {block['name']}({json.dumps(block['input'], ensure_ascii=False)})") allowed, reason = permission_gate(block["name"]) if not allowed: print(f"[应用] 权限门拒绝: {reason[:30]}...") results.append({"tool_use_id": block["id"], "content": reason, "is_error": True}) continue fail = tool_fails and block["name"] == "get_weather" tool_input = dict(block["input"], _id=block["id"]) result = execute_tool(block["name"], tool_input, fail=fail) print(f"[应用] 执行完成 is_error={result['is_error']}: {result['content'][:40]}...") results.append(result) tools_by_id[block["id"]] = block["name"] messages.append(tool_result_message(results)) # tool_result 在前,无额外 text print("[应用] 达到最大步数,强制停止(生产:配预算与停止条件)") def main() -> None: run_agent("旧金山现在天气怎么样?", tool_fails=False) # 路径 A:一步成功 run_agent("旧金山现在天气怎么样?", tool_fails=True) # 路径 B:错误回传 -> Claude 改道 run_agent("帮我把订单 A-1024 删了", tool_fails=False) # 路径 C:人审门拒绝 # 自检:消息结构符合官方硬约束 msg = tool_result_message( [{"tool_use_id": "t1", "content": "ok", "is_error": False}], extra_text="还有什么吗?") assert msg["content"][0]["type"] == "tool_result" assert msg["content"][1]["type"] == "text" # text 只能在 tool_result 之后 assert HUMAN_APPROVAL_REQUIRED == {"delete_order_record"} print("\n自检通过:tool_result 优先排序与权限门策略符合模块 04 规则") if __name__ == "__main__": main()