FastMCP × LangChain / LangGraph 结合

核心组件：langchain‑mcp‑adapters，LangChain官方适配器，把MCP Server（FastMCP写的）的tools自动转换成LangChain BaseTool，直接喂给LangGraph Agent，不用手动写wrapper、不用手写function‑call schema。

两种集成模式：

1. 模式A：LangGraph作为Host（客户端）调用FastMCP服务（最常用）
   • FastMCP = 独立MCP服务进程（stdio / SSE‑HTTP），暴露工具；
   • LangGraph Agent 通过 MultiServerMCPClient 远程发现并调用MCP工具；
   • 优势：工具与Agent解耦，可多Agent共享同一套MCP服务，可给Claude Desktop同时使用。
2. 模式B：FastMCP服务内部嵌入LangGraph
   • FastMCP的tool内部直接调用LangGraph图实例；MCP对外暴露Agent能力；
   • 外部客户端（Claude Desktop等）调用MCP，实际执行LangGraph智能体。

依赖安装

pip install fastmcp langgraph langchain-openai langchain-mcp-adapters python‑dotenv

模式A：LangGraph Agent调用FastMCP服务（完整可运行）

第一步：编写 FastMCP Server mcp_server.py

# mcp_server.py
from fastmcp import FastMCP

mcp = FastMCP("demo‑mcp‑service")

@mcp.tool()
def add(a:int, b:int) -> int:
    """两个数字相加
    Args:
        a:数字1
        b:数字2
    """
    return a + b

@mcp.tool()
def multiply(a:int, b:int) -> int:
    """两个数字相乘"""
    return a * b

if __name__ == "__main__":
    # stdio模式，子进程调用；不要直接运行这个脚本，由MultiServerMCPClient拉起
    mcp.run(transport="stdio")

第二步：LangGraph Agent 客户端 langgraph_mcp_agent.py

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

async def main():
    # MultiServerMCPClient：可以同时连接多个FastMCP服务
    async with MultiServerMCPClient({
        "math_service": {
            "command": "python",
            "args": ["mcp_server.py"],
            "transport": "stdio",
        }
    }) as client:
        # 自动从MCP服务发现全部工具，转为LangChain BaseTool列表
        tools = await client.get_tools()

        llm = ChatOpenAI(model="gpt‑4o‑mini", api_key="xxx")
        # 开箱即用 ReAct Agent
        agent = create_react_agent(llm, tools)

        resp = await agent.ainvoke({"messages": "计算 (10 + 20) * 5"})
        for msg in resp["messages"]:
            print(f"{msg.type}: {msg.content}")

if __name__ == "__main__":
    asyncio.run(main())

运行：python langgraph_mcp_agent.py，MultiServerMCPClient会自动拉起mcp_server.py子进程，走stdio协议通信。

远程SSE/Streamable‑HTTP模式（FastMCP服务单独部署）

先启动FastMCP为HTTP服务：

# mcp_server.py
if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8001)

LangGraph客户端连接远程地址：

async with MultiServerMCPClient({
    "math_service": {
        "url": "http://127.0.0.1:8001/mcp",
        "transport": "streamable_http"
    }
}) as client:
    tools = await client.get_tools()

手动搭建StateGraph（不使用create_react_agent）

自定义节点，完全控制图逻辑，适合复杂业务：

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition

async def main():
    async with MultiServerMCPClient({
        "math": {"command":"python","args":["mcp_server.py"],"transport":"stdio"}
    }) as client:
        tools = await client.get_tools()
        llm = ChatOpenAI(model="gpt‑4o‑mini", api_key="xxx").bind_tools(tools)

        def agent_node(state: MessagesState):
            resp = llm.invoke(state["messages"])
            return {"messages":[resp]}

        builder = StateGraph(MessagesState)
        builder.add_node("agent", agent_node)
        builder.add_node("tools", ToolNode(tools))

        builder.add_edge(START, "agent")
        builder.add_conditional_edges("agent", tools_condition)
        builder.add_edge("tools", "agent")

        graph = builder.compile()
        res = await graph.ainvoke({"messages":[{"role":"user","content":"(7+3)*9"}]})
        print(res["messages"][-1].content)

asyncio.run(main())

模式B：FastMCP Server内部封装LangGraph Agent

场景：外部客户端（Claude Desktop、其他Agent）通过MCP协议调用你的LangGraph智能体。

from fastmcp import FastMCP
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

mcp = FastMCP("agent‑gateway")

# 内部实例化LangGraph Agent
llm = ChatOpenAI(model="gpt‑4o‑mini", api_key="xxx")
agent = create_react_agent(llm, [])

@mcp.tool()
async def run_agent(query:str) -> str:
    """执行LangGraph智能体，处理用户问题
    Args:
        query: 用户提问
    """
    result = await agent.ainvoke({"messages": [{"role":"user","content":query}]})
    return result["messages"][-1].content

if __name__ == "__main__":
    mcp.run(transport="stdio")

此时 Claude Desktop / 任意MCP Client调用run_agent工具，内部跑LangGraph。

关键坑点与最佳实践

1. 全部是异步：MultiServerMCPClient、get_tools()、agent调用都必须async/await，不要同步调用。
2. stdio模式：FastMCP服务中禁止print，stdout是MCP协议报文；日志用logging模块输出stderr。
3. 会话状态：
   • stdio模式：MCP服务无会话，每次工具调用是独立session；
   • streamable‑http(SSE)模式才有session_id，可在FastMCP的Context拿会话信息保存状态。
4. 多MCP服务：MultiServerMCPClient支持配置多个不同MCP服务，自动合并全部tools。
5. 资源释放：必须用async with MultiServerMCPClient()上下文管理器，自动关闭子进程/连接；否则会产生僵尸进程。
6. 版本：fastmcp>=2.x，langchain‑mcp‑adapters保持最新，老版本有兼容性bug。

架构图

【用户】
    ↓
LangGraph Agent（Host）
    ↓ langchain‑mcp‑adapters
MultiServerMCPClient
    ↓ MCP‑JSONRPC2
FastMCP Server 进程1（math工具）
FastMCP Server 进程2（数据库工具）

拓展：把LangChain工具暴露成MCP服务

FastMCP提供to_fastmcp_tool()，可以把LangChain BaseTool直接包装成MCP工具，不用重写：

from fastmcp import FastMCP
from fastmcp.langchain import to_fastmcp_tool
from langchain_community.tools import WikipediaQueryRun

mcp = FastMCP()
wiki_tool = WikipediaQueryRun()
mcp.add_tool(to_fastmcp_tool(wiki_tool))