跳到主要内容

工具即 MCP 服务:子进程、搜索/抓取/沙箱/推理工具

30 秒导读: 模型能"想",但要真去搜网页、跑代码、抓页面,得有"手脚"。MiroThinker 把每一件手脚都做成一个独立的 MCP 服务(Model Context Protocol,一种标准化的"模型-工具"通信协议)。本章讲这层手脚怎么实现:ToolManager 每调一次工具就起一个子进程会话,把结果统一成 {server_name, tool_name, result|error} 回给编排循环;并逐个看几个真实工具——搜索、抓取、沙箱、"强模型当工具"、待办清单。

本章讲的是"手脚层"本身怎么落地。相邻章节各管一段:工具怎么被配置进来在 01-config-and-assembly;工具怎么被主循环调度在 02-orchestrator-loop;模型给的调用参数怎么容错修复在 03-robustness-rollback。这里只关心:一次工具调用在底层是怎么跑起来、跑完的,以及每个工具各干什么。

1. 这是什么(零基础也能懂)​

先建立一个心智模型​

一个"深度研究 agent"的模型本身只会一件事:输出文字。它没法真的去 Google 搜索、没法真的跑一段 Python、没法真的打开一个网页。这些动作必须由外部程序替它执行,再把结果喂回去。这层外部程序,就是"工具"。

MiroThinker 没有把工具写成"函数直接调用",而是把每一类工具做成一个独立的小服务,用 MCP 协议跟主程序通信。

  • MCP(Model Context Protocol): 一种约定"模型这边怎么发现工具、怎么调工具、工具怎么回结果"的标准协议。你可以把它想成"工具界的 USB 口":只要按这个口的形状做,主程序就能即插即用地挂上任意工具。
  • 服务(server): 一个工具服务就是一个能独立启动的 Python 程序(比如 search_and_scrape_webpage.py),里面用 @mcp.tool() 装饰器登记了若干工具函数。

一句话类比​

把 ToolManager 想成一个外包调度台:主 agent 说"帮我搜一下 X",调度台就临时雇一个工人(启动一个工具子进程)、把活派过去、拿回成果、然后把工人辞退(关掉子进程)。下次再有活,再临时雇一个。唯一的例外是浏览器——那个工人贵、且要记住"我现在停在哪个页面",所以长期留用。

它能做什么(本章覆盖的几个工具)​

工具(服务)干什么源文件
google_search / sogou_search网络搜索(Serper / 腾讯云搜狗)dev_mcp_servers/search_and_scrape_webpage.py
create_sandbox / run_python_code …在 E2B 云沙箱里跑代码mcp_servers/python_mcp_server.py
reasoning把 claude-3-7-sonnet 当一个"推理工具"来调mcp_servers/reasoning_mcp_server.py
add_todo / list_todos / complete_todo给 agent 一份可读写的待办清单dev_mcp_servers/task_planner.py
playwright(浏览器)常驻会话,操作真实浏览器mcp_servers/browser_session.py

2. 顶层全景(一次工具调用怎么跑)​

本节讲"大盘":从主循环喊出"调用某工具",到底层子进程跑完、结果回来,中间经过了什么。

怎么读这张图​

从上到下是时间顺序。核心是中间那个虚线框——它对每一次调用都完整走一遍:起子进程 → 建会话 → 调工具 → 关会话。

   编排主循环 (02 章)
        │  execute_tool_call(server_name, tool_name, arguments)
        ▼
  ┌───────────────── ToolManager.execute_tool_call ─────────────────┐
  │  ① 查配置:这个 server_name 的启动参数是什么?                    │
  │  ② 是 "playwright" 吗?—— 是 → 走【常驻】PlaywrightSession        │
  │                            否 → 走【一次性】子进程会话 ↓         │
  │  ┌─────────────── 一次性 MCP 会话(每调用一次) ──────────────┐  │
  │  │  stdio_client(params)  ← 起一个工具子进程                    │  │
  │  │    └ ClientSession     ← 握手 initialize()                   │  │
  │  │        └ call_tool(tool_name, arguments)  ← 真正干活          │  │
  │  │        └ 取 content[-1].text 当结果                          │  │
  │  │  (with 语句退出 → 子进程关闭)                                │  │
  │  └──────────────────────────────────────────────────────────┘  │
  │  ③ 防作弊:是在抓 HF 数据集找答案吗?→ 是则替换成警告文本         │
  │  ④ 出错兜底:scrape 失败 → 试 MarkItDown fallback               │
  │  ⑤ 打包成 {server_name, tool_name, result | error}             │
  └──────────────────────────────────────┬─────────────────────────┘
        │                                 │
        ▼                                 ▼
  tool_executor 后处理                  回到主循环
  (DEMO_MODE 截断到 20000 字)

部件一句话职责​

部件干什么在哪
ToolManager工具调用的总入口:发现工具、执行工具、收口结果manager.py:48 ToolManager
get_all_tool_definitions启动时连上所有 server,列出它们有哪些工具(喂给 prompt)manager.py:104
execute_tool_call执行一次工具调用,是本章主角manager.py:198
PlaywrightSession浏览器专用的常驻会话(不每次重启)browser_session.py:15 PlaywrightSession
with_timeout装饰器:给任意 async 函数套一个超时manager.py:19 with_timeout
ToolExecutor(app 层)调用前的参数修复、调用后的结果后处理tool_executor.py:30 ToolExecutor

3. 核心原理​

3.1 每次调用起一个子进程会话(MCP 生命周期)​

它要解决的小问题: 工具服务是独立程序,主程序怎么跟它对话?MiroThinker 的答案是——不常驻,用完即弃。

思路: MCP 的 stdio 传输(标准输入输出)本质是:启动一个子进程,通过它的 stdin/stdout 收发 JSON-RPC 消息。stdio_client(server_params) 是个异步上下文管理器——进入 with 时起子进程,退出 with 时关子进程。所以"一次调用 = 一次 with"。

原理演示(帮你建立直觉):

# 示意,非源码 —— 一次工具调用的骨架
async with stdio_client(server_params) as (read, write):   # 起子进程,拿到读写管道
    async with ClientSession(read, write) as session:      # 建 MCP 会话
        await session.initialize()                         # 握手
        result = await session.call_tool(name, arguments)  # 真正调工具
        text = result.content[-1].text                     # 取最后一段文本当结果
# with 退出 → 子进程被关掉,下次调用重新起

真实实现: 执行分支在 manager.py:250-264——判断参数是 StdioServerParameters 就走 stdio,取 tool_result.content[-1].text 作为结果:

# manager.py:257-264
tool_result = await session.call_tool(tool_name, arguments=arguments)
result_content = (
    tool_result.content[-1].text if tool_result.content else ""
)

还有一条 SSE / HTTP 分支: 如果 server_params 是个 http:// / https:// 字符串,就改用 sse_client(server_params) 走 SSE(Server-Sent Events)远程连接(manager.py:279-282),逻辑与 stdio 分支对称。这让工具既能是本地子进程,也能是远程 HTTP 服务。

列工具定义时同理: get_all_tool_definitions(manager.py:104)在启动阶段对每个 server 各起一次会话、session.list_tools() 拉工具清单;若某 server 连不上,不崩,而是把这台的 tools 记成一条 {"error": ...} 继续(manager.py:190-193),保证其它工具照常可用。

3.2 浏览器是例外:常驻会话​

它要解决的小问题: 浏览器有"状态"——你翻到第 3 页,下一步操作得接着这一页。若每次调用都重开浏览器,状态全丢了。

思路: 对 server_name == "playwright",不走"用完即弃",而是第一次调用时建一个长活会话并缓存,之后所有浏览器操作复用它。

真实实现: execute_tool_call 里对 playwright 单开一支(manager.py:228-240):self.browser_session 为空才 PlaywrightSession(...) + connect(),否则直接复用:

# manager.py:230-235
if self.browser_session is None:
    self.browser_session = PlaywrightSession(server_params)
    await self.browser_session.connect()
tool_result = await self.browser_session.call_tool(tool_name, arguments=arguments)

PlaywrightSession(browser_session.py:15)把 read/write/session 存成实例属性、手动 __aenter__ 保活,直到显式 close() 才退出——这正是"常驻"与"一次性"的区别。

3.3 超时护栏:with_timeout 装饰器​

它要解决的小问题: 工具可能卡死(网络挂起、沙箱死循环)。不能让整个 agent 陪着它一起卡。

思路: 用一个装饰器把任意 async 函数包进 asyncio.wait_for(...),到点就抛 TimeoutError。

真实实现: with_timeout(manager.py:19-36)。execute_tool_call 头上挂的是 @with_timeout(1200)(manager.py:197)——单次工具调用最多 1200 秒:

# manager.py:31-32
async def wrapper(*args, **kwargs) -> R:
    return await asyncio.wait_for(func(*args, **kwargs), timeout=timeout_s)

3.4 防作弊:不许抓 HuggingFace 数据集找答案​

它要解决的小问题: MiroThinker 常在 benchmark 上跑,而很多题的标准答案就躺在 HuggingFace 的 dataset / space 里。如果 agent 学会"直接去抓那个数据集页面",就等于作弊,分数不真实。

思路: 两道闸——搜索端过滤 + 抓取端拦截。

闸一(搜索结果里就抹掉): google_search 在整理 organic 结果时,_is_banned_url 命中 huggingface.co/datasets、huggingface.co/spaces、unifuncs 就跳过(search_and_scrape_webpage.py:61-74、:168)——agent 根本看不到这些链接。

闸二(就算拿到 URL 也抓不成): _should_block_hf_scraping(manager.py:87-98)判断"工具是 scrape 且 URL 是 HF 数据集/space"。命中后,即便子进程已经抓回内容,也把结果整个替换成一句警告(manager.py:266-267):

# manager.py:266-267
if self._should_block_hf_scraping(tool_name, arguments):
    result_content = "You are trying to scrape a Hugging Face dataset for answers, please do not use the scrape tool for this purpose."

注意这是事后(post-hoc)拦截——先抓、再判、再替换文本,而非提前拒调用。stdio 和 SSE 两条分支各放了一份同样的检查(manager.py:266、:297)。

3.5 抓取失败的兜底:MarkItDown fallback​

它要解决的小问题: 主抓取工具偶尔会对某些页面报 "unhandled errors"。直接返回失败太可惜,值得再试一条路。

思路: 捕到 scrape 类工具的这种错误时,改用 MarkItDown(一个能把 URL/文档转成 Markdown 的库)再抓一次,成了就返回它的文本。

真实实现: execute_tool_call 的外层 except(manager.py:337-364):

# manager.py:337-354(节选)
if (tool_name in ["scrape", "scrape_website"]
        and "unhandled errors" in error_message
        and "url" in arguments and arguments["url"] is not None):
    from markitdown import MarkItDown
    md = MarkItDown(docintel_endpoint="<document_intelligence_endpoint>")
    result = md.convert(arguments["url"])
    return {..., "result": result.text_content}

fallback 再失败也不抛,记个日志、落到最终的统一 error 返回(manager.py:365-381)。

3.6 结果统一成一个信封​

关键设计: 不管成功、失败、被拦、还是 fallback,execute_tool_call 永远返回同一个形状的字典,循环那头只认这一种信封:

字段含义
server_name哪个服务
tool_name哪个工具
result 或 error二选一:文本结果,或错误说明

成功走 manager.py:321-325,各类失败走对应的 error 分支。统一信封让 02 章 的循环和 03 章 的回滚判断都能用一套逻辑处理。

4. 具体工具举例​

本节逐个看几个工具"里面长什么样"。它们都用 @mcp.tool() 登记,函数的 docstring 就是给模型看的说明书(MCP 会把它当工具描述)。

4.1 搜索:google_search / sogou_search​

search_and_scrape_webpage.py 里两个搜索工具走不同后端:

  • google_search(:77): 打 Serper(https://google.serper.dev/search)。带 make_serper_request(:47)做重试(tenacity,指数退避,最多 3 次)。有个小巧思——若结果为空且 query 里含引号,就去掉引号再搜一次(:179-185),因为过度精确的引号短语常常搜不到东西。
  • sogou_search(:226): 打腾讯云 SearchPro(搜狗),docstring 明说"中文查询比 Google 更好"(:232),把返回精简成 title/url/passage/date/site 几个字段。

两者都先校验 API key 和空 query,再返回 JSON 字符串;URL 都过一遍 decode_http_urls_in_dict 解码。

4.2 沙箱:python_mcp_server(E2B 云沙箱执行代码)​

python_mcp_server.py 把"跑代码"外包给 E2B(一个云端隔离沙箱服务)。核心是沙箱按 id 复用:

  • create_sandbox(:89): 起一个 Linux 沙箱,返回一句含 sandbox_id 的文本(:114)。带 5 次重试 + 指数退避;超时上限锁死在 DEFAULT_TIMEOUT=600 秒(:100)。
  • run_python_code(:173)/ run_command(:129): 拿着 sandbox_id 连回同一个沙箱跑代码(Sandbox.connect(sandbox_id)),每次执行都刷新 timeout。docstring 反复叮嘱"尽量复用已有沙箱"(:178)——这样多步之间的文件、变量都还在。
  • 禁用名单 INVALID_SANDBOX_IDS(:31-53): 一堆模型爱瞎编的假 id("default"、"sandbox"、"auto"、"0"…)。传这些进来会被拒(:139-140),run_python_code 则退化成"临时起一个、跑完就 kill"的无状态模式(:184-198)。这正好接住了 03 章 提到的容错——模型漏填 sandbox_id 时,app 层补个 "default",到这里就触发无状态兜底。

4.3 推理:reasoning(强模型当工具)​

reasoning_mcp_server.py 只有一个工具 reasoning(:20),模式很特别——把一个更强的模型当成一件"工具"来调。

它内部直接 new 一个 Anthropic 客户端,调 claude-3-7-sonnet-20250219,并开启 extended thinking(budget_tokens=19000,:46-49),专门啃"硬数学 / 谜题 / IQ 题"。docstring 明确警告"简单问题别用它"(:22)。

# reasoning_mcp_server.py:43-52(节选)
response = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=21000,
    thinking={"type": "enabled", "budget_tokens": 19000},
    messages=messages_for_llm, stream=False,
)

返回时取 content[-1].text;若只回了思考内容没回正文,就退而取 .thinking(:54-58)。这体现了"模型即工具"的设计:主 agent 用便宜/快的模型跑流程,遇到真难题就"外包"给一个开了深度思考的强模型。

4.4 待办:task_planner(给 agent 做计划)​

task_planner.py 给 agent 一份能读写的待办清单,逼它先规划再动手:

  • add_todo(titles)(:99): 一次加一条或多条任务;docstring 用大写强调"动手前 MUST 先建完整计划"(:104)。
  • list_todos()(:159)/ complete_todo(todo_ids)(:174): 查看全部、把某几条标记完成(id 支持只给前 8 位,:206)。
  • 按 TASK_ID 隔离: 清单存成 todos_{TASK_ID}.json(:32),而 TASK_ID 是必需环境变量,缺了直接 raise(:25-30)。这样并发跑多个任务时,各自的待办互不串台。

清单每次都渲染成 Markdown checklist 返回(format_todos_as_markdown,:62),模型读起来一目了然。

5. 结果后处理:DEMO_MODE 截断​

工具结果回到 app 层后,还有一道后处理——但只在 demo 模式生效。抓取回来的网页动辄几十万字,会撑爆上下文;演示时截短能多撑几轮对话。

真实实现: tool_executor.py 的 post_process_tool_call_result(:214):只有 DEMO_MODE == "1" 且工具是 scrape / scrape_website 时,才对结果调 get_scrape_result(:193)截断:

# tool_executor.py:230-237(节选)
if os.environ.get("DEMO_MODE") == "1":
    if "result" in tool_call_result and tool_name in ["scrape", "scrape_website"]:
        tool_call_result["result"] = self.get_scrape_result(tool_call_result["result"])

get_scrape_result(:193-212)把结果当 JSON 解析、取 text 字段,超过 DEMO_SCRAPE_MAX_LENGTH = 20_000(:27)就切到 2 万字;解析不了就当纯字符串同样切。

一个坑: 沙箱那边(python_mcp_server.py)有它自己的 MAX_RESULT_LEN = 20_000(:27)和 truncate_result(:73),对代码执行结果无条件截断——和这里的 demo 截断是两套独立机制,别混淆:一个总在沙箱内生效,一个只在 demo 模式对 scrape 生效。

6. 边界与局限​

  • 每次调用重起子进程有开销。 非浏览器工具"用完即弃",省心但每次都要付进程启动 + initialize() 握手的成本;高频调用不是免费的。
  • 防作弊是"事后"的。 HF 拦截先抓再替换(manager.py:266),流量/时间已经花掉了,只是不给模型看内容。
  • MarkItDown fallback 很窄。 只对 scrape/scrape_website 且错误串里含 "unhandled errors" 触发(manager.py:337-342);别的抓取失败模式不会走兜底。
  • 沙箱禁用名单是硬编码黑名单。 INVALID_SANDBOX_IDS(python_mcp_server.py:31)只能拦"已知的常见假 id",模型若编个没列进去的假 id,会走到 Sandbox.connect 失败再报错,而非被提前拦下。
  • reasoning 工具把模型型号写死。 claude-3-7-sonnet-20250219 直接硬编码(reasoning_mcp_server.py:44),换模型要改代码。

7. 横向对比(同 shelf)​

  • 与 01-config-and-assembly:那里讲"哪些 server 被装进 ToolManager";本章讲"装进来之后每个 server 怎么被起、被调"。
  • 与 02-orchestrator-loop:循环负责"决定调哪个工具、按什么节奏调";本章的 execute_tool_call 是循环真正落地的那一下。
  • 与 03-robustness-rollback:ToolExecutor.fix_tool_call_arguments 在调用前修参数(如补 sandbox_id="default");本章的兜底(HF 拦截、MarkItDown、无状态沙箱)发生在调用中/后。

8. 代码地图(导航索引)​

主题文件路径符号名
工具管理总入口libs/miroflow-tools/src/miroflow_tools/manager.pyToolManager
列出所有工具定义(启动阶段)libs/miroflow-tools/src/miroflow_tools/manager.pyget_all_tool_definitions
执行一次工具调用(MCP 生命周期)libs/miroflow-tools/src/miroflow_tools/manager.pyexecute_tool_call
超时装饰器libs/miroflow-tools/src/miroflow_tools/manager.pywith_timeout
HF 抓取防作弊(事后拦截)libs/miroflow-tools/src/miroflow_tools/manager.py_should_block_hf_scraping
浏览器常驻会话libs/miroflow-tools/src/miroflow_tools/mcp_servers/browser_session.pyPlaywrightSession
Google/Sogou 搜索 + 搜索端 URL 过滤libs/miroflow-tools/src/miroflow_tools/dev_mcp_servers/search_and_scrape_webpage.pygoogle_search、sogou_search、_is_banned_url
E2B 沙箱创建/复用 + 禁用名单libs/miroflow-tools/src/miroflow_tools/mcp_servers/python_mcp_server.pycreate_sandbox、run_python_code、INVALID_SANDBOX_IDS
强模型当工具(推理)libs/miroflow-tools/src/miroflow_tools/mcp_servers/reasoning_mcp_server.pyreasoning
待办清单(按 TASK_ID 隔离)libs/miroflow-tools/src/miroflow_tools/dev_mcp_servers/task_planner.pyadd_todo、list_todos、complete_todo
结果后处理 / demo 截断apps/miroflow-agent/src/core/tool_executor.pypost_process_tool_call_result、get_scrape_result