前言

给 Agent 配上 tool 才能干活——查时间、调 API、操作日历。你可以直接用 LangChain 的 @tool 装饰器写本地函数给 Agent 用, 但如果这个 tool 也想通过 API 暴露给外部客户端远程调用, 那就得上 MCP 了。

本文记录在一个实际项目中用 FastMCP 写 datetime 工具服务、再用 langchain-mcp-adapters 的 MultiServerMCPClient 接入 deepagents 的过程。

依赖精简：

fastmcp>=3.3.1                     # MCP server 框架
langchain-mcp-adapters>=0.2.2      # MCP -> LangChain tool 适配
deepagents>=0.4.12
fastapi>=0.135.2

为什么要把 tool 写成 MCP server, 而不是直接用 @tool 装饰器写几个 LangChain tool 完事？

关键区别在于调用方是谁。如果你的 tool 只给 Agent 自己用——比如一个内部格式化函数, 那写成 LangChain 的 @tool 就够了, 简单直接。但如果这个 tool 还想通过 HTTP API 暴露给外部客户端（比如另一个服务、一个前端页面、甚至 Postman 调试）, 那就得走 MCP——它本身就是标准协议, 任何支持 MCP 的客户端都能连上来用。

所以项目里的策略是：datetime 类工具做成 MCP server, 因为后续可能会给其他服务调用；而某些纯内部辅助工具就继续用 @tool。两条腿走路, 不搞一刀切。

用 FastMCP 写一个 MCP Server

FastMCP 是目前 Python 侧写 MCP server 最省事的框架了。一个带时间工具的 server 长这样：

from fastmcp import FastMCP
from fastmcp.server.middleware.error_handling import ErrorHandlingMiddleware
from mcp.types import ToolAnnotations

mcp = FastMCP(
    name="datetime_server",
    instructions="A server for datetime related operations",
)
mcp.add_middleware(
    ErrorHandlingMiddleware(
        include_traceback=True,
        transform_errors=True,
    )
)

@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
async def get_time_info(dt_str: str = "") -> BasicTimeInfo:
    """根据输入的时间字符串获取基本时间信息。未提供则默认使用当前UTC时间"""
    ...

@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
async def timezone_convert(utc_datetime: str, target_timezone: str) -> str:
    """将UTC时间转换为指定时区的时间"""
    ...

@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
async def datetime_diff(dt_str1: str, dt_str2: str) -> DatetimeDiff:
    """计算两个ISO格式日期时间字符串的时间差"""
    ...

几点值得说的：

• ToolAnnotations(readOnlyHint=True)：告诉调用方这些 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 调用时知道字段含义。

FastMCP 写好了 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": f"Bearer {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 Server

FastMCP 内置了 ASGI app 的生成方法, 直接丢给 FastAPI 的 mount() 就行。但这里有个细节：lifespan 管理。

FastMCP 有自己的 lifespan 逻辑（启动时注册 tool schema、注册 http handler 等）, FastAPI 也有自己的 lifespan（比如数据库连接池初始化等）。如果你直接 app.mount(), FastAPI 不会自动管子 app 的 lifespan, 启动顺序就可能出问题。

解决办法是用 fastmcp.utilities.lifespan.combine_lifespans：

from 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(lifespan=combine_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 = f"http://127.0.0.1:{cfg.SERVER_PORT}/api/mcp/datetime"

Agent 侧接入

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(
            model=self._llm,
            tools=self._tools,          # 包含 MCP tool + 本地 tool
            checkpointer=checkpointer,
            system_prompt="...",
            middleware=[
                ToolRetryMiddleware(
                    max_retries=2,
                    retry_on=(TimeoutException,),
                    on_failure="continue",
                )
            ],
        )

ToolRetryMiddleware 在这里有两个作用：

• max_retries=2：MCP tool 走的 HTTP, 网络抖动、server 临时不可用都可能发生, 设 2 次重试可以避免一次抖动就把整个流程打断。
• on_failure="continue"：重试用完还是失败了怎么办？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 stdio：streamable_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的配置可以放在配置文件里面，这样以后还能动态配置。