工具运行时
工具注册、参数校验、并行执行、超时、结果截断与错误回传。
一句话理解:工具运行时是模型和真实世界之间的“执行车间”:它把模型发出的调用请求,依次经过查找、校验、把关、限时执行、结果整形,变成一条模型读得懂的结果;任何一道工序出问题,都要变成一条结果交还给模型,而不是让整个循环崩掉。
模型说“我要运行 npm test,再读一下 auth.ts”,这只是一段结构化文本。真正去跑命令、读文件、等结果、处理报错的,是 Harness 里的工具运行时。Demo 阶段它常常只有十几行:找到函数、调用、把返回值塞回去。一上真实任务,问题就来了:某个测试挂起不退出,整个 Agent 跟着卡死;一次日志查询返回几十万字,上下文直接爆掉;模型编了一个不存在的参数,程序抛异常退出,前面十步的工作全部白费。
工具模块已经讲过怎么设计一个好工具(工具设计原则)以及哪些调用能并行(并行调用与调用链)。这一课换一个视角:站在 Harness 代码里,一次工具调用从进来到出去要经过哪些工序,每道工序的默认值该怎么定。
一次调用的七道工序
把一次工具调用拆开,是一条固定的流水线:
其中第 04 道“权限判定”是下一课的主题,这里只需要知道它插在“校验通过”和“真正执行”之间。其余六道,逐个看。
注册表:工具不只是一个函数
最简单的 Harness 用一个字典把工具名映射到函数。生产级的注册表要为每个工具多记几项“运行时属性”,后面每道工序都靠它们做决定:
| 属性 | 作用 | Coding Agent 里的例子 |
|---|---|---|
| 名称、描述、参数 Schema | 发给模型,也用于校验 | run_tests(path) |
| 是否只读 | 决定能否与其他调用并发 | 读文件、搜索:是;改文件、跑命令:否 |
| 超时 | 限制单次调用的最长时间 | 测试 5 分钟,读文件 10 秒 |
| 输出上限 | 超过就触发截断策略 | 测试输出 2 万字符 |
| 风险等级 | 交给权限判定使用 | 推送代码:高 |
注册表还决定“这一轮把哪些工具告诉模型”。工具越多,定义越占上下文,所以成熟的 Harness 会按需加载:Claude Agent SDK 默认延迟加载 MCP 工具的 Schema,需要时再通过工具搜索取回(装配策略见 上下文装配,MCP 等外部工具如何挂进来见 扩展点)。
“是否只读”这一项尤其值得默认保守。Claude Agent SDK 的做法是:内置的只读工具可以并发,改状态的工具按顺序执行;自定义工具默认串行,只有在注解里标了 readOnlyHint 才会并发。忘了标,最多慢一点;标错了,两个写操作就可能互相覆盖。
解析与校验:别直接相信模型给的参数
不同 API 交来的调用形态不同。Claude 的 tool_use 块里 input 已经是对象;OpenAI Responses API 的 function_call 里 arguments 是一段 JSON 字符串,需要自己解析,解析本身就可能失败。
校验分两层:
- Schema 校验:类型、必填、枚举。两家都提供严格模式(
strict: true),能保证参数符合 Schema,从源头消灭“缺参数、类型错”。 - 业务校验:Schema 表达不了的规则,例如“这个文件路径在项目目录内”“这张工单属于当前客户”。严格模式管不到这一层,必须在运行时代码里检查。
校验失败时,不要抛异常,而是回一条带 is_error: true 的结果,写清哪里不对、怎么改。Claude 的文档提到,参数缺失时把错误交还给模型,它通常会补全信息重试两三次。
调度:一轮里的多个调用怎么排
模型一轮可能发出好几个调用。运行时要决定执行方式,并保证结果能一一对上:
图里有三条运行时必须守住的规则:
- 按“只读 / 会改状态”分批:连续的只读调用放进同一批并发,遇到写操作或命令就先等前一批结束,再单独执行。
- 结果按原顺序、按 id 配对:Claude 要求每个
tool_use都有对应的tool_result,全部放在紧接着的同一条用户消息里,并排在任何文字之前;OpenAI 用call_id对应function_call_output。 - 没执行的也要有结果:如果前一个调用失败、用户中途打断,后面没跑的调用也要回一条错误结果,说明“未执行及原因”,否则对话结构不完整,下一次请求会直接报错。
需要彻底关闭并行时,Claude 用 disable_parallel_tool_use,OpenAI 用 parallel_tool_calls: false,细节见 并行调用与调用链。
超时:“不等了”和“停下来”是两回事
每个工具都要有超时,而且超时要分两种想:
- 总时长上限:无论如何,超过这个时间就结束。
- 空闲超时:长时间没有任何输出或进度,就判定卡死。
Claude Code 是一个现成的参照:Bash 命令默认 2 分钟超时,模型最多能申请到 10 分钟,超时的命令会被转到后台继续跑,而不是直接杀掉;MCP 工具既可以按 Server 配置硬性的总时长,也有空闲超时,一直没有响应也没有进度通知就报错。
落到代码里,有两个容易漏的点。第一,超时后要 真正终止或转移 子进程,只是“不再等待”会留下一个继续改文件的孤儿进程。第二,写操作超时不等于没执行,返回给模型的信息要写明“结果未知,先查询是否已生效”,做到一半的恢复方法见 错误恢复与重试。
结果整形:截断要留下线索
工具返回的原始数据,还要经过序列化、脱敏、截断三步才能进上下文:
这张图其实就是 Claude Code 的做法(据其工具参考文档):Bash 输出正常时,大约 3 万字符以内直接返回,超过则存成会话目录里的文件,只回文件路径和开头约 2000 字符的预览,模型需要时再去读或搜索;命令失败时,只保留约 1 万字符的头尾摘录。MCP 工具输出超过 1 万 Token 会给出警告,默认上限 2.5 万 Token,超出的结果同样存成文件、在对话里换成文件路径。
还有两件整形的事别忘了:脱敏,把结果里的密钥、手机号在进入上下文之前抹掉;标记来源不可信,网页、邮件、用户上传的内容可能夹带指令,应放在工具结果里而不是系统提示里,防护思路见 Prompt 注入。
把七道工序写成代码
下面是一段伪代码(不对应任何特定 SDK),把上面的工序串起来。可以对照 实操:从零写一个最小 Agent 里的 run_tool,看看多出来的部分都在防什么:
# 伪代码:一次工具调用在 Harness 里的完整路径
import asyncio
async def run_one(call, registry, policy, session):
"""输入一个调用,输出一条 tool_result;任何失败都变成结果,而不是异常。"""
def result(text, is_error=False):
return {"type": "tool_result", "tool_use_id": call.id,
"content": text, "is_error": is_error}
tool = registry.get(call.name) # 02 查注册表
if tool is None:
return result(f"未知工具 {call.name},可用:{registry.names()}", True)
try: # 01 解析 + 03 校验
args = parse_args(call) # OpenAI 的 arguments 是 JSON 字符串,需自行解析
tool.schema.validate(args) # 类型、必填、枚举
tool.check_business(args, session) # 例如:路径必须在项目目录内
except ValidationError as e:
return result(f"参数不合法:{e}。请按工具说明修正后重试", True)
decision = await policy.decide(tool, args, session) # 04 权限判定(下一课)
if decision.denied:
return result(f"操作被拒绝:{decision.reason}", True)
try: # 05 执行,带超时
raw = await asyncio.wait_for(tool.handler(**args), timeout=tool.timeout_s)
except asyncio.TimeoutError:
await tool.cleanup(call.id) # 真正终止子进程,而不只是不再等待
hint = "写操作结果未知,请先查询是否已生效" if tool.writes else "可缩小范围后重试"
return result(f"{call.name} 超过 {tool.timeout_s} 秒未完成。{hint}", True)
except Exception as e:
return result(f"{type(e).__name__}: {e}", True)
text = redact(to_text(raw)) # 06 整形:脱敏 + 截断
if len(text) > tool.max_chars:
path = session.save_artifact(call.id, text)
text = (f"[输出共 {len(text)} 字符,已截断;完整内容在 {path},可分段读取]\n"
+ text[: tool.preview_chars])
return result(text)
async def run_turn(calls, registry, policy, session):
"""07 回传:连续的只读调用并发,其余逐个执行,结果保持原顺序。"""
results, batch = [], []
async def flush():
results.extend(await asyncio.gather(
*(run_one(c, registry, policy, session) for c in batch)))
batch.clear()
for c in calls:
tool = registry.get(c.name)
if tool and tool.read_only:
batch.append(c)
else:
await flush()
results.append(await run_one(c, registry, policy, session))
await flush()
return results # 全部放进同一条 user 消息,排在任何文字之前
注意 run_one 里只有一个出口类型:无论成功还是失败,交出去的都是一条 tool_result。这就是图 1 下方那条虚线在代码里的样子。
常见误区
- “开了严格模式就不用校验了。” 严格模式只保证参数符合 Schema,“路径在不在项目里”“工单归不归当前客户”这类业务规则仍要运行时检查。
- “超时就是
wait_for一下。” 不再等待不等于停止执行。子进程、远程任务要真正终止或转入后台管理,写操作还要确认是否已经生效。 - “截断就是切前 N 个字。” 不告诉模型被截断,它会把片段当全部;失败输出只留开头,最关键的报错行反而被切掉。
设计要点
- 注册表里每个工具是否都写明了:是否只读、超时、输出上限、风险等级?自定义工具是否默认串行?
- 参数校验是否同时覆盖 Schema 和业务规则?校验失败的信息,模型读了能改对吗?
- 每个工具的超时是多少?超时后进程怎么处理?写操作超时后模型看到什么?
- 大结果走哪种截断?截断提示是否写明原长度和取剩余内容的方法?
- 一轮多个调用时,结果是否按 id 全部配对回传,包括没执行的那些?
- 每次调用的耗时、状态、是否截断,是否都写进了 Tracing?
练习
以「客服工单 Agent」为对象(约 45 分钟)。它有 5 个工具:查询工单、搜索知识库、查询订单、回复客户、关闭工单。
- 为每个工具填一张注册表:是否只读、超时、输出上限、截断方式、失败时返回给模型的一句话。
- 假设模型一轮同时请求“查询工单、搜索知识库、回复客户”,画出调度顺序,并写出三条结果回传的顺序。
- 如果你写过 最小 Agent,给它的
run_tool加上超时和截断,故意让一个工具睡眠超时、另一个返回超长文本,观察模型怎么应对。
延伸阅读
- Handle tool calls(Claude):
tool_result的格式要求、is_error与错误信息写法。 - How the agent loop works(Claude Agent SDK):只读工具并发、自定义工具默认串行与
readOnlyHint。 - Claude Code 工具参考:Bash 的超时与输出截断规则。
- Claude Code MCP 文档:MCP 工具的输出上限与超时配置。
- Function calling(OpenAI):
call_id、function_call_output、parallel_tool_calls与严格模式。
下一步
图 1 里第 04 道工序“权限判定”还是个黑盒,下一课「权限规则与审批模式」打开它:allow / ask / deny 规则、权限模式和沙箱如何一起决定一次调用能不能执行。