一、三种 MCP 通讯方式对比MCPModel Context Protocol目前支持三种通讯方式各有特点1. stdio标准输入输出工作原理通过本地进程的 stdin/stdout 进行通信客户端以子进程方式启动 MCP 服务器双方通过管道交换 JSON-RPC 格式消息换行符分割适用场景本地进程间通信如命令行工具、文件系统操作简单的批处理任务或工具调用优点限制实现简单低延迟仅限本地不支持分布式部署无需网络配置服务端不能输出控制台日志会污染协议流适合本地开发——————————————————2. SSEServer-Sent Events⚠️ 已弃用工作原理基于 HTTP 长连接实现单向消息推送客户端通过GET /sse建立连接服务器通过 SSE 流发送 JSON-RPC 消息客户端通过POST /message发送请求适用场景远程服务调用如云服务、多客户端监控需要实时数据推送的场景如流式对话优点限制支持实时单向推送2025年3月后已被 Streamable HTTP 取代适合流式交互连接中断后无法恢复————————需维持长连接资源消耗较高3. Streamable HTTP流式 HTTP✅ 官方推荐工作原理2025年3月引入通过统一的/message端点实现双向通信客户端通过 HTTP POST 发送请求服务器可将响应升级为 SSE 流式传输按需支持无状态模式无需维持长连接核心优势✅ 支持连接恢复无需重新开始✅ 无需维持长连接降低资源消耗✅ 统一端点设计/message简化接口✅ 兼容现有基础设施负载均衡、中间件等适用场景高并发远程服务调用需要灵活流式响应的场景如 AI 助手动态输出二、stdio 模式实现架构流程图创建 Server → 启动服务 → Client 连接 → 加载 Tools → Agent 调用Step 1创建 MCP Serverfrom mcp.server.fastmcp import FastMCP mcp FastMCP(Math Tools) mcp.tool() def add(a: int, b: int) - int: return a b mcp.tool() def multiply(a: int, b: int) - int: return a * b if __name__ __main__: mcp.run(transportstdio) # ← 关键stdio 模式说明使用mcp.tool()注册工具方法run(transportstdio)启动服务。Step 2启动 MCP Serverpython server.py 服务启动后会持续监听 stdin/stdout 的读写事件。Step 3开发 MCP Client含 Agent3.1 定义 Server 参数server_params StdioServerParameters( commandpython, args[path/to/mcp_stdio_server.py] )3.2 加载 MCP Toolsasync with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) print(f成功加载 {len(tools)} 个工具)3.3 创建 Agent 并调用agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, ) resp await agent.ainvoke(1417*5?)完整示例代码import os import asyncio from dotenv import load_dotenv from mcp import StdioServerParameters, ClientSession from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools from langchain.agents import initialize_agent, AgentType from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelqwen-plus, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://ws-kcaoxz5olbi6r3qa.cn-beijing.maas.aliyuncs.com/compatible-mode/v1, streamingTrue, temperature0.7, ) async def create_mcp_stdio_client(): server_params StdioServerParameters( commandpython, args[D:/sd14/ai-agent/app/mcp_/stdio/mcp_stdio_server.py] ) try: async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) print(f成功加载 {len(tools)} 个工具) agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, ) resp await agent.ainvoke(1417*5?) print(f\n回答: {resp}) return resp except Exception as e: print(f连接失败: {e}) traceback.print_exc() if __name__ __main__: asyncio.run(create_mcp_stdio_client())三、SSE 模式实现已弃用仅供参考⚠️官方已废弃建议直接使用 Streamable HTTPServer 端差异if __name__ __main__: mcp.run(transportsse) # ← 仅 transport 参数不同Client 端差异from langchain_mcp_adapters.client import MultiServerMCPClient client MultiServerMCPClient({ math: { url: http://127.0.0.1:8000/sse, transport: sse, # ← 指定传输方式 } }) tools await client.get_tools()SSE 模式完整代码SSE Server 端 (mcp_sse_server.py)from mcp.server.fastmcp import FastMCP # 创建 MCP 服务器实例 mcp FastMCP(Math Tools - SSE) # 注册工具方法 mcp.tool() def add(a: int, b: int) - int: 计算两个整数的和 return a b mcp.tool() def multiply(a: int, b: int) - int: 计算两个整数的乘积 return a * b mcp.tool() def subtract(a: int, b: int) - int: 计算两个整数的差 (a - b) return a - b mcp.tool() def divide(a: int, b: int) - float: 计算两个整数的商 (a / b) if b 0: raise ValueError(除数不能为0) return a / b mcp.tool() def power(base: int, exponent: int) - int: 计算幂运算 (base ^ exponent) return base ** exponent # 启动服务 if __name__ __main__: mcp.run(transportsse)SSE Client 端 (mcp_sse_client.py)import os import asyncio from dotenv import load_dotenv from langchain_mcp_adapters.client import MultiServerMCPClient from langchain.agents import initialize_agent, AgentType from langchain_openai import ChatOpenAI # 加载环境变量 load_dotenv() # 初始化 LLM llm ChatOpenAI( modelqwen-plus, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://ws-kcaoxz5olbi6r3qa.cn-beijing.maas.aliyuncs.com/compatible-mode/v1, streamingTrue, temperature0.7, ) # 创建 SSE 客户端 async def create_mcp_sse_client(): try: # 创建多服务器客户端 client MultiServerMCPClient( { math: { url: http://127.0.0.1:8000/sse, # SSE 端点 transport: sse, # 指定传输方式 } } ) # 获取所有工具 tools await client.get_tools() print(f✅ 成功加载 {len(tools)} 个工具:) for tool in tools: print(f - {tool.name}: {tool.description}) # 创建 Agent agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, handle_parsing_errorsTrue, ) # 测试调用 print(\n *50) print( 测试 1: 基础运算) print(*50) resp1 await agent.ainvoke(请计算 14 17 * 5 ?) print(f\n 回答: {resp1[output]}\n) print(*50) print( 测试 2: 复杂表达式) print(*50) resp2 await agent.ainvoke(计算 (100 - 25) * 4 / 3 ?) print(f\n 回答: {resp2[output]}\n) print(*50) print( 测试 3: 幂运算) print(*50) resp3 await agent.ainvoke(计算 2 的 10 次方等于多少) print(f\n 回答: {resp3[output]}\n) return resp1 except Exception as e: print(f❌ 连接失败: {e}) import traceback traceback.print_exc() return None # 入口 if __name__ __main__: print( 启动 SSE MCP 客户端...) print( 请确保 SSE 服务器已启动: python mcp_sse_server.py) print() asyncio.run(create_mcp_sse_client())SSE 模式启动步骤# 终端1: 启动 SSE Server python mcp_sse_server.py # 终端2: 启动 SSE Client python mcp_sse_client.py四、Streamable HTTP 模式实现 ✅ 推荐Server 端if __name__ __main__: mcp.run(transportstreamable-http)Client 端client MultiServerMCPClient({ math: { url: http://127.0.0.1:8000/mcp, transport: streamable_http, # ← 注意下划线 } }) tools await client.get_tools()Streamable HTTP 完整代码Streamable HTTP Server 端 (mcp_streamable_server.py)from mcp.server.fastmcp import FastMCP # 创建 MCP 服务器实例 mcp FastMCP(Math Tools - Streamable HTTP) # 注册工具方法 mcp.tool() def add(a: int, b: int) - int: 计算两个整数的和 return a b mcp.tool() def multiply(a: int, b: int) - int: 计算两个整数的乘积 return a * b mcp.tool() def subtract(a: int, b: int) - int: 计算两个整数的差 (a - b) return a - b mcp.tool() def divide(a: int, b: int) - float: 计算两个整数的商 (a / b) if b 0: raise ValueError(除数不能为0) return a / b mcp.tool() def power(base: int, exponent: int) - int: 计算幂运算 (base ^ exponent) return base ** exponent mcp.tool() def factorial(n: int) - int: 计算阶乘 (n!) if n 0: raise ValueError(阶乘只支持非负整数) if n 0 or n 1: return 1 result 1 for i in range(2, n 1): result * i return result mcp.tool() def fibonacci(n: int) - list: 生成前 n 个斐波那契数列 if n 0: return [] if n 1: return [0] fib [0, 1] for i in range(2, n): fib.append(fib[i-1] fib[i-2]) return fib # 启动服务 if __name__ __main__: mcp.run(transportstreamable-http)Streamable HTTP Client 端 (mcp_streamable_client.py)import os import asyncio from dotenv import load_dotenv from langchain_mcp_adapters.client import MultiServerMCPClient from langchain.agents import initialize_agent, AgentType from langchain_openai import ChatOpenAI from langchain.callbacks import StreamingStdOutCallbackHandler # 加载环境变量 load_dotenv() # 初始化 LLM llm ChatOpenAI( modelqwen-plus, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://ws-kcaoxz5olbi6r3qa.cn-beijing.maas.aliyuncs.com/compatible-mode/v1, streamingTrue, temperature0.7, callbacks[StreamingStdOutCallbackHandler()], ) # 创建 Streamable HTTP 客户端 async def create_mcp_streamable_client(): try: # 创建多服务器客户端 client MultiServerMCPClient( { math: { url: http://127.0.0.1:8000/mcp, # Streamable HTTP 端点 transport: streamable_http, # 注意下划线 } } ) # 获取所有工具 tools await client.get_tools() print(f\n✅ 成功加载 {len(tools)} 个工具:) for tool in tools: print(f - {tool.name}: {tool.description}) # 创建 Agent agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, handle_parsing_errorsTrue, max_iterations5, ) # 测试调用 test_cases [ 请计算 14 17 * 5 ?, 计算 (100 - 25) * 4 / 3 ?, 2 的 10 次方等于多少, 计算 5!5的阶乘等于多少, 生成前 10 个斐波那契数列, ] for i, query in enumerate(test_cases, 1): print(\n * 60) print(f 测试 {i}: {query}) print( * 60) try: resp await agent.ainvoke(query) print(f\n 最终回答:\n{resp[output]}\n) except Exception as e: print(f❌ 调用失败: {e}) return True except Exception as e: print(f❌ 连接失败: {e}) import traceback traceback.print_exc() return False # 高级用法上下文管理 async def advanced_usage(): 展示如何优雅地管理客户端生命周期 async with MultiServerMCPClient( { math: { url: http://127.0.0.1:8000/mcp, transport: streamable_http, } } ) as client: tools await client.get_tools() print(f✅ 加载了 {len(tools)} 个工具) # 创建 Agent agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, ) # 多轮对话 queries [ 计算 3 4 * 5, 计算 10!, 生成前 5 个斐波那契数列, ] for query in queries: print(f\n 用户: {query}) resp await agent.ainvoke(query) print(f AI: {resp[output]}) # 入口 if __name__ __main__: print( 启动 Streamable HTTP MCP 客户端...) print( 请确保 Streamable HTTP 服务器已启动: python mcp_streamable_server.py) print( 访问地址: http://127.0.0.1:8000/mcp) print() # 基础用法 asyncio.run(create_mcp_streamable_client()) # 高级用法取消注释使用 # asyncio.run(advanced_usage())Streamable HTTP 模式启动步骤# 终端1: 启动 Streamable HTTP Server python mcp_streamable_server.py # 终端2: 启动 Streamable HTTP Client python mcp_streamable_client.py快速对比总览特性stdioSSE已弃用Streamable HTTP ✅适用场景本地开发远程调用生产环境/云服务网络需求无需网络需要 HTTP需要 HTTP连接状态进程级需维持长连接支持无状态连接恢复N/A❌ 不支持✅ 支持资源消耗低高低官方推荐本地开发❌ 已淘汰✅强烈推荐五、补充如何选择合适的传输方式┌─────────────────────────────────────────────────────┐ │ 你的需求是什么 │ ├─────────────────────────────────────────────────────┤ │ • 本地开发/测试 → 使用 stdio │ │ • 远程服务/生产部署 → 使用 Streamable HTTP│ │ • 需要流式响应 → 使用 Streamable HTTP│ │ • 高并发场景 → 使用 Streamable HTTP│ │ • 旧项目维护不推荐新项目→ SSE但建议迁移 │ └─────────────────────────────────────────────────────┘六、踩坑提醒 问题解决方案stdio 模式服务端不能print()使用logging输出到文件SSE 连接中断迁移到 Streamable HTTPStreamable HTTP端口被占用更换端口或检查服务是否已启动工具加载失败检查 server 端的 transport 参数是否匹配环境变量未生效确保.env文件在项目根目录使用load_dotenv()