程序化工具调用(PTC)¶
PTC(Programmatic Tool Calling,程序化工具调用)让模型在一次模型回合里,用一段 Python 代码去编排多个工具调用,而不是逐个回合地调用工具。
开启后,Agent 会注册一个 run_code(code, intent) 工具。模型把 Python 代码作为 code
传进来,代码在一次性的本地 Python 子进程里执行;代码通过注入的 tools 对象回调宿主
进程里的真实工具(包括 MCP 工具),宿主把结果以统一的 envelope 返回给代码,代码用
print(...) 输出日志、用 return 返回最终值。
不是安全沙箱
子进程只是崩溃隔离和确定性 I/O 边界,不是操作系统权限沙箱。模型代码以与用户 shell 相同的权限运行:普通 Python 可以直接读写文件、起进程、访问网络,绕开工具白名单。 子进程的精简环境变量不是权限隔离,它只是让运行更确定、避免把宿主密钥带进去。详见 下文「权限与安全」。
适用场景¶
PTC 的核心价值是把一段可机械编排的工作压进一次模型回合,而不是给每次工具调用加一层包装。 原生工具可见时,单次简单调用直接使用原生工具;只有确实能减少模型往返或显著压缩输出时才优先用 PTC:
- 结果驱动的依赖链:搜索 → 从规范结果提取路径、ID 或 URL → 分批获取详情 → 去重、 分组、排序 → 返回证据。后一步参数来自前一步,不必每一步都交回模型。
- 相关查询的批量编排:在一次
run_code里提交相关查询,聚合后返回;实际并行能力由 工具契约决定,不是写了gather就一定并行。 - 有界分页与归约:跟随返回的游标,在页数、条目数、调用数和输出预算内完成处理; 返回汇总与必要证据,而不是把所有原始结果再次打印给模型。
它不是用来替代 Agent 自己的规划循环:Agent 仍然决定「做什么」,PTC 只负责把其中 一段可机械编排的工作用代码跑完。需要审批、需要写会话状态、需要返回图状态的工具不适合 放进 PTC(见下文「工具编排规则」)。 提取路径、跟随游标、应用用户给定的筛选条件属于机械步骤;判断根因、选择修复方案属于语义判断, 应交回模型。遇到未知结果结构或预算耗尽时也应停止,报告已获得的部分结果。
如何选择工具¶
run_code 的工具描述、完整 SDK 和超预算时的精简 SDK 使用一致的选择规则:
| 场景 | 建议 |
|---|---|
| 单次简单调用,原生工具可见 | 直接调用原生工具;不要仅为执行脚本而包装一次 execute,不为凑批量增加无关查询。 |
| 相关的独立查询 | 能减少模型往返时优先用 PTC;用 asyncio.gather 批量提交,遵守并发上限(默认 8)和工具契约。 |
| 下一步参数可从结果机械确定 | 在同一块内继续依赖链;阶段之间顺序 await,各阶段内部可批量提交独立查询。 |
| 分页、过滤、关联、聚合或显著的输出归约 | 优先 PTC;使用已发布的 data schema,限制工作量和输出,标注完整性。 |
| 下一步需要语义判断、结果结构未知或预算耗尽 | 结束代码块,返回必要证据、失败信息及部分结果,再由模型决定。 |
| 写/编辑类调用,以及任何会改变状态的操作 | 原生工具可见时保持原生调用;code 模式下工具被隐藏时,仅在获准情况下用最小 run_code 调用。不得绕过审批或工具限制。 |
run_code 不可用,或目标工具不可编排 |
使用可用的原生工具;不更改模式或绕过限制。 |
用户明确要求使用 run_code 时遵循该要求。以上规则通过提示引导工具选择,不是运行时强制拦截;权限检查仍独立生效。
code 模式下原生工具被折叠时,单次调用仍可能需要最小包装,这是工具可见性例外,而不是要求制造批次。
内置基础系统提示使用同样的选择规则;PTC 中间件还会在每次模型请求的系统消息中
追加完整或精简 SDK,因此已存在的外部 system_prompt.md 也能收到新的 PTC 指引。
外部提示文件不会被自动覆盖;若其中仍要求“所有只读调用默认包装,单次也不例外”,应手动删除冲突条款。
不要猜测结果 schema,或从不稳定文本中拼出未经确认的结构。分页遇到不前进的游标时停止;
仅对已知的瞬态只读失败做有界重试,不自动重试拒绝、权限错误或写操作。
每次运行都是新子进程,不存在跨 run_code 调用的 Python 变量。
启用¶
三种模式¶
tool_mode 控制 PTC 的开关与工具暴露方式:
| 模式 | 行为 |
|---|---|
native(默认) |
不注册 run_code,继续使用原生工具调用。搜索和 MCP 工具也可携带结构化 artifact,工具错误会明确标为失败。 |
both |
原生工具和 run_code 都可见。模型可以照常调用原生工具,也可以选择用 run_code 编排。 |
code |
把「可编排」工具折叠进 run_code;审批模式下需要审批的工具、以及会话状态工具仍保持原生可见。自动放行时写工具也可折叠。 |
code 模式下,如果某个请求的完整 SDK 描述超出体积预算,会退回到一个显式的精简 SDK
(只列工具名、不列 schema),并放弃折叠——也就是把全部原生工具重新暴露给模型,让模型
仍能从原生工具定义里读到准确的参数 schema,绝不会给出被截断的半个 schema。
通过环境变量¶
# 开启 both 模式(原生 + run_code)
$env:AGENT_TOOL_MODE = "both"
# 或只暴露 run_code 并折叠可编排工具
$env:AGENT_TOOL_MODE = "code"
通过 settings.json¶
PTC 的开关和上限都是非密配置项,可以写进分层 settings.json(用户层
~/.synapse/settings.json,项目层 <workspace>/.synapse/settings.json,项目层覆盖用户层):
{
"tool_mode": "both",
"ptc_timeout_seconds": 120,
"ptc_max_calls": 100,
"ptc_max_parallel": 8,
"ptc_max_output_bytes": 64000,
"ptc_max_result_bytes": 4000000,
"ptc_max_code_bytes": 64000
}
字段名与环境变量一一对应(见下文「资源上限」)。
生效方式¶
工具集在 Agent 构建时确定:改完 tool_mode 或任何上限后,需要重建 Agent / 重启会话
才会生效。通过上述环境变量或配置文件设置;PTC 没有专门的运行时开关或 UI 开关。
运行中的会话不会中途切换模式。
当前仅主 Agent 注册 PTC;子 Agent 不会继承 run_code 或 SDK。
运行模型¶
flowchart LR
M[模型] -->|"run_code(code, intent)"| MW[PTC 中间件]
MW -->|NDJSON over stdio| W[Python 子进程 worker]
W -->|"tools.name(...) / tools.call(...)"| MW
MW -->|嵌套调用| T[真实工具 / MCP]
T -->|ToolMessage| MW
MW -->|ToolEnvelope| W
W -->|"logs / value"| MW
MW -->|JSON ToolMessage| M
一次 run_code 启动一个 Python 子进程,宿主与子进程之间用逐行 JSON(NDJSON)在
stdin/stdout 上通信,帧大小有界。子进程是一次性的:跑完即销毁。打包(frozen)版本复用
同一个可执行文件(--synapse-ptc-worker 路由到 worker 入口),无需用户额外部署任何东西。
run_code(code, intent)¶
| 参数 | 类型 | 说明 |
|---|---|---|
code |
str |
作为 async 函数体执行的 Python 代码;超过 ptc_max_code_bytes 会被拒绝。 |
intent |
str |
非空字符串,说明这段代码要做什么(供审批/审计)。 |
code 会被包进一个 async def 函数体,所以顶层 await 和 return 都合法。
包装在语法树(AST)层完成,不会修改字符串字面量中的缩进、换行或尾随空格;三引号里的
Python 脚本和 PowerShell here-string 内容会保持原样。语法错误行号对应传入的 code。
注入的命名空间¶
模型代码里可以使用以下名字:
| 名字 | 说明 |
|---|---|
tools |
工具代理对象。await tools.<name>(**kwargs) 调用一个工具;await tools.call("<name>", {...}) 用于名字或参数名不是合法 Python 标识符的场景。两者都是协程,必须 await。 |
tools.available |
本次运行可调用工具名的元组(只读)。可用它自查:if "find_files" in tools.available:。 |
asyncio |
标准库 asyncio,用于 gather 等并发编排。 |
json |
标准库 json。 |
ToolCallError |
工具调用失败时抛出的异常类型。 |
只有 SDK 里列出的工具可调用。 其他名字——包括你在别处见过的、或用户提示里提到的——都
不存在,调用它会失败。若脚本里出现字面量形式的不可用工具名(tools.<name> 或
tools.call("<name>", ...)),整段运行会在启动子进程之前被拒绝,返回
kind="unknown_tool" 并附上可用工具清单,因此不会产生"先写文件、再报错"的半执行状态。
用变量动态取名字(tools.call(name, ...))无法静态判定,只能在运行期抛 ToolCallError。
结果 envelope¶
每个工具调用成功时返回一个 ToolEnvelope(dict):
{
"content": str | list, # 面向人的文本;可能被截断
"data": JSON | None, # 规范的机器可读载荷;工具未发布时为 None
"truncated": bool | None, # True/False 表示已知是否截断;None 表示完整性未知
}
data只在工具发布了规范载荷时才非空:find_files、search_files以及带structuredContent的 MCP 工具会发布,read_file等不发布。- envelope 是普通 dict,不是对象:用
res["content"]/res["data"]/res["truncated"]取值,不要写res.content。 - 不要猜
data的形状:当它是None时去读content,不要假设data["matches"]一定存在。 truncated为None表示完整性未知,不能当成「完整」。
find_files / search_files 的 data 形状:
# find_files
{"matches": [{"path": str, "is_dir": bool}, ...],
"offset": int, "next_offset": int | None}
# search_files
{"matches": [{"path": str, "line": int, "text": str}, ...],
"offset": int, "next_offset": int | None,
"output_mode": "files_with_matches" | "content" | "count"}
matches 只是当前这一页;next_offset 非 None 表示还有下一页。count 模式的计数是
本页计数,不是整仓总数。
错误¶
工具调用失败会抛出 ToolCallError,字段为:
kind:"denied"/"unknown"/"tool_error"/"command"/"limit"等;name/tool_name:出错的工具名;message:人类可读的原因。
被拒绝、未注册或需要审批的工具抛同一个异常,且从不执行。此外,若代码带着仍在飞行的工具
调用就 return,整段运行以 unfinished_calls 报错(而不是假称成功)。
工具编排规则¶
- 可编排:只读且并发安全的工具(如
read_file、find_files、search_files),以及 自动放行模式下的写工具、execute、MCP 工具。写/执行/MCP 会走正常的权限路径。 - 禁止编排(
DENIED_TOOLS):run_code自身(递归)、task、write_todos、create_goal、update_goal,以及start/check/update/cancel/list_async_task。这些工具 会写会话状态或返回图Command,从子调用里合并是不安全的,必须用原生工具。 - 被排除的工具(
excluded_tools)在代码里同样不可调用。 --require-approval下:外层run_code调用本身需要用户批准;嵌套调用里,需要审批 的写工具和没有审批契约的未知工具(例如 MCP)会被拒绝,并提示改用原生工具——这样审批 始终发生在顶层,避免嵌套调用触发 LangGraph checkpoint 重放。- 调度:只读并发安全的工具最多
ptc_max_parallel个并行;写工具和未知契约的工具 (例如没有只读契约的 MCP 工具)默认串行(独占),以免并发写互相踩踏。 通用execute的契约不是只读且不保证并发安全,因此即便放进gather,也仍然独占执行。 批量提交可以减少模型往返,但不能据此宣称 shell 执行加速。
示例¶
下面每段都可以直接作为 run_code 的 code 传入;工具必须在当前 SDK 中可用。
单次简单文件读取直接用原生 read_file,无需使用这里的包装。
有界分页与分组¶
跟随 find_files 的规范游标,最多取 3 页,每页 50 条;去重后只返回顶层目录计数。
预算耗尽或游标异常时返回部分汇总,完整性未知时不假称全部。
from collections import Counter
offset, pages = 0, 0
paths = set()
complete, reason = False, "page_budget"
unknown = False
for _ in range(3):
res = await tools.find_files(
pattern="**/*.py", path="/src", max_results=50, offset=offset,
)
pages += 1
data = res["data"]
if not isinstance(data, dict):
complete, reason = None, "unknown_data_shape"
break
paths.update(m["path"] for m in data["matches"] if not m["is_dir"])
unknown = unknown or res["truncated"] is None
next_offset = data["next_offset"]
if next_offset is None:
complete = None if unknown else not res["truncated"]
reason = "end" if complete is True else "completeness_unknown_or_truncated"
offset = None
break
if next_offset <= offset or next_offset > 1000:
reason = "invalid_cursor"
break
offset = next_offset
groups = Counter(p.lstrip("/").split("/")[0] for p in paths)
return {
"files_seen": len(paths), "by_top_dir": groups.most_common(12),
"groups_limited": len(groups) > 12, "pages": pages,
"complete": complete, "reason": reason, "next_offset": offset,
}
搜索后按命中结果读取证据¶
第二阶段的路径和行号来自第一阶段,不需要模型逐个挑选;这里的固定规则是每个文件取首个命中, 最多读 8 个文件,每批 4 个,每个文件只取命中附近 40 行、返回至多 1200 字符。 例子的批次大小不改变宿主配置的并发上限。这里只准备证据,根因和修复方案仍交回模型判断。
res = await tools.search_files(
pattern="def main", path="/src", output_mode="content", max_results=30,
)
data = res["data"]
if not isinstance(data, dict):
return {"complete": None, "reason": "unknown_data_shape"}
by_path = {}
for match in data["matches"]:
by_path.setdefault(match["path"], match)
selected = list(by_path.values())[:8]
async def read_evidence(match):
try:
doc = await tools.read_file(
file_path=match["path"], offset=max(0, match["line"] - 6), limit=40,
)
except ToolCallError as exc:
return {"path": match["path"], "error": {"kind": exc.kind, "message": exc.message}}
text = doc["content"]
if not isinstance(text, str):
return {"path": match["path"], "error": {"kind": "unknown_content_shape"}}
return {
"path": match["path"], "line": match["line"], "excerpt": text[:1200],
"preview_limited": len(text) > 1200, "truncated": doc["truncated"],
}
evidence = []
for start in range(0, len(selected), 4):
evidence.extend(await asyncio.gather(
*(read_evidence(m) for m in selected[start:start + 4])
))
discovery_complete = (
False if data["next_offset"] is not None or res["truncated"] is True
else None if res["truncated"] is None else True
)
return {
"discovery_complete": discovery_complete, "next_offset": data["next_offset"],
"files_in_page": len(by_path), "selection_limited": len(by_path) > len(selected),
"evidence": evidence,
}
并发读取多个文件:
files = ["/src/synapse/cli.py", "/src/synapse/entry.py"]
docs = await asyncio.gather(*(tools.read_file(file_path=p) for p in files))
return [d["content"][:80] for d in docs]
批量 MCP 调用:
下例的 docs__search 是占位工具名,须替换为当前 SDK 中真实注册的工具。
gather 表达批量提交;没有并发安全契约的 MCP 工具仍会被宿主串行执行。
queries = ["alpha", "beta", "gamma"]
results = await asyncio.gather(
*(tools.call("docs__search", {"query": q}) for q in queries)
)
return [r["content"] for r in results]
名字/参数名不是合法标识符时用 tools.call:
try:
res = await tools.call("my-server__some-tool", {"arg-name": 1})
return res["content"]
except ToolCallError as exc:
print(exc.kind, exc.name, exc.message)
return None
资源上限¶
| 配置字段 | 环境变量 | 默认值 | 取值范围 |
|---|---|---|---|
tool_mode |
AGENT_TOOL_MODE |
native |
native / both / code |
ptc_timeout_seconds |
AGENT_PTC_TIMEOUT_SECONDS |
120 |
0.1–3600 |
ptc_max_calls |
AGENT_PTC_MAX_CALLS |
100 |
1–10000 |
ptc_max_parallel |
AGENT_PTC_MAX_PARALLEL |
8 |
1–64 |
ptc_max_output_bytes |
AGENT_PTC_MAX_OUTPUT_BYTES |
64000 |
256–16000000 |
ptc_max_result_bytes |
AGENT_PTC_MAX_RESULT_BYTES |
4000000 |
1024–64000000 |
ptc_max_code_bytes |
AGENT_PTC_MAX_CODE_BYTES |
64000 |
256–1000000 |
含义:
ptc_timeout_seconds:单次运行的总超时。超时会终止 worker 并清理可追踪的进程树、停止派发,并请求 取消在途调用。ptc_max_calls:单次运行允许的工具调用次数上限。超预算会终止整段运行。ptc_max_parallel:同时进行的子调用数上限。ptc_max_code_bytes:code的字节上限。ptc_max_result_bytes:单个中间工具结果及最终结果帧的上限。工具结果过大抛出ToolCallError(kind="limit");最终值过大则整段运行失败。它不是模型上下文的容量。ptc_max_output_bytes:最终{logs, value, error?}的整体上限。超出时以显式的output_limit错误返回(保留能放下的日志前缀),绝不静默成功——这样大value也无法绕过 整体输出预算混进模型上下文。
权限与安全¶
- 只读模式(
--readonly)下run_code完全禁用:普通 Python 仍可直接写操作系统,所以 唯一安全的只读行为就是根本不提供这个工具。显式把run_code放进excluded_tools也会 禁用。 --require-approval:外层run_code需要用户批准;嵌套的写工具和未知契约工具被拒绝 (提示改用原生工具),审批只在顶层发生。- 自动放行模式:写/
execute/MCP 经正常权限路径自动放行;只读并发安全的工具最多max_parallel个并行,写工具与未知工具串行。 - 不是沙箱:模型代码与用户 shell 同权限,可绕过工具白名单直接操作系统。空/白名单环境变量 不等于权限隔离。
- 超时/取消:会终止 worker、清理可追踪的进程树、停止派发并请求取消在途调用;但同步工具跑在线程上, 线程无法被可靠强杀,已经发生的写操作不会回滚。若某个同步处理器在运行超时后仍在后台跑, 结果里会附带一条明确警告。
- 进程树限制:未使用 Windows Job Object。如果脚本主动
os._exit()提前退出,Windows 无法再用taskkill /T追踪其子进程;主动脱离进程组的后代也不受普通进程组清理保证。 - 预算限制执行时长、调用数与传输大小,没有 OS 级内存配额,不能防止任意 Python 耗尽内存。
- 不要据此宣称「绝对安全」或任何性能数字——这些边界都受宿主、工具实现和操作系统调度影响。
UI 与状态¶
- TUI / 时间线会把
run_code的子调用作为嵌套行显示(开始/结束、参数摘要、结果预览); 参数摘要有长度限制,移除常见凭据字段并摘要化正文;这不替代工具自身对敏感结果的保护。 - 子调用的中间结果不会进入模型的 messages:只有父级
run_code的那条ToolMessage会被写入会话历史。子调用结果只以 envelope 形式返回给子进程内的代码。
限制与注意事项¶
- PTC 是单次运行:一次
run_code= 一个子进程,跑完即销毁,没有跨调用状态。 - 审批模式下需审批的工具,以及写会话状态、返回图
Command的工具,必须使用原生调用。 - 工具集与上限在 Agent 构建时固定,改动需重建 Agent / 重启会话;没有运行时或 UI 开关。
- 子进程是 stdlib-only,不引入新依赖。