从协议握手到LangGraph多Server调用：MCP Agent工具集成实践

为什么需要 MCP

AI Agent 落地时最常见的需求不是换模型，而是让模型安全、稳定地访问外部能力：查询数据库、读取文件、调用业务 API、执行计算、搜索知识库。如果没有协议约定，每个模型框架定义一套工具格式，每个业务系统写一套适配代码，工具参数、错误、权限、日志和传输方式各不相同，Agent 跨平台迁移时大量集成代码需要重写。

Model Context Protocol（MCP）把这部分集成标准化：能力提供方抽象为 Server，能力使用方抽象为 Client，用 JSON-RPC 和明确的初始化流程统一工具、资源、提示词、生命周期和传输。它的价值不是让模型更聪明，而是让工具调用从私有约定变成可发现、可复用、可测试的协议能力。

本文代码固定使用 Python 3.11 至 3.13、mcp==2.2.0、LangChain 1.x、LangGraph 1.x、Pydantic 2.x。需要特别注意：MCP Python SDK 2.x 已将旧版 FastMCP 改名为 MCPServer，导入路径为 from mcp.server.mcpserver import MCPServer；SDK 1.x 常见导入是 from mcp.server.fastmcp import FastMCP，两者不能混用。

三个角色与握手流程

Host 是用户真正使用的应用，例如 IDE、聊天客户端、桌面 Agent 或企业工作台，负责模型、会话、用户权限和交互体验。Client 由 Host 创建，负责连接一个 MCP Server，管理协议握手、请求发送、响应解析、能力协商和连接关闭，Python 中核心对象通常是 ClientSession。Server 暴露 Tools、Resources、Prompts 三类能力，并参与协议初始化、能力声明、生命周期管理和传输协商。

最小 Server 只需元数据即可运行：

from mcp.server.mcpserver import MCPServer

server = MCPServer(
    name="server-client-demo",
    title="MCP Server 与 Client 调用",
    description="演示初始化握手、能力协商和安全关闭。",
    instructions="演示初始化握手、能力协商和 ping",
    version="1.0.0",
)

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

客户端通过 stdio 启动 Server 并初始化会话：

async with stdio_client(server_params) as (read_stream, write_stream):
    async with ClientSession(read_stream, write_stream) as session:
        initialized = await session.initialize()
        print(initialized.server_info.name)
        print(initialized.protocol_version)
        print(session.instructions)
        await session.send_ping()

初始化阶段完成三件事：Client 声明自身协议版本和能力；Server 返回自身信息、协议版本和能力；双方在共同支持范围内建立会话。这一步决定了后续能否调用工具、读取资源以及使用扩展能力。

一个常见错误是把调试日志直接打印到 stdout。对于 stdio 传输，stdout 是 JSON-RPC 数据通道，任何额外输出都可能破坏协议流，调试信息应写 stderr。

Tools：让模型调用外部能力

Tools 是使用频率最高的能力，承载可变的业务操作。SDK 提供装饰器注册方式，自动完成参数校验和工具描述生成：

@server.tool()
def add(a: int, b: int) -> int:
    """返回两个整数之和。"""
    return a + b

函数名、类型标注和 docstring 会被转换成工具名称、输入 Schema 和描述。客户端先获取工具列表，再调用：

result = await session.list_tools()
for tool in result.tools:
    print(tool.name, tool.description)

result = await session.call_tool("add", {"a": 20, "b": 22})
text = "".join(
    item.text for item in result.content
    if getattr(item, "type", None) == "text"
)

工具可以是异步函数，并通过 Context 上报进度：

@server.tool()
async def slow_echo(text: str, ctx: Context, delay: float = 0.05) -> str:
    await ctx.report_progress(1, 2, "开始")
    await asyncio.sleep(delay)
    await ctx.report_progress(2, 2, "完成")
    return text.upper()

客户端可传入 progress_callback 接收进度。但进度更新不是结果本身，真正的业务结果仍应通过工具返回值表达，进度只用于观测。

错误分两类：预期业务错误（除零、记录不存在、状态不允许）应抛出 ToolError，客户端收到 is_error=True 的结果，模型可读取错误内容决定下一步；非预期异常（数据库断开、代码缺陷、依赖故障）不应伪装成业务结果，服务端记录堆栈，客户端只获得受控错误信息。

结构化输出、Resources 与 Prompts

传统工具调用常返回 JSON 字符串，调用方还需自行解析校验。MCP 支持结构化输出，服务端可直接声明 Pydantic 返回模型：

class ReportDetails(BaseModel):
    owner: str
    reviewed: bool
    notes: list[str] = Field(default_factory=list)

class Report(BaseModel):
    title: str
    score: float
    tags: list[str]
    details: ReportDetails

@server.tool(structured_output=True)
def build_report(topic: str, score: float) -> Report:
    return Report(
        title=f"{topic} Report", score=score,
        tags=["python", "mcp"],
        details=ReportDetails(owner="Ada", reviewed=True,
                              notes=["schema validated", "nested object"]),
    )

客户端通过 result.structured_content 读取。协议原生字段为 structuredContent，Python SDK 2.2.0 已自动适配为蛇形命名，无需手动转换。结构化输出的优势是数据自动校验、嵌套结构适配、类型规范统一，减少对模型文本格式自律的依赖。

Resources 强调读取内容，支持静态 URI 和 URI 模板：

@server.resource("demo://config", name="config",
                 mime_type="application/json")
def config_resource() -> dict[str, object]:
    return {"server": "examples", "features": ["tools", "resources", "prompts"]}

@server.resource("demo://users/{user_id}", name="user_profile",
                 mime_type="application/json")
def user_profile(user_id: str) -> UserProfile:
    return UserProfile(user_id=user_id, name="Ada Lovelace", role="admin")

客户端可 list_resources、list_resource_templates、read_resource。Resources 适合项目文档、配置、数据源 Schema、用户资料、可版本化知识片段。读取配置更适合 Resource，执行数据库修改更适合 Tool，两者不应互相替代。

Prompts 是 Server 提供的提示词模板，帮助 Host 统一提示结构、参数和消息序列：

@server.prompt(name="explain_code",
               description="根据编程语言和源码生成代码解释任务。")
def explain_code(language: str, code: str) -> list[UserMessage]:
    return [UserMessage(content=(
        f"请解释下面这段 {language} 代码。\n"
        f"```{language.lower()}\n{code}\n```"
    ))]

客户端先 list_prompts，再按参数 get_prompt 渲染。Prompt 提供的是消息模板，不是自动执行流程，是否使用、如何组合、是否发送给模型仍由 Host 决定。

Lifespan 与 LangGraph 接入

真实 Server 需要初始化连接池、加载模型、读取配置或创建缓存，这些工作应放入 lifespan：

@dataclass
class AppState:
    values: list[str] = field(default_factory=list)
    started: bool = False
    closed: bool = False

@asynccontextmanager
async def lifespan(server: MCPServer[AppState]) -> AsyncIterator[AppState]:
    state = AppState(started=True)
    try:
        yield state
    finally:
        state.closed = True

server = MCPServer(name="lifespan-demo", version="1.0.0", lifespan=lifespan)

工具中通过 ctx.request_context.lifespan_context 访问状态。适合放入生命周期的有数据库连接池、HTTP 客户端、全局缓存、模型句柄、启动校验、关闭日志；不适合的是单次请求临时变量、用户独立会话、不适合全局共享的可变数据。

接入 LangChain 与 LangGraph 需要一层适配。目前 langchain-mcp-adapters 0.3.1 仅支持 MCP 1.x，无法兼容 2.2.0 新版接口，因此可手动实现轻量桥接 MCPToolkit：

class MCPToolkit:
    async def get_tools(self) -> list[StructuredTool]:
        tools: list[StructuredTool] = []
        for server_name, connection in self.connections.items():
            async with self._open(connection) as session:
                result = await session.list_tools()
                tools.extend(
                    self._convert(server_name, connection, mcp_tool)
                    for mcp_tool in result.tools
                )
        return tools

每个 MCP 工具转换为 LangChain StructuredTool，调用时打开会话、执行 call_tool，is_error 为真则抛 ToolException。Agent 使用 LangChain 当前推荐入口：

from langchain.agents import create_agent

tools = await MCPToolkit(connections).get_tools()
agent = create_agent(model=model, tools=tools)
result = await agent.ainvoke(
    {"messages": [HumanMessage(content="Use multiply to calculate 6 * 7.")]}
)

LangGraph 1.x 已逐步废弃 create_react_agent，建议优先使用 create_agent。示例支持双模式调试：离线 Fake 模型用于单元测试与 CI，配置 OPENAI_API_KEY 后切换真实 ChatOpenAI，便于分层定位协议连接、工具转换和 Agent 执行问题。

多 Server 聚合与传输选型

企业级 Agent 项目通常按业务维度拆分独立服务：数据库、文件处理、业务 API、搜索等，权限边界更清晰、可独立部署扩容、通用能力可复用、单服务故障影响可控。多 Server 示例同时配置两个 stdio Server：

connections = {
    "math": StdioConnection(command=sys.executable,
                            args=[str(ROOT / "math_server.py")], cwd=str(ROOT)),
    "text": StdioConnection(command=sys.executable,
                            args=[str(ROOT / "text_server.py")], cwd=str(ROOT)),
}
tools = await MCPToolkit(connections).get_tools()
agent = create_agent(model=model, tools=tools)

多个 Server 暴露同名工具时需制定命名策略，例如 math_add、text_add，也可通过命名空间、工具白名单管控冲突。同时要关注服务启动超时、单服务故障降级、权限隔离、链路追踪、并发限流、多服务结果聚合，建议搭配简易路由治理层。

传输模式选择不能只看能否连上，还要看部署边界、生命周期、认证和可观测性。stdio 由 Client 作为子进程启动，通过标准输入输出通信，适合本地 CLI、IDE 插件、桌面 Agent、单用户工具进程，部署简单、进程边界明确，但难以跨网络复用，服务与客户端生命周期绑定。SSE 使用长连接接收服务端事件、通过 HTTP 发送消息，适合老旧项目兼容和需要服务端主动推送的场景；新项目如需 HTTP 类传输，可优先考虑 Streamable HTTP。Streamable HTTP 适合远程 Server、多客户端复用、与现有 HTTP 基础设施集成、需要网关、负载均衡和统一鉴权的场景。

无论哪种传输，都建议统一处理连接超时、重试和退避、会话关闭、空闲连接回收、认证授权、请求日志和链路追踪。

生产落地参考清单

能力边界上，Tool 负责执行业务操作、产生数据变更；Resource 承载只读上下文、可缓存数据；Prompt 负责通用模板复用，避免混用。输入输出上，参数命名语义清晰，用枚举约束固定状态值，复杂场景优先结构化输出，返回值尽量版本兼容，报错信息兼顾可读性、可追溯性与安全性。异常处理上，可预判的业务异常通过 ToolError 返回给模型，非预期异常在服务端记录堆栈、对外只暴露受控信息。

多 Server 场景还需补齐启动超时、故障降级、权限隔离、链路追踪和并发限流。这些工程细节决定了 MCP 集成能否从 Demo 走向生产可用。