前言给 Agent 配上 tool 才能干活——查时间、调 API、操作日历。你可以直接用 LangChain 的tool装饰器写本地函数给 Agent 用, 但如果这个 tool 也想通过 API 暴露给外部客户端远程调用, 那就得上 MCP 了。本文记录在一个实际项目中用 FastMCP 写 datetime 工具服务、再用langchain-mcp-adapters的MultiServerMCPClient接入 deepagents 的过程。依赖精简fastmcp3.3.1 # MCP server 框架 langchain-mcp-adapters0.2.2 # MCP - LangChain tool 适配 deepagents0.4.12 fastapi0.135.2MCP tool vs Agent tool为什么要把 tool 写成 MCP server, 而不是直接用tool装饰器写几个 LangChain tool 完事关键区别在于调用方是谁。如果你的 tool 只给 Agent 自己用——比如一个内部格式化函数, 那写成 LangChain 的tool就够了, 简单直接。但如果这个 tool 还想通过 HTTP API 暴露给外部客户端比如另一个服务、一个前端页面、甚至 Postman 调试, 那就得走 MCP——它本身就是标准协议, 任何支持 MCP 的客户端都能连上来用。所以项目里的策略是datetime 类工具做成 MCP server, 因为后续可能会给其他服务调用而某些纯内部辅助工具就继续用tool。两条腿走路, 不搞一刀切。用 FastMCP 写一个 MCP ServerFastMCP 是目前 Python 侧写 MCP server 最省事的框架了。一个带时间工具的 server 长这样from fastmcp import FastMCP from fastmcp.server.middleware.error_handling import ErrorHandlingMiddleware from mcp.types import ToolAnnotations mcp FastMCP( namedatetime_server, instructionsA server for datetime related operations, ) mcp.add_middleware( ErrorHandlingMiddleware( include_tracebackTrue, transform_errorsTrue, ) ) mcp.tool(annotationsToolAnnotations(readOnlyHintTrue)) async def get_time_info(dt_str: str ) - BasicTimeInfo: 根据输入的时间字符串获取基本时间信息。未提供则默认使用当前UTC时间 ... mcp.tool(annotationsToolAnnotations(readOnlyHintTrue)) async def timezone_convert(utc_datetime: str, target_timezone: str) - str: 将UTC时间转换为指定时区的时间 ... mcp.tool(annotationsToolAnnotations(readOnlyHintTrue)) async def datetime_diff(dt_str1: str, dt_str2: str) - DatetimeDiff: 计算两个ISO格式日期时间字符串的时间差 ...几点值得说的ToolAnnotations(readOnlyHintTrue)告诉调用方这些 tool 是只读的, 不会产生副作用。对 Agent 的推理有帮助, LLM 可以放心调用。如果你的工具会写数据比如创建日历事件, 就别加这个 annotation。ErrorHandlingMiddleware不然 tool 报错了 MCP server 给你返回一个干巴巴的 error code, 根本不知道哪出了问题。开了include_traceback和transform_errors之后, 错误信息会带着 traceback 返回给 client, 排查方便很多。Pydantic 模型做返回值BasicTimeInfo、DatetimeDiff这些用 Pydantic 定义好的 model 作为 tool 返回值, 既有类型安全, 又在 MCP 协议层自动生成了 schema 描述, Agent 调用时知道字段含义。MultiServerMCPClient从 MCP 加载 toolFastMCP 写好了 server, 怎么让 deepagents 能用上这些 tool用langchain-mcp-adapters提供的MultiServerMCPClient。它本质上是一个多 server 的客户端管理器——你告诉它每个 MCP server 的地址和传输方式, 它帮你建立连接、发现工具、转换成 LangChain 标准的BaseTool列表from langchain_mcp_adapters.client import MultiServerMCPClient client MultiServerMCPClient( { datetime: { transport: streamable_http, url: http://127.0.0.1:3002/api/mcp/datetime, } } ) tools await client.get_tools() # tools 就是 list[BaseTool], 可以直接喂给 create_deep_agent()transport目前主流就两种streamable_http适合网络调用、跨进程共享和stdio适合本地子进程、单机部署。本项目用的是前者, 因为 MCP server 是作为 FastAPI sub-app 挂载的, 不依赖进程间管道。如果是需要鉴权的远程 MCP server, 加headers就行{ didatick: { transport: streamable_http, url: https://mcp.dida365.com, headers: { Authorization: fBearer {cfg.DIDA_TOKEN}, }, } }为什么单独搞了个 MCPServers 类项目里没有把MultiServerMCPClient直接丢在AIAgent里, 而是单独抽了个MCPServers出来class MCPServers: def __init__(self): self._client_datetime: MultiServerMCPClient None self._datetime_tools: list[BaseTool] [] self._client_didatick: MultiServerMCPClient None self._didatick_tools: list[BaseTool] [] async def get_datetime_tools(self) - list[BaseTool]: if not self._datetime_tools: await self._init_datetime_mcp() return self._datetime_tools async def get_didatick_tools(self) - list[BaseTool]: if not self._didatick_tools: await self._init_aididatick_mcp() return self._didatick_tools这么拆的理由很简单SubAgent 可能只需要部分 MCP tool。如果你的应用里有多个 SubAgent——比如一个只负责查日历, 一个只负责查时间——让每个 SubAgent 只拿到它需要的 tool 列表, 可以避免 tool 过多导致 LLM 选择困难所谓的 tool overload 问题。每个 MCP server 的 tool 按get_xxx_tools()独立暴露, 上层可以自由组合class AIAgent: def __init__(self): self._mcp_servers MCPServers() async def _init_tools(self): # 这个 Agent 需要所有 tool tools_mcp_datetime await self._mcp_servers.get_datetime_tools() tools_mcp_didatick await self._mcp_servers.get_didatick_tools() self._tools.extend(tools_mcp_datetime) self._tools.extend(tools_mcp_didatick) # 另一个 SubAgent 可能只拿 datetime: # tools await self._mcp_servers.get_datetime_tools()另外每个方法的懒加载 缓存逻辑保证了同一个 MCP server 只连一次。FastAPI 如何 mount FastMCP ServerFastMCP 内置了 ASGI app 的生成方法, 直接丢给 FastAPI 的mount()就行。但这里有个细节lifespan 管理。FastMCP 有自己的 lifespan 逻辑启动时注册 tool schema、注册 http handler 等, FastAPI 也有自己的 lifespan比如数据库连接池初始化等。如果你直接app.mount(), FastAPI 不会自动管子 app 的 lifespan, 启动顺序就可能出问题。解决办法是用fastmcp.utilities.lifespan.combine_lifespansfrom fastmcp.utilities.lifespan import combine_lifespans def create_app() - FastAPI: # FastMCP 实例生成 ASGI app datetime_mcp_app datetime_mcp.http_app(path/datetime) # 合并两个 lifespan, 保证初始化顺序 app FastAPI(lifespancombine_lifespans(lifespan, datetime_mcp_app.lifespan)) # 挂载到 /api/mcp 路径下 app.mount(/api/mcp, datetime_mcp_app) return app实际访问路径就变成了/api/mcp/datetime——app.mount的/api/mcp是前缀,http_app(path/datetime)的/datetime是子路径。如果你的项目里有多个 MCP server 要挂载, 每个都调http_app(path...)然后逐个 mount, 别忘了把它们的 lifespan 都塞到combine_lifespans里。最后, MCP client 连的就是本进程内的 server, 所以 endpoint 用127.0.0.1就行endpoint fhttp://127.0.0.1:{cfg.SERVER_PORT}/api/mcp/datetimeAgent 侧接入create_deep_agent()接收 tool 列表, 不管这些 tool 是本地tool函数还是从 MCP 加载的, 对它来说都一样class AIAgent: async def _init_deep_agent(self): if self._agent: return if not self._tools: await self._init_tools() self._agent create_deep_agent( modelself._llm, toolsself._tools, # 包含 MCP tool 本地 tool checkpointercheckpointer, system_prompt..., middleware[ ToolRetryMiddleware( max_retries2, retry_on(TimeoutException,), on_failurecontinue, ) ], )ToolRetryMiddleware在这里有两个作用max_retries2MCP tool 走的 HTTP, 网络抖动、server 临时不可用都可能发生, 设 2 次重试可以避免一次抖动就把整个流程打断。on_failurecontinue重试用完还是失败了怎么办continue表示不中断 Agent 执行, 而是把失败信息交给 LLM。比如 AI 传错了参数导致 tool 报ValidationError,ToolRetryMiddleware把这个异常信息原样塞回给 LLM, LLM 看到之后可以自己修正参数重新调用。这比直接崩掉要实用得多。当然你也可以设成raise让异常直接抛出去, 看你业务需要。改进点MCP server 独立部署目前datetime_mcp作为 FastAPI sub-app 跟 Agent 跑在同一个进程里。如果 server 本身很重比如需要 GPU 推理的视觉工具, 应该拆出去独立部署, client 用远程 HTTP 连接。多 server 的 tool 冲突如果两个 MCP server 刚好提供了同名的 tool,MultiServerMCPClient.get_tools()的行为取决于实现——可能会覆盖、也可能报错。最好在设计时就给每个 server 的 tool 加上有意义的前缀或命名空间。streamable_http vs stdiostreamable_http方便调试你可以直接用 curl 打 MCP server。stdio有个隐藏的坑——如果每次 Agent 请求都创建新的 MCP 连接用MultiServerMCPClient的connect()/disconnect()模式, 后台会残留一堆 MCP server 子进程, 杀都杀不干净。而如果提前建好长连接复用, 又得自己管生命周期。回到本项目——Agent 调的是自己进程内的 MCP API, HTTP 走127.0.0.1也就多一层本地网络栈的 overhead, 实际延迟几乎感觉不到。除非你的 tool 调用量巨大到本地 HTTP 成为瓶颈, 否则streamable_http完全够用。鉴权远程 MCP server 目前靠 header 传 Bearer token。生产环境如果有 Token 过期策略最好加上 token 过期刷新和重试逻辑, 不要在连接初始化的时候因为 401 就直接挂了。MCP Server的配置可以放在配置文件里面这样以后还能动态配置。