Skip to content

Batch Dispatch — 批量工具调用与 Eval 沙箱

源码:python/dcc_mcp_core/batch.py · Issue #406

English

服务端批量执行工具调用,减少 Agent 往返和 Token 消耗。中间结果不会进入模型上下文——只有最终聚合值返回。

何时使用

  • batch_dispatch — 已知要连续调用的 N 个工具;只关心合并后的汇总,不希望逐步响应进入上下文。
  • EvalContext — Agent 写一段短 Python 脚本,根据中间结果条件选择工具("Cloudflare 模式")。配合 DccApiExecutor 覆盖大型 DCC API。

两者均为纯 Python,在 Rust 级 tools/batch MCP 端点落地前即可使用任意 ToolDispatcher

导入

python
from dcc_mcp_core import batch_dispatch, EvalContext

batch_dispatch(dispatcher, calls, *, aggregate="list", stop_on_error=False) -> dict

按顺序执行 (tool_name, arguments_dict) 列表,返回单一聚合汇总。

参数

名称类型默认说明
dispatcherToolDispatcher需暴露 .dispatch(name, json_str) -> dict
callslist[tuple[str, dict]]有序 (tool_name, args) 列表
aggregate"list" | "merge" | "last""list"详见下表
stop_on_errorboolFalseTrue 时首次失败即中止

聚合模式

模式结果键形态
"list""results"单次调用返回值列表
"merge""merged"合并每次的 output 字典(后者覆盖前者)
"last""last"仅最后一次结果

返回值始终包含 totalsucceedederrors,以及按 aggregate 附加的 results / merged / last 之一。

python
summary = batch_dispatch(
    dispatcher,
    [
        ("get_scene_objects", {}),
        ("get_render_stats", {"layer": "beauty"}),
        ("get_render_stats", {"layer": "specular"}),
    ],
    aggregate="merge",
)
print(summary["total"], summary["succeeded"])
print(summary["merged"])

EvalContext(dispatcher, *, sandbox=True, timeout_secs=30)

dispatch(name, args) 的沙箱脚本执行环境——对应规划中的 dcc_mcp_core__eval MCP 内建工具。交给 Agent 一个受限的 Python 解释器,在循环中编排几十次工具调用,而不让中间值进入上下文。

参数类型默认说明
dispatcherToolDispatcher
sandboxboolTrue__builtins__ 中剥离 openexeceval__import__compilegetattr/setattr/delattrvars/dirglobals/locals
timeout_secsint | None30仅 POSIX(signal.SIGALRM);Windows 下自动跳过

方法 .run(script: str) -> Any

  • 会把脚本包装成函数体,使顶层 return <expr> 生效。
  • 最后一行表达式不会隐式返回(与 REPL 不同)。
  • 超时抛 TimeoutError(仅 POSIX),其他异常抛 RuntimeError
  • 脚本内可使用 dispatch(tool_name, args_dict) 调用任意已注册工具。
python
ctx = EvalContext(dispatcher)
keyframes = ctx.run("""
frames = []
for i in range(1, 11):
    r = dispatch("get_frame_data", {"frame": i})
    if r.get("output", {}).get("has_keyframe"):
        frames.append(i)
return frames
""")
# 仅最终列表回到 Agent——10 次调用只消耗 1 次返回的 Token 量。

安全注意

  • sandbox=True尽力而为——隐藏了危险 builtin 但不是 OS 级隔离。可信度视作"自家 Agent 生成的半可信代码",而非任意用户输入。
  • 处理不可信输入时请配合 SandboxPolicy 并运行于子进程/容器。
  • 脚本里工具调用失败不会抛异常——返回标准错误字典,便于脚本按惯用方式处理。

落地现状

Python 助手已可用。Rust 级 tools/batchdcc_mcp_core__eval 内建 MCP 工具跟踪于 issue #406,将复用本模块逻辑。

参见

Released under the MIT License.