用 Claude 构建多工具 Agent:分发、并行与失败边界
在只读模拟工具上理解多工具调度,保留调用 ID、验证参数,并为副作用建立单独审批。
先划分工具与执行责任
客户端工具由你的应用执行,Claude 返回工具名与参数;服务端工具则可能由 Anthropic 执行。本文仅讨论自行定义的客户端工具。工具描述要说明用途、输入含义、返回单位、失败条件及是否有副作用,不能用“万能助手”隐藏权限。
模型选择工具不等于授权。读取公开库存和生成草稿可以与发送邮件、修改数据库分开注册;后两类必须经过身份、对象与权限检查。不要将模型生成的工具名交给 eval,或允许任意 URL、SQL 和 shell 参数直接执行。
原创只读并行分发示例
模拟任务是查询虚构商品的库存和配送说明,两者互不依赖,可同时执行。下面只处理本地静态资料,保留每个调用的 ID,并把未知工具或非法输入返回为错误。已做本地断言检查,没有查询真实库存或调用模型。
import json
from concurrent.futures import ThreadPoolExecutor
def execute(call):
result = {"type": "tool_result", "tool_use_id": call["id"]}
try:
args = call["input"]
if not isinstance(args, dict) or args != {"sku": "DEMO-1"}:
raise ValueError("只接受演示商品 DEMO-1")
if call["name"] == "stock":
value = {"demo": True, "available": 4}
elif call["name"] == "shipping":
value = {"demo": True, "method": "自提"}
else:
raise ValueError("工具不在允许列表")
result["content"] = json.dumps(value, ensure_ascii=False)
except (KeyError, TypeError, ValueError) as exc:
result.update(is_error=True, content=str(exc))
return result
calls = [
{"id": "a", "name": "stock", "input": {"sku": "DEMO-1"}},
{"id": "b", "name": "shipping", "input": {"sku": "DEMO-1"}}
]
with ThreadPoolExecutor(max_workers=2) as pool:
results = list(pool.map(execute, calls))
assert [r["tool_use_id"] for r in results] == ["a", "b"]
assert execute({"id": "c", "name": "delete", "input": {"sku": "DEMO-1"}})["is_error"]把结果正确放回对话
收到模型响应后,保留完整 assistant 内容;下一条 user 消息立即回传该轮每个 tool_use 对应的 tool_result,并保持 ID 对应。不能只返回成功项,失败项也要表达明确错误。示例 results 只是这一步的数据,完整应用还要再次调用模型并检查结束原因。
并行只适合独立操作。先找客户再查该客户订单,属于有依赖的串行链;两个写操作即使参数不同也可能竞争同一资源。限制并发、轮数和总预算,检测相同参数的重复失败,并避免把含敏感内容的异常原文完整回显。
上线前需要补齐什么
加入参数 schema 验证、目标允许列表、网络超时、重试策略、日志脱敏与人工停止按钮。工具返回的网页或用户文字仍是不可信数据,不能提升为系统指令。外发操作应在实际发送前核对收件人、内容和授权。
验收至少包括未知工具、缺参数、部分失败、结果乱序和达到轮数上限。线程示例没有生产级取消和持久化能力;“模型会自动编排”也不能代替这些应用责任。
参考来源
资料核对日期:2026-10-04。涉及产品与账户条件时,请以当前官方说明为准。