LangGraph实战SSE协议打通远程MCP Server的完整指南最近在构建一个需要集成多种外部工具的智能体时我遇到了一个典型问题如何让运行在云端的LangGraph智能体安全、高效地调用部署在内网环境或另一台服务器上的专用工具传统的本地进程调用stdio方式显然行不通而简单的HTTP API调用又缺乏统一的工具发现和描述机制。这时MCPModel Context Protocol配合SSEServer-Sent Events协议的组合为我提供了一个优雅的解决方案。这篇文章我将从一个实践者的角度分享如何一步步搭建一个支持SSE远程调用的MCP Server并将其无缝集成到LangGraph构建的智能体中。整个过程不仅涉及代码实现更会深入探讨配置细节、调试技巧以及在实际项目中可能遇到的“坑”。无论你是希望扩展智能体能力的中级开发者还是对MCP协议远程调用感兴趣的探索者这篇指南都将提供可直接复用的代码和清晰的思路。1. 理解核心为什么是MCP与SSE在深入代码之前我们有必要厘清几个核心概念以及它们组合在一起所解决的痛点。MCPModel Context Protocol本质上是一套标准化的“通信协议”。你可以把它想象成智能体世界的“USB接口”标准。它定义了工具Tools应该如何被描述、发现和调用。一个遵循MCP协议的Server会向Client比如你的LangGraph智能体宣告“我这里有哪些工具可用每个工具叫什么名字需要什么参数返回什么格式。” Client无需关心工具背后的实现是Python函数、Shell脚本还是远程API它只需要按照MCP协议规定的格式去请求即可。这极大地解耦了工具提供方和工具使用方。那么通信方式有哪些MCP支持多种传输层协议最常见的有两种stdio标准输入/输出适用于Client和Server在同一台机器、同一进程或能直接启动子进程的场景。它简单直接但缺乏网络能力。SSEServer-Sent Events这是一种基于HTTP的服务器推送技术。它允许Server主动向Client发送事件流。在MCP语境下Client通过一个HTTP端点连接到Server双方通过这个长连接双向交换JSON格式的消息。提示SSE常被与WebSocket混淆。两者都支持全双工通信但SSE是单向通道Server - Client而MCP over SSE的实现通常利用了两个独立的SSE连接来模拟双向通信这在一些库如mcp中已封装好。当你的工具MCP Server需要部署在独立的服务器、容器或者位于防火墙后、与智能体运行环境隔离时SSE协议就成了连接它们的桥梁。它基于HTTP易于穿越常见的网络设施也便于做认证和负载均衡。下表对比了两种传输方式的典型场景特性stdio 传输SSE 传输部署位置同一主机/进程跨网络可远程部署网络要求无需要HTTP网络可达安全性进程间隔离依赖HTTPS、Token等网络层安全适用场景本地脚本、快速原型微服务架构、生产环境、工具服务化连接管理进程生命周期绑定独立的长连接可重连理解了“为什么”接下来我们就进入“怎么做”的阶段。我们将从零开始构建两个MCP Server一个通过SSE暴露另一个本地运行最后在LangGraph中统一调用。2. 构建你的第一个SSE MCP Server我们将构建一个简单的“利息计算器”MCP Server并通过SSE协议将其暴露为远程服务。这里选择Starlette作为Web框架因为它轻量且对异步支持友好与MCP的异步特性很匹配。首先确保你的环境已安装必要的包pip install mcp starlette uvicorn接下来是服务器核心代码interest_sse_server.pyimport logging import uvicorn from mcp.server.fastmcp import FastMCP from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.requests import Request from starlette.routing import Route, Mount # 配置日志方便调试 logging.basicConfig( format%(asctime)s - %(name)s - %(levelname)s - %(message)s, levellogging.INFO ) logger logging.getLogger(__name__) # 初始化FastMCP应用并为其命名 mcp FastMCP(InterestCalculator) # 定义第一个工具计算单利 mcp.tool() def yearly_simple_interest(principal: float, rate: float) - float: 根据本金和年利率计算单利。 公式利息 本金 * 利率 / 100 logger.info(f计算单利: 本金{principal}, 利率{rate}%) interest principal * rate / 100.0 return round(interest, 2) # 定义第二个工具计算复利 mcp.tool() def yearly_compound_interest(principal: float, rate: float) - float: 根据本金和年利率计算一年期复利。 公式本息和 本金 * (1 利率 / 100) logger.info(f计算复利: 本金{principal}, 利率{rate}%) amount principal * (1 rate / 100.0) interest amount - principal return round(interest, 2) # SSE连接处理函数 async def handle_sse(request: Request): 处理客户端发起的SSE连接请求。 这里建立了Server到Client的事件流通道。 sse_transport SseServerTransport(/messages/) async with sse_transport.connect_sse( request.scope, request.receive, request._send ) as (read_stream, write_stream): # 在此连接上运行MCP Server await mcp._mcp_server.run( read_stream, write_stream, mcp._mcp_server.create_initialization_options() ) if __name__ __main__: # 创建Starlette应用定义路由 # /sse 端点用于客户端初始连接建立SSE流 # /messages/ 端点用于客户端向服务器发送消息 app Starlette( routes[ Route(/sse, endpointhandle_sse), Mount(/messages/, appSseServerTransport(/messages/).handle_post_message), ] ) # 启动服务器 host 0.0.0.0 # 监听所有网络接口 port 8000 logger.info(f利息计算器 MCP Server 启动于 http://{host}:{port}) uvicorn.run(app, hosthost, portport)关键点解析FastMCP这是mcp库提供的高层封装让你能用装饰器快速定义工具无需手动处理底层协议消息。SseServerTransportMCP库提供的SSE传输层实现。它负责将MCP的JSON-RPC消息封装到SSE事件流中。双端点设计GET /sse客户端通过此端点建立SSE长连接用于接收服务器推送的消息如工具调用结果。POST /messages/客户端通过此端点向服务器发送消息如调用工具的请求。SseServerTransport.handle_post_message方法会处理这些请求。运行使用uvicorn作为ASGI服务器启动应用。确保你的防火墙或安全组允许对指定端口如8000的访问。启动服务器python interest_sse_server.py如果一切正常日志会显示服务器已启动。此时一个支持SSE远程调用的MCP Server就已经在运行了。3. 补充一个本地Stdio MCP Server作为对照为了展示LangGraph如何同时管理不同传输方式的工具我们再创建一个简单的本地MCP Server。这个服务器提供一个执行Shell命令的工具使用stdio传输。创建文件shell_mcp_server.pyimport subprocess import logging from mcp.server.fastmcp import FastMCP logging.basicConfig( format%(asctime)s - %(name)s - %(levelname)s - %(message)s, levellogging.INFO ) logger logging.getLogger(ShellServer) mcp FastMCP(ShellCommandExecutor) mcp.tool() def execute_shell_command(command: str) - str: 执行一个Shell命令并返回其输出。 注意在生产环境中需严格限制可执行的命令范围以防安全风险。 logger.warning(f执行Shell命令: {command}) try: # 安全提示此处仅作演示实际应使用白名单或严格校验command参数 result subprocess.run( command, shellTrue, checkTrue, textTrue, capture_outputTrue, timeout10 # 增加超时限制 ) return result.stdout except subprocess.CalledProcessError as e: return f命令执行失败 (返回码 {e.returncode}): {e.stderr} except subprocess.TimeoutExpired: return 命令执行超时超过10秒。 except Exception as e: return f执行过程中发生未知错误: {str(e)} if __name__ __main__: logger.info(Shell命令执行MCP Server启动stdio模式...) # 使用stdio传输模式运行这会阻塞直到连接关闭 mcp.run(transportstdio)这个服务器更简单它通过mcp.run(transportstdio)以标准输入输出模式运行等待来自父进程即我们的LangGraph Client的调用。4. 在LangGraph中集成多源MCP工具现在进入最激动人心的部分让LangGraph智能体同时使用远程SSE和本地stdio的MCP工具。我们将使用langchain-mcp-adapters库中的MultiServerMCPClient它是实现这一目标的关键。首先安装必要的LangChain生态库pip install langchain langchain-ollama langchain-mcp-adapters langgraph假设我们使用Ollama本地运行的Llama模型作为智能体的“大脑”。我们创建一个.env文件来管理配置# .env 文件 LLM_TEMPERATURE0.2 OLLAMA_MODELllama3.2:latest OLLAMA_BASE_URLhttp://127.0.0.1:11434 PY_PROJECT_DIR/path/to/your/project/ # 替换为你的项目路径 SSE_BASE_URLhttp://YOUR_SERVER_IP:8000/sse # 替换为你的SSE服务器IP注意SSE_BASE_URL需要指向你运行interest_sse_server.py的机器的实际IP地址和端口。如果Client和Server在同一台机器可以使用http://127.0.0.1:8000/sse。接下来是LangGraph智能体的主客户端代码multi_mcp_agent.pyimport asyncio import logging import os from dotenv import load_dotenv from langchain_ollama import ChatOllama from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent # 加载配置 load_dotenv() llm_temperature float(os.getenv(LLM_TEMPERATURE, 0.2)) ollama_model os.getenv(OLLAMA_MODEL) ollama_base_url os.getenv(OLLAMA_BASE_URL) py_project_dir os.getenv(PY_PROJECT_DIR) sse_base_url os.getenv(SSE_BASE_URL) # 初始化LLM llm ChatOllama( base_urlollama_base_url, modelollama_model, temperaturellm_temperature ) logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) async def main(): 主函数初始化多MCP客户端创建智能体并执行示例任务。 logger.info(初始化多MCP服务器客户端...) # 配置MultiServerMCPClient同时连接SSE和stdio服务器 client MultiServerMCPClient( servers{ InterestCalculator: { url: sse_base_url, # 远程SSE服务器 transport: sse }, ShellCommandExecutor: { command: python, # 启动本地stdio服务器的命令 args: [os.path.join(py_project_dir, shell_mcp_server.py)], transport: stdio } } ) # 从所有配置的服务器获取工具列表 tools await client.get_tools() logger.info(f成功加载工具: {[tool.name for tool in tools]}) # 使用获取到的工具和LLM创建ReAct智能体 agent create_react_agent(llm, tools) # 测试案例1询问复利定义会触发LLM自身知识或工具调用 logger.info(\n 测试案例1: 解释复利定义 ) response_1 await agent.ainvoke({ messages: [(human, 请向我解释一下复利compound interest的定义。)] }) # 打印智能体的最终回答 last_message response_1[messages][-1] if hasattr(last_message, content): print(f智能体回答: {last_message.content}) # 测试案例2计算复利应调用远程InterestCalculator工具 logger.info(\n 测试案例2: 计算复利 ) response_2 await agent.ainvoke({ messages: [(human, 如果本金是1000元年利率是3.75%那么一年的复利是多少)] }) last_message response_2[messages][-1] if hasattr(last_message, content): print(f智能体回答: {last_message.content}) # 测试案例3执行Shell命令应调用本地ShellCommandExecutor工具 logger.info(\n 测试案例3: 检查系统内存 ) response_3 await agent.ainvoke({ messages: [(human, 请用free命令查看一下当前系统的内存使用情况。)] }) last_message response_3[messages][-1] if hasattr(last_message, content): print(f智能体回答:\n{last_message.content}) # 测试案例4混合任务智能体需自主规划工具使用顺序 logger.info(\n 测试案例4: 混合任务 ) response_4 await agent.ainvoke({ messages: [(human, 先帮我计算一下2000元本金、5%利率的单利然后告诉我当前目录下有哪些Python文件。)] }) print(\n智能体执行混合任务的过程:) for msg in response_4[messages]: if hasattr(msg, content) and msg.content: print(f- {msg.content[:200]}...) # 截断长输出 # 非常重要异步清理客户端连接 await client.aclose() logger.info(客户端连接已关闭。) if __name__ __main__: asyncio.run(main())代码深度解读与注意事项MultiServerMCPClient这是核心类。它接受一个服务器配置字典键是服务器标识符可自定义值是该服务器的连接配置。对于SSE服务器需要url和transport: sse对于stdio服务器需要command、args和transport: stdio。它会自动管理这些连接的生命周期。工具获取与绑定await client.get_tools()会连接到所有配置的服务器获取它们暴露的工具列表并封装成LangChain Tool对象。这些工具随后被注入到ReAct智能体中。智能体工作流create_react_agent创建了一个基于ReActReasoning Acting模式的智能体。当用户提出问题时智能体会自主思考Reasoning决定是否需要调用工具、调用哪个工具然后执行Acting根据工具返回结果再进行下一步思考或给出最终答案。异步清理await client.aclose()至关重要。它会优雅地关闭所有SSE连接和stdio子进程释放资源。务必在程序结束前调用。运行这个客户端你将看到智能体如何动态选择工具、执行计算和命令并将结果整合到对话中。这种架构的威力在于你可以随时扩展新的MCP Server无论是本地还是远程只需在配置中添加一项智能体就能立即获得新能力无需修改核心代码。5. 生产环境进阶考量与调试技巧将上述Demo部署到生产环境还需要考虑更多因素。以下是一些关键点和实用技巧安全加固SSE服务器认证上述示例的SSE端点没有认证。在生产中你需要在Starlette应用中添加中间件例如使用Bearer Token验证。# 示例简单的Token验证中间件 from starlette.middleware import Middleware from starlette.middleware.authentication import AuthenticationMiddleware # 实现一个简单的BearerTokenAuthBackendHTTPS务必通过反向代理如Nginx为SSE服务器配置HTTPS防止通信被窃听。Shell命令限制示例中的Shell工具极其危险。生产环境必须使用命令白名单、参数严格校验、沙箱环境如Docker容器来执行并避免使用shellTrue。性能与稳定性连接池与重连MultiServerMCPClient在初始化时建立连接。对于需要高可用的场景你可能需要实现客户端的重连逻辑以应对网络波动或服务器重启。超时设置为SSE连接和工具调用设置合理的超时时间避免智能体因某个工具挂起而长时间阻塞。错误处理在客户端代码中增强错误处理例如某个MCP Server不可用时是跳过该服务器所有工具还是让整个智能体失败需要有明确的策略。监控与调试日志分级为MCP库、你的服务器和客户端设置详细的日志logging.DEBUG级别可以清晰看到每一条MCP协议消息的发送和接收对于排查通信问题无比重要。使用mcp-cli进行测试在部署LangGraph智能体之前可以先用MCP官方CLI工具测试你的Server是否正常工作。# 测试SSE服务器 npx modelcontextprotocol/inspector http://your-server:8000/sse # 测试stdio服务器 npx modelcontextprotocol/inspector python shell_mcp_server.py这个Inspector工具会提供一个Web界面列出所有可用工具并允许你手动调用是验证Server正确性的利器。网络检查确保Client能访问SSE服务器的/sse(GET) 和/messages/(POST) 端点。防火墙或CORS策略可能是常见的绊脚石。我在实际项目中将一个内部数据分析工具包封装成MCP Server通过SSE暴露给部署在Kubernetes中的LangGraph智能体。最大的收获是清晰的协议边界MCP和灵活的传输层SSE使得团队协作变得顺畅。后端团队可以独立升级工具实现只要MCP接口不变AI团队就无需修改智能体代码。这种解耦带来的效率提升远超过初期的搭建成本。