a-cdm × deer-flow 源码分析

主题

MCP 集成 ​

本章目标:

  • 讲清 DeerFlow 如何用 langchain-mcp-adapters 的 MultiServerMCPClient 统一管理多个 MCP 服务器,并把它们的工具暴露给 Lead Agent。
  • 解释「懒加载 + mtime 缓存失效」机制为何存在,以及它如何让 Gateway 进程改写配置后 LangGraph 运行时无需重启即可感知。
  • 拆解三种 transport(stdio / SSE / HTTP)与 OAuth(client_credentials / refresh_token)自动刷新 + Authorization 注入的完整链路。

TL;DR ​

DeerFlow 通过 deerflow.mcp 子包集成 Model Context Protocol:client.py 把 extensions_config.json 中每个启用的服务器翻译成 MultiServerMCPClient 参数,tools.py 创建客户端并一次性拉取所有工具,cache.py 以模块级全局变量缓存结果并按配置文件 mtime 判断是否过期,oauth.py 为 HTTP/SSE 服务器提供 token 自动获取、缓存、刷新与 Authorization 头注入。MCP 工具采用懒加载,首次调用 get_cached_mcp_tools() 时才连接服务器;之所以用 mtime 失效而非进程内事件,是因为 Gateway API 与 LangGraph 运行时各自读盘,mtime 是跨进程感知配置变更的最简可靠信号。

Overview ​

为什么 MCP 工具要懒加载 + mtime 缓存失效 ​

MCP 服务器是外部进程或远程 HTTP 端点。每次构建 Agent 都重新连接、握手、列举工具会带来明显延迟,因此结果必须缓存。但 DeerFlow 的运行架构带来一个特殊约束:Gateway API(端口 8001 的 REST 层)与执行 Agent 图的 LangGraph 运行时是逻辑上分离的读者——PUT /api/mcp/config 由 Gateway 处理并写入 extensions_config.json,而真正消费 MCP 工具的代码运行在 Agent 图里。进程内的「重置缓存」事件无法可靠跨越这条边界。

解决方案是把磁盘文件的修改时间(mtime)当作单一事实来源:缓存初始化时记录 extensions_config.json 的 mtime cache.py:76,每次取缓存前比较当前 mtime 是否更新 cache.py:49-51。一旦 Gateway 改写了文件,下次取工具时检测到 mtime 变大,自动重置并重新连接 cache.py:99-101。懒加载则保证首个真正需要工具的请求才付出连接成本,且能在 FastAPI 与 LangGraph Studio 两种事件循环上下文中都正确工作 cache.py:83-91。

Architecture ​

deerflow.mcp 是一个高内聚子包,对外只导出 6 个符号 __init__.py:7-14。四个内部模块各司其职:client.py 负责配置翻译,tools.py 负责装配客户端与拦截器,cache.py 负责生命周期与失效,oauth.py 负责令牌管理。调用入口是工具系统 tools/tools.py,它在 include_mcp=True 时从磁盘重读 ExtensionsConfig 并调用 get_cached_mcp_tools() tools/tools.py:119-128。

Source 列表:

Components / Subsystems ​

client.py — 配置翻译 ​

职责:把 McpServerConfig 翻译成 MultiServerMCPClient 能识别的参数字典,不做任何网络 I/O。

build_server_params() 按 type 字段分支:stdio 要求 command,组装 command/args/env client.py:24-31;sse 与 http 要求 url,组装 url/headers client.py:32-38;其它类型抛 ValueError client.py:39-40。build_servers_config() 遍历所有启用的服务器,逐个翻译;单个服务器配置失败只记错误并跳过,不影响其它服务器 client.py:60-68。

关键函数:build_server_params client.py:11、build_servers_config client.py:45。

tools.py — 客户端装配 ​

职责:创建 MultiServerMCPClient、注入 OAuth 初始头与拦截器、拉取所有工具并做同步包装。

get_mcp_tools() 总是用 ExtensionsConfig.from_file() 直接读盘以拿到 Gateway 的最新改动 tools.py:32。它先为 SSE/HTTP 服务器注入初始 Authorization 头(用于工具发现/会话初始化)tools.py:43-51,再装配 OAuth 拦截器与可选自定义拦截器 tools.py:53-77,然后以 tool_name_prefix=True 创建客户端并调用 client.get_tools() tools.py:79-83。最后对只有 coroutine 而无 func 的工具补一个同步包装,因为 DeerFlow 客户端同步流式消费 tools.py:86-88。若 langchain-mcp-adapters 未安装,直接返回空列表并告警 tools.py:23-26。

关键函数:get_mcp_tools tools.py:16。

cache.py — 生命周期与失效 ​

职责:用模块级全局变量缓存工具列表,提供懒加载、并发安全的初始化与 mtime 失效。

缓存状态由 4 个全局变量持有:_mcp_tools_cache、_cache_initialized、_initialization_lock、_config_mtime cache.py:11-14。initialize_mcp_tools() 在 asyncio.Lock 保护下只初始化一次,并记录配置 mtime cache.py:66-79。get_cached_mcp_tools() 是对外主入口,先判失效再判初始化,并能在「事件循环正在运行」时改用线程池中的新循环执行异步初始化 cache.py:106-118。reset_mcp_tools_cache() 清空三态供测试或强制重载 cache.py:133-142。

关键函数:get_cached_mcp_tools cache.py:82、_is_cache_stale cache.py:31、initialize_mcp_tools cache.py:56。

oauth.py — 令牌管理 ​

职责:为启用 OAuth 的 HTTP/SSE 服务器获取、缓存、刷新令牌,并以拦截器形式注入 Authorization 头。

OAuthTokenManager 按服务器名持有令牌与各自的 asyncio.Lock,从 ExtensionsConfig 中筛出 oauth.enabled 的服务器 oauth.py:33-39。get_authorization_header() 采用双重检查:先无锁判未过期则直接返回,否则加锁再判,仍过期才真正请求新令牌 oauth.py:47-65。_fetch_token() 按 grant_type 区分 client_credentials(需 client_id+client_secret)与 refresh_token(需 refresh_token),POST 到 token_url 并按可配置字段名解析响应 oauth.py:85-119。对外两个出口:build_oauth_tool_interceptor() 返回每次工具调用前覆盖 headers 的拦截器 oauth.py:122-137;get_initial_oauth_headers() 返回连接建立时用的初始头 oauth.py:140-150。

关键类:OAuthTokenManager oauth.py:25、_OAuthToken(dataclass)oauth.py:16-22。

Data Flow ​

首次取工具:懒加载 → 连接多 server → 加载工具 ​

OAuth 令牌刷新流程 ​

Implementation Details ​

mtime 失效的核心判定只有几行,却是整个跨进程一致性的关键 cache.py:37-53:

python
if not _cache_initialized:
    return False  # 尚未初始化,谈不上过期

current_mtime = _get_config_mtime()

# 取不到 mtime(文件不存在等)时保守认为未过期
if _config_mtime is None or current_mtime is None:
    return False

# 配置文件在缓存之后被改动过 -> 过期
if current_mtime > _config_mtime:
    logger.info("MCP config file has been modified ...")
    return True
return False

解读:三个早返回都偏向「不失效」,这是有意的保守策略——只有在「已初始化」且「两次都成功读到 mtime」且「当前 mtime 严格大于记录值」时才判过期。用严格 > 而非 !=,避免因时钟回拨或文件被反向触碰而误判。判过期后 get_cached_mcp_tools() 调 reset_mcp_tools_cache() 把三态清空,下一次进入即触发完整重连 cache.py:99-101。

OAuth 过期判定同样精炼——令牌在到期前 refresh_skew_seconds(默认 60s)即视为「临近过期」,提前刷新避免请求途中失效 oauth.py:67-70。

速查表 ​

Transport 类型 ​

type必填字段可选字段含义Source
stdio(默认)commandargs、env启动子进程作为 MCP 服务器client.py:24-31
sseurlheaders、oauthServer-Sent Events 远程端点client.py:32-38
httpurlheaders、oauthStreamable HTTP 远程端点client.py:32-38
其它——抛 ValueError,该 server 被跳过client.py:39-40

OAuth grant 矩阵 ​

grant_type必填字段选填字段适用Source
client_credentials(默认)client_id、client_secretscope、audience、extra_token_params机器对机器oauth.py:85-89
refresh_tokenrefresh_tokenclient_id、client_secret、scope用刷新令牌换访问令牌oauth.py:90-97
其它——抛 ValueErroroauth.py:98-99

令牌响应字段名均可配置:token_field、token_type_field、expires_in_field、default_token_type,以适配非标准 OAuth 提供方 oauth.py:106-118。

Configuration ​

mcpServers 中每个服务器的字段(McpServerConfig,本表为入口速览,完整配置层细节见 06 章):

字段类型默认说明Source
enabledbooltrue是否启用该服务器extensions_config.py:39
typestrstdiotransport:stdio/sse/httpextensions_config.py:40
commandstr?nullstdio 启动命令extensions_config.py:41
argslist[]stdio 命令参数extensions_config.py:42
envdict{}stdio 子进程环境变量 ⚠️ 含敏感凭据,勿提交明文extensions_config.py:43
urlstr?nullsse/http 服务器 URLextensions_config.py:44
headersdict{}sse/http 自定义请求头 ⚠️ 常含 API key/Bearer,视为机密;OAuth 注入的 Authorization 会覆盖此处同名键extensions_config.py:45
oauthobj?nullOAuth 配置(McpOAuthConfig)⚠️ client_secret/refresh_token 为高敏机密,建议用 $ENV 形式注入而非写死extensions_config.py:46
descriptionstr""人类可读描述extensions_config.py:47

oauth 子对象关键字段:token_url(必填)、grant_type、client_id、client_secret、refresh_token、scope、audience、refresh_skew_seconds(默认 60)extensions_config.py:13-33。安全提示:env、headers、oauth.client_secret、oauth.refresh_token 均可能泄漏凭据——extensions_config.json 进版本库前务必脱敏,优先用环境变量占位符。运行时改配置走 Gateway PUT /api/mcp/config,落盘后由 mtime 机制驱动 LangGraph 重连,无需重启进程。

Common Pitfalls / Tips ​

  • 改了配置工具没变化:mtime 失效靠「严格大于」判定。若编辑工具未改动 mtime,或文件系统 mtime 精度问题导致相等,缓存不会刷新。可调用 reset_mcp_tools_cache() 或确保通过 Gateway PUT 写入(它会真实更新 mtime)cache.py:49-53。
  • 未安装适配器静默无工具:缺少 langchain-mcp-adapters 时只 WARNING 并返回空列表,不报错。MCP 工具消失时先确认依赖已装 tools.py:23-26。
  • 单服务器故障不阻断其它:build_servers_config 对单个配置异常只记日志跳过;某 server 配错不会拖垮整批工具加载 client.py:62-67。
  • OAuth 仅对 HTTP/SSE 生效:oauth 字段对 stdio 无意义;初始头注入只对 transport 为 sse/http 的 server 应用 tools.py:48-51。
  • 工具名带前缀:客户端以 tool_name_prefix=True 创建,工具名会带 server 名前缀,避免多服务器同名工具冲突;编写依赖工具名的逻辑时需注意 tools.py:79。
  • 事件循环上下文:get_cached_mcp_tools() 在循环已运行时会把初始化丢进线程池新循环执行,失败时吞异常返回空列表;排查时看 Failed to lazy-initialize MCP tools 日志 cache.py:106-128。

References ​

页面关系
06-MCP与Skills扩展配置上游:extensions_config.json 完整结构、解析优先级与 Gateway 改写细节
20-工具系统与内置工具上游:get_available_tools() 如何把 MCP 工具与内置/社区工具合并装配进 Agent
13-Gateway-API与路由体系协作:PUT /api/mcp/config 写盘后触发本章 mtime 失效的另一进程入口

公司内部参考 · 由 claude-wiki-gen 基于源码自动生成的二次分析

a-cdm（公司内部）· deer-flow © ByteDance