跳到主要内容

工具层:MCP server 池、执行与容错纠错

30 秒导读: 如果说 main agent 是大脑,那这一章讲的是它的手脚——工具系统。MiroFlow 把每个工具(搜索、跑 Python、读文件、看图、听音频……)做成一个独立的 MCP server 子进程;ToolManager 是这些进程的总管:开机时把它们的工具清单列给大脑看,运行时按"哪个 server、哪个工具、什么参数"去调用,并在调错 server、爬网页失败、调用超时等各种意外里自动纠错兜底。本章只讲工具的定义 → 加载 → 执行 → 容错,不碰"LLM 怎么从文本里解析出工具调用"(那是 04)。

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

一句话定义: 工具层 = 一组独立的外部能力进程 + 一个管理它们的 ToolManager。

为什么 agent 需要工具? 语言模型本身只会"生成文字",它不会真的去 Google、不会真的跑一段 Python、不会真的打开浏览器。要让它"做事",就得给它手脚:一套可以被调用、能产生真实副作用(联网、算数、读文件)的外部工具。

MCP 是什么? MCP(Model Context Protocol,模型上下文协议)是一个标准约定:工具端起一个 server,声明"我有哪些工具、每个工具要什么参数",客户端连上去就能列举和调用。MiroFlow 用它把每个工具做成一个可插拔的独立进程。

一句话直觉/类比: 把工具层想成一个外包中介公司:

  • 每个 MCP server = 一个专业外包团队(搜索团队、代码团队、视觉团队……),各自独立办公(独立进程)。
  • ToolManager = 中介:大脑说"我要搜个东西",中介知道该派哪个团队、怎么把活派过去、团队罢工了怎么补救。
  • 好处:某个团队崩了(某个工具进程报错),不影响别的团队,也不会拖垮大脑——中介会兜底。

用起来什么样(概念示意): 大脑产出一个工具调用意图,ToolManager 负责把它落地:

# 示意,非源码:一次工具调用在管理层看起来就这么简单
result = await tool_manager.execute_tool_call(
    server_name="tool-searching",   # 派给哪个"外包团队"
    tool_name="google_search",      # 让它做哪件事
    arguments={"q": "MiroFlow architecture"},
)
# result 是 {"server_name":..., "tool_name":..., "result": "..."} 或 {..., "error": "..."}

本节不出现底层细节。 记住三件事:工具是独立进程、由 ToolManager 统一调度、意外时会自动兜底。

2. 顶层全景(它大概怎么转)​

工具层的生命周期分三段:配置 → 池化(拉起进程)→ 执行与容错。

怎么读这张图:从左到右是时间顺序;上半是"加载期",下半是"运行期"。

【加载期】
config/tool/*.yaml ──create_mcp_server_parameters()──► [server_configs 列表 + blacklist]
  (每个 yaml = 一个工具)         (tool_utils.py)              │
                                                              ▼
                                                    ToolManager(configs, blacklist)
                                                              │
                                          get_all_tool_definitions() 逐个连一遍
                                                              │  拿到"每个 server 有哪些工具"
                                                              ▼
                                                    交给 prompt 生成器 → 展示给 LLM

【运行期】  LLM 产出 (server_name, tool_name, arguments)
                                                              │
                                                execute_tool_call(...)  @with_timeout(900)
                                                              │
                    ┌─────────────────────────────────────────┼───────────────────────────┐
                    ▼                                          ▼                           ▼
             server 找不到?                            server == "playwright"?         普通 stdio/SSE
             _find_servers_with_tool                   走持久 PlaywrightSession        起子进程调用
             自动纠错 + 递归重试                                                          │
                                                                          scrape 失败 → MarkItDown 兜底
                                                                          爬 HF 数据集 → 拒绝(防作弊)

部件一句话职责:

部件干什么在哪个文件
create_mcp_server_parameters读 config/tool/*.yaml,把每个工具变成 StdioServerParameters,并收集黑名单src/utils/tool_utils.py:21-58
ToolManager工具总管:列举 / 执行 / 容错src/tool/manager.py:66-517
get_all_tool_definitions逐个连 server,列出可用工具清单(过黑名单、失败不致命)src/tool/manager.py:171-252
execute_tool_call执行一次工具调用,带超时 + 一堆兜底src/tool/manager.py:254-517
_find_servers_with_toolserver 名错了时,反查"哪个 server 有这个工具"src/tool/manager.py:110-169
PlaywrightSession浏览器工具的持久会话(跨调用保活)src/tool/mcp_servers/browser_session.py:21-65
各 MCP server 本体真正干活的工具进程(FastMCP + @mcp.tool())src/tool/mcp_servers/*.py

主线走一遍(高层):

  1. 编排器(01)初始化时,先用 create_mcp_server_parameters 把 yaml 配置变成 server 池,交给 ToolManager。
  2. 开跑前调 get_all_tool_definitions,把"有哪些工具"喂给 prompt,让 LLM 知道能用什么。
  3. 循环里 LLM 说要调某工具,编排器调 execute_tool_call,由它落地、容错、返回结果。

3. 核心原理(逐个机制,由浅入深)​

3.1 配置怎么变成 server 池​

它要解决的小问题: 我有一堆工具,每个工具怎么启动、要哪些环境变量各不相同——怎么用一份声明式配置统一描述,并批量拉起?

思路: 一个工具 = 一个 config/tool/*.yaml。yaml 里只写三样:进程用什么命令启动(tool_command)、传什么参数(args)、注入哪些环境变量(env)。加载时把它翻译成 MCP 标准的 StdioServerParameters。

yaml 长什么样(真实配置):

# config/tool/tool-searching.yaml —— 一个工具就是一份这样的声明
name: "tool-searching"
tool_command: "python"
args:
  - "-m"
  - "src.tool.mcp_servers.searching_mcp_server"   # 以模块方式起这个进程
env:
  SERPER_API_KEY: "${oc.env:SERPER_API_KEY}"       # 运行时从 .env 注入
  JINA_API_KEY: "${oc.env:JINA_API_KEY}"

真实实现: create_mcp_server_parameters 遍历 agent_cfg["tool_config"] 里列的工具名,逐个 OmegaConf.load 对应 yaml,组装成 {"name":..., "params": StdioServerParameters(...)},见 src/utils/tool_utils.py:create_mcp_server_parameters:21-58。

一个细节:tool_command 若是 "python",实际用的是 sys.executable(当前解释器),保证子进程和主进程用同一个 Python 环境:

# src/utils/tool_utils.py:41-46 —— "python" 被换成当前解释器路径
command=sys.executable if tool_cfg["tool_command"] == "python" else tool_cfg["tool_command"],
args=tool_cfg.get("args", []),
env=tool_cfg.get("env", {}),

黑名单也在这里收集。 某个 agent 想禁用某个(server, tool)组合,就在配置里列 tool_blacklist,加载时被转成一个 set,元素是 (server_name, tool_name) 元组,返回给 ToolManager,见 src/utils/tool_utils.py:55-58。

容错细节: 单个 yaml 加载失败(文件缺失、格式错)只 logger.error 后 continue,不会让整个池化崩掉(tool_utils.py:49-53)——一个工具坏了,不拖累其它工具。

关于 sub-agent 如何"伪装成工具"混进这个池(expose_sub_agents_as_tools,tool_utils.py:78-96),见 02,本章不展开。

3.2 列举工具清单:get_all_tool_definitions​

它要解决的小问题: LLM 得先知道"我手上有哪些工具、每个工具叫什么、要什么参数",才能决定调谁。怎么从这一池子进程里把清单收集齐?

思路: 逐个 server 连上去问一句 "你有哪些工具"(MCP 的 list_tools),边收集边过黑名单。

两种传输方式: 一个 server 的参数要么是 StdioServerParameters(本地子进程,用 stdio_client),要么是一个 http:///https:// 字符串(远程 SSE 端点,用 sse_client)。代码对两种分别处理,见 src/tool/manager.py:185-235。

连接 + 列举的骨架(真实实现节选):

# src/tool/manager.py:186-207 —— stdio 分支:起子进程、初始化会话、列工具、过黑名单
async with stdio_client(update_server_params_with_context_var(server_params)) as (read, write):
    async with ClientSession(read, write, sampling_callback=None) as session:
        await session.initialize()
        tools_response = await session.list_tools()
        for tool in tools_response.tools:
            if (server_name, tool.name) in self.tool_blacklist:   # 黑名单命中就跳过
                continue
            one_server_for_prompt["tools"].append(
                {"name": tool.name, "description": tool.description, "schema": tool.inputSchema}
            )

失败不致命(关键设计): 如果某个 server 连不上或列举报错,不是抛异常终止,而是照样把这个 server 记进结果,只是它的工具列表里放一条 {"error": ...},见 src/tool/manager.py:242-250。这样一个工具进程挂了,LLM 仍能看到并使用其它所有工具。

收尾: 返回一个 [{"name": server, "tools": [...]}, ...] 列表,交给 prompt 生成器渲染成 LLM 能读的工具说明。

3.3 执行一次调用:execute_tool_call 与它的容错巧思​

它要解决的小问题: 拿到 (server_name, tool_name, arguments) 后,怎么可靠地把这次调用落到真实进程上——并且当出各种意外时不至于让整轮任务崩掉?

先看外壳:超时。 整个方法被 @with_timeout(900) 包住,即任何一次工具调用最多跑 900 秒(15 分钟),超时自动 asyncio.wait_for 中断,见 src/tool/manager.py:254(装饰器定义在 with_timeout:37-54)。防止某个工具(比如卡死的爬虫)无限期挂住主循环。

execute_tool_call 的返回恒定是一个字典,要么含 "result",要么含 "error"——从不抛给上层,把失败也变成一条可读的数据。这是全局容错哲学的一部分(见 05)。

下面是它内部的几处容错巧思,由浅入深。

巧思 A:server 名找错了,自动纠错并递归重试​

场景: LLM 有时会把 server 名和 tool 名搞混,或写了一个不存在的 server。

做法: 先 get_server_params 查不到,就调 _find_servers_with_tool(tool_name) 反查——遍历所有 server,问它们各自的工具清单,看谁有这个 tool_name(src/tool/manager.py:110-169)。然后按命中数量分三档处理:

反查结果处理代码
恰好 1 个 server 有这工具自动纠错:用正确 server 名递归调用自己,并在结果里加 [Auto-corrected: ...] 说明manager.py:276-307
多个 server 都有不猜,把候选列表塞进 error,提示 LLM 自己选manager.py:309-310
没有提示"可能把 server/tool 搞混了,请重试并核对"manager.py:311-315

递归纠错的核心几行:

# src/tool/manager.py:283-296 —— 只有唯一候选时才敢自动改,并如实标注
corrected_result = await self.execute_tool_call(correct_server, tool_name, arguments)
if "result" in corrected_result:
    correction_note = f"[Auto-corrected: Server '{server_name}' not found, ... used '{correct_server}' instead] "
    corrected_result["result"] = correction_note + str(corrected_result["result"])
    return corrected_result

还有第二处自动纠错: 就算 server 名对,但调用后返回内容里出现字符串 "Unknown tool:",同样触发 _find_servers_with_tool,唯一命中就递归改派,见 src/tool/manager.py:456-467。两处都只在"候选唯一"时才敢自动改,把误判风险降到最低。

巧思 B:playwright 走持久会话,别每次都重开浏览器​

场景: 浏览器操作是有状态的——导航到某页、再截图、再点击,必须在同一个浏览器会话里连续做。若每次调用都起一个新 stdio 子进程,状态全丢。

做法: 对 server_name == "playwright" 特判:第一次调用时创建并 connect 一个 PlaywrightSession,之后复用这个持久会话,见 src/tool/manager.py:327-357。

# src/tool/manager.py:329-334 —— 浏览器会话只建一次,后续复用
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 自己维护 read/write/session 并手动管理 __aenter__/__aexit__ 生命周期(而不是用 async with 一次性开关),这正是它能"跨调用保活"的原因,见 src/tool/mcp_servers/browser_session.py:31-42(connect)与 :44-52(call_tool)。

巧思 C:scrape 失败,回退 MarkItDown​

场景: 网页抓取工具(scrape)对某些页面会报底层错误(如 "unhandled errors")。

做法: 捕获到这类异常时,不直接放弃,而是换一个引擎——用 MarkItDown 直接把 URL 转成文本再返回,见 src/tool/manager.py:485-510。

# src/tool/manager.py:485-504 —— scrape 报特定错误时,换 MarkItDown 兜底
if tool_name == "scrape" 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 {"server_name": server_name, "tool_name": tool_name, "result": result.text_content}

兜底本身若也失败,只记日志,最终仍返回原始错误(manager.py:505-517)——兜底是尽力而为,绝不制造新的崩溃。

巧思 D:屏蔽爬 HuggingFace 数据集(防作弊)​

场景: MiroFlow 常在 benchmark(如 GAIA)上跑,而这些题的答案往往托管在 HuggingFace 的 dataset/space 页面。若让 browsing agent 直接去爬那个页面,就等于偷看答案。

做法: 在 scrape 结果返回前做一次事后检查:若工具是 scrape 且 URL 指向 huggingface.co/datasets 或 huggingface.co/spaces,直接把结果替换成一句拒绝语,见 src/tool/manager.py:395-396(检查)、_should_block_hf_scraping:93-104、_is_huggingface_dataset_or_space_url:83-91。

# src/tool/manager.py:100-104 —— 只拦 scrape 工具 + HF 数据集/空间 URL
return (
    tool_name == "scrape"
    and arguments.get("url")
    and self._is_huggingface_dataset_or_space_url(arguments["url"])
)

4. 深入实现:MCP server 本体长什么样​

前面讲的都是"管理侧"。那被管理的工具进程本体怎么写?答案很短:用 fastmcp 的 FastMCP + @mcp.tool() 装饰器,一个普通 async 函数就是一个工具。

举一例(搜索 server):

# src/tool/mcp_servers/searching_mcp_server.py:42 —— 建一个具名 server
mcp = FastMCP("searching-mcp-server")

# :88-89 —— 一个 async 函数 + @mcp.tool() 就注册成一个工具
@mcp.tool()
async def google_search(q: str, gl: str = "us", hl: str = "en", num: int = 10, ...) -> str:
    """Perform google searches via Serper API and retrieve rich results. ..."""
    ...

# :176 —— 同一个 server 里可以有多个工具
@mcp.tool()
async def wiki_get_page_content(entity: str, first_sentences: int = 10) -> str:
    ...

进程末尾用 stdio 传输启动,和 §3.1 里 yaml 的 -m src.tool.mcp_servers.searching_mcp_server 对上:

# src/tool/mcp_servers/searching_mcp_server.py:701-702 —— 以 stdio 方式跑起来
if __name__ == "__main__":
    mcp.run(transport="stdio", show_banner=False)

几个关键点:

  • 函数的类型注解 + docstring 会被 FastMCP 自动转成 MCP 的 inputSchema 和 description——这正是 §3.2 里 list_tools 能拿到的东西。
  • 工具签名的默认值(如 gl="us")成为可选参数;docstring 里对每个参数的说明,最终会展示给 LLM。

各 MCP server 职责表:

server(FastMCP 名)主要工具职责文件
searching-mcp-servergoogle_search、wiki_get_page_content、search_wiki_revision、search_archived_webpage、scrape_website联网搜索:Google(Serper)、维基百科、网页存档、抓网页searching_mcp_server.py:42,88,176,285,449,681
e2b-python-interpretercreate_sandbox、run_command、run_python_code、upload_file_*、download_file_*在 E2B 沙箱里跑代码 / shell、传文件python_server.py:15,138,178,228,267,312,361
reading-mcp-serverread_file按 URI 读取本地/远程文件内容reading_mcp_server.py:21,27
reasoning-mcp-serverreasoning把难题转发给一个"思考型"LLM 单独推理reasoning_mcp_server.py:25,29
vision-mcp-servervisual_question_answering、visual_audio_youtube_analyzing图像问答(VQA)、YouTube 视频分析vision_mcp_server.py:37,287,357
audio-mcp-serveraudio_transcription、audio_question_answering音频转写、音频问答audio_mcp_server.py:31,132,203
(markitdown-mcp)——外部包提供的文档转 markdown server(见 tool-markitdown.yaml)config/tool/tool-markitdown.yaml

注:仓库里另有 *_os.py 变体(如 reasoning_mcp_server_os.py),是面向开源/本地模型的对应版本,由不同 yaml(如 tool-reasoning-os.yaml)挂载,机制同上。

5. 巧妙之处(可借鉴的技术)​

  • "失败也是数据":execute_tool_call 恒返回字典、get_all_tool_definitions 把连不上的 server 记成 {"error":...}——让 agent 能读到失败原因并自我修复,而不是抛异常炸掉主循环。manager.py:242-250、469-517。
  • 只在唯一候选时才自动纠错:server 名/工具名错配时,反查后恰好一个候选才敢自动改派并如实标注 [Auto-corrected],多个候选就交回给 LLM 决策——把自动化的误伤降到最低。manager.py:276-315、456-467。
  • 有状态工具用持久会话:浏览器操作跨调用保活,靠手动管理会话生命周期而非 async with。browser_session.py:31-52。
  • 同一入口做防作弊:把"禁止偷看 benchmark 答案"塞进工具执行的事后检查里,而非散落各处。manager.py:93-104、395-396。
  • 换引擎兜底:scrape 失败自动换 MarkItDown,兜底失败也只记日志、返回原错。manager.py:485-510。
  • 声明式工具池:一个 yaml = 一个工具进程,加载失败 continue 不致命;工具即插即用。tool_utils.py:21-58。

6. 边界与局限​

  • 纠错只覆盖"唯一命中":若两个 server 有同名工具且 LLM 写错了 server,不会自动改,只提示——避免猜错,但也把责任推回 LLM。manager.py:309-310。
  • playwright 会话不自动回收:browser_session 建了之后本类未在此处显式 close;PlaywrightSession.close 存在(browser_session.py:54-65)但何时被调用取决于上层生命周期管理。
  • MarkItDown 兜底只针对 scrape + 特定错误串:错误信息里必须含 "unhandled errors" 才触发,其它抓取失败不会走这条兜底。manager.py:485-490。
  • HF 屏蔽是启发式:仅按 URL 里是否含 huggingface.co/datasets|spaces 判断,且只拦 scrape 工具;换个抓取方式或改写 URL 理论上可绕过。manager.py:83-104。
  • 900 秒硬超时:超长的合法任务(大文件转写、长跑代码)可能被 with_timeout(900) 掐断。manager.py:254。

7. 横向对比​

工具层是本组文档的一环,建议配合阅读:

  • 工具从哪来、给谁调:main agent 的循环怎么触发 execute_tool_call → 01 编排主循环。
  • sub-agent 也被当成一个工具塞进这个池:expose_sub_agents_as_tools 的把戏 → 02 分层 sub-agent。
  • LLM 怎么从文本里解析出要调哪个工具(本章不碰的那一半)→ 04 LLM 抽象层。
  • 配置矩阵与全局容错哲学(本章多处兜底的总纲)→ 05 配置与容错。
  • 整体架构与阅读地图 → index。

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

主题文件路径符号名
配置 → server 池 + 黑名单src/utils/tool_utils.pycreate_mcp_server_parameters
sub-agent 伪装成工具src/utils/tool_utils.pyexpose_sub_agents_as_tools
工具总管src/tool/manager.pyToolManager
列举工具清单(过黑名单、失败不致命)src/tool/manager.pyget_all_tool_definitions
执行一次调用(超时 + 容错)src/tool/manager.pyexecute_tool_call
超时装饰器src/tool/manager.pywith_timeout
server 名反查纠错src/tool/manager.py_find_servers_with_tool
注入 TASK_ID 到子进程环境src/tool/manager.pyupdate_server_params_with_context_var
防作弊:拦 HF 数据集抓取src/tool/manager.py_should_block_hf_scraping / _is_huggingface_dataset_or_space_url
浏览器持久会话src/tool/mcp_servers/browser_session.pyPlaywrightSession
MCP server 本体范例src/tool/mcp_servers/searching_mcp_server.pymcp = FastMCP(...) / @mcp.tool() / google_search / wiki_get_page_content
代码沙箱 serversrc/tool/mcp_servers/python_server.pyrun_python_code / create_sandbox
视觉 serversrc/tool/mcp_servers/vision_mcp_server.pyvisual_question_answering
音频 serversrc/tool/mcp_servers/audio_mcp_server.pyaudio_transcription
推理转发 serversrc/tool/mcp_servers/reasoning_mcp_server.pyreasoning
工具配置样例config/tool/tool-searching.yaml(yaml: name / tool_command / args / env)