1. 从一次工具调用超时说起MCP 的 stdio 与 SSE 到底怎么选MCPModel Context Protocol是让大模型调用外部工具的协议层你可以把它理解成给模型装插件的 USB 接口。它规定了模型怎么发现工具、怎么传参、怎么拿回结果。而 stdio 和 SSE 是 MCP 客户端连接服务端的两种通信方式stdio 走标准输入输出把服务端当子进程拉起SSE 走 HTTP 长连接服务端是独立进程客户端通过 URL 连过去。我最早做多工具 Agent 时三个服务端全用 stdio本地跑得好好的一放到内网服务器上就出问题——客户端进程一挂所有工具全断日志还散在四个终端里。后来把天气、SQL、Python 执行三个服务端改成 SSE 独立部署客户端只负责连 URL稳定性立刻不一样。但反过来纯本地做代码补全、文件读写这类工具stdio 又比 SSE 省事得多不用管端口、不用管进程存活。所以这篇不空谈概念直接给你两套可复制的客户端骨架client_stdio.py和client_sse.py配上三个真实服务端天气、SQL、Python 执行再演示启动、连通性验证、以及怎么在两种方式之间切换。适合正在搭 MCP 工具链、纠结通信方式、或者被连不上/工具列表为空卡住的开发者。核心检索词就三个MCP、stdio、SSE下面全部围绕它们展开。2. 前置准备TaoToken 统一 Key 与 API 通道不管用哪种通信方式客户端最终都要调 LLM 来做 Function Calling。这里我用 TaoToken 作为统一的模型接入通道好处是一个 Key 走通所有模型不用在多个平台之间切来切去。先去控制台拿 Key控制台入口https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI Keys 管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite拿到 Key 之后API 基地址用https://taotoken.net/api注意这个地址不加 UTM 参数直接写进代码。如果你要长期跑编码类 Agent可以看下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite环境依赖用requirements.txt固定版本避免mcp和openai版本打架mcp1.6.0 openai1.74.0 httpx0.28.1 httpx-sse0.4.0 pymysql1.1.1 pandas2.2.3 python-dotenv1.1.0 uvicorn0.34.1 sse-starlette2.2.1.env文件统一放模型配置两种通信方式共用同一份BASE_URLhttps://taotoken.net/api MODELqwen2.5-coder:14b OPENAI_API_KEY你的TaoTokenKey注意BASE_URL结尾不要多加/v1OpenAI SDK 会自己拼路径。我试过写成https://taotoken.net/api/v1结果 404排查了半小时。3. 可复制配置stdio 与 SSE 两套客户端骨架3.1 stdio 客户端把服务端当子进程拉起stdio 的核心是StdioServerParameters它告诉客户端用什么命令启动服务端脚本。Python 脚本用pythonJS 脚本用node。import asyncio import os import json from typing import Dict from contextlib import AsyncExitStack from openai import OpenAI from dotenv import load_dotenv from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client load_dotenv() class MultiServerMCPClient: def __init__(self): self.exit_stack AsyncExitStack() self.client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(BASE_URL) ) self.model os.getenv(MODEL) self.sessions: Dict[str, ClientSession] {} self.all_tools [] async def _start_one_server(self, script_path: str) - ClientSession: is_python script_path.endswith(.py) is_js script_path.endswith(.js) if not (is_python or is_js): raise ValueError(服务端脚本必须是 .py 或 .js 文件) command python if is_python else node server_params StdioServerParameters( commandcommand, args[script_path], envNone ) stdio_transport await self.exit_stack.enter_async_context( stdio_client(server_params) ) read_stream, write_stream stdio_transport session await self.exit_stack.enter_async_context( ClientSession(read_stream, write_stream) ) await session.initialize() return session async def connect_to_servers(self, servers: dict): for server_name, script_path in servers.items(): session await self._start_one_server(script_path) self.sessions[server_name] session resp await session.list_tools() for tool in resp.tools: function_name f{server_name}_{tool.name} self.all_tools.append({ type: function, function: { name: function_name, description: tool.description, parameters: tool.inputSchema } }) print(\n已连接到下列服务端:) for name in servers: print(f - {name}: {servers[name]}) print(\n汇总的工具:) for t in self.all_tools: print(f - {t[function][name]}) async def _call_mcp_tool(self, tool_full_name: str, tool_args: dict) - str: parts tool_full_name.split(_, 1) if len(parts) ! 2: return f无效的工具名称: {tool_full_name} server_name, tool_name parts session self.sessions.get(server_name) if not session: return f找不到服务端: {server_name} resp await session.call_tool(tool_name, tool_args) return resp.content if resp.content else 工具执行无输出 async def chat_loop(self): print(\n多服务端 MCP 客户端已启动输入 quit 退出。) messages [] while True: query input(\n你: ).strip() if query.lower() quit: break messages.append({role: user, content: query}) messages messages[-20:] response self.client.chat.completions.create( modelself.model, messagesmessages, toolsself.all_tools ) if response.choices[0].finish_reason tool_calls: tool_call response.choices[0].message.tool_calls[0] tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) print(f\n[调用工具: {tool_name}, 参数: {tool_args}]\n) result await self._call_mcp_tool(tool_name, tool_args) messages.append(response.choices[0].message.model_dump()) messages.append({ role: tool, content: result, tool_call_id: tool_call.id }) response self.client.chat.completions.create( modelself.model, messagesmessages ) print(f\nAI: {response.choices[0].message.content}) async def cleanup(self): await self.exit_stack.aclose() async def main(): servers { weather: weather_server.py, SQLServer: sql_server.py, PythonServer: python_server.py } client MultiServerMCPClient() try: await client.connect_to_servers(servers) await client.chat_loop() finally: await client.cleanup() if __name__ __main__: asyncio.run(main())3.2 SSE 客户端连 URL 而不是拉进程SSE 客户端把stdio_client换成sse_client参数从脚本路径变成 URL。其余逻辑完全一致。from mcp.client.sse import sse_client async def _start_one_server(self, server_url: str) - ClientSession: sse_transport await self.exit_stack.enter_async_context( sse_client(urlserver_url) ) read_stream, write_stream sse_transport session await self.exit_stack.enter_async_context( ClientSession(read_stream, write_stream) ) await session.initialize() return session async def main(): servers { weather: http://127.0.0.1:8001/sse, PythonServer: http://127.0.0.1:8002/sse, SQLServer: http://127.0.0.1:8003/sse } client MultiServerMCPClient() try: await client.connect_to_servers(servers) await client.chat_loop() finally: await client.cleanup()3.3 服务端切换一行注释决定通信方式三个服务端脚本weather_server.py、sql_server.py、python_server.py的末尾都长这样切换通信方式只改一行if __name__ __main__: # 以标准 I/O 方式运行 MCP 服务端 # mcp.run(transportstdio) # 使用 SSE 方式运行 MCP 服务端 mcp.run(transportsse)用FastMCP初始化时指定端口SSE 模式才会监听对应端口from mcp.server.fastmcp import FastMCP mcp FastMCP(WeatherServer, port8001)4. 启动与连通性验证两种方式完整动作4.1 SSE 模式启动每个服务端单独开一个终端用uv run拉起uv run weather_server.py uv run python_server.py uv run sql_server.py看到类似Uvicorn running on http://127.0.0.1:8001就说明 SSE 服务端起来了。然后另开终端跑客户端uv run client_sse.py客户端启动后会打印已连接的服务端和汇总工具列表。如果工具列表为空说明 SSE 连接没建立成功先检查端口和 URL 路径/sse不能少。4.2 stdio 模式启动stdio 不需要单独启动服务端客户端会自己拉起子进程。直接跑uv run client_stdio.py启动日志里会看到三个服务端被依次拉起工具列表同样会打印出来。这里有个坑stdio 模式下服务端脚本里的mcp.run(transportsse)必须注释掉否则子进程会去抢端口客户端反而连不上。4.3 连通性验证发一条真实请求不管哪种模式启动后输入一句需要调工具的话比如你: 帮我查一下南昌的天气正常输出会先打印[调用工具: weather_query_weather, 参数: {...}]然后返回格式化后的天气文本。如果只返回模型自己的话、没有工具调用日志说明all_tools是空的回到上一步检查工具列表。再验证 SQL 工具你: 查一下 school 数据库里 students 表有多少行Python 执行工具你: 用 Python 算一下 1 到 100 的和三个工具都能触发说明 MCP 链路完全通了。5. 本篇常见错排查5.1 工具列表为空 / 连接超时SSE 模式下最常见。先确认服务端真的在监听curl http://127.0.0.1:8001/sse应该返回事件流而不是拒绝连接。如果服务端在另一台机器上把127.0.0.1换成实际内网 IP并确认防火墙放行。stdio 模式下工具为空多半是command写错比如系统里只有python3没有python改成python3即可。5.2 端口冲突SSE 模式三个服务端端口必须不同8001/8002/8003。如果报Address already in use用lsof -i:8001找到占用进程杀掉或者改FastMCP初始化时的port参数。5.3 Function Calling 参数格式报错MCP 返回的inputSchema字段名和 OpenAI 要求的parameters不一致。如果你直接透传OpenAI 会报Invalid schema。上面代码里我做了字段映射把inputSchema转成parameters并保留type、properties、required三个字段。这一步不能省。5.4 模型不触发工具调用有些模型对 Function Calling 支持不完整会直接把工具描述当普通文本回复。换一个明确支持 tools 参数的模型即可。在 TaoToken 的模型对话页可以先手动验证模型是否支持工具调用https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite5.5 stdio 子进程残留客户端异常退出时stdio 拉起的子进程可能没被回收。AsyncExitStack的aclose()负责清理确保finally里调用了cleanup()。如果还是残留手动pkill -f server.py。6. 选型结论与接入入口回到最初的问题本地工具链、单机开发、工具数量少且生命周期跟客户端绑定用 stdio省去端口和进程管理远程服务、多客户端共享、需要独立部署和监控用 SSE服务端和客户端解耦。两者可以混用——比如天气服务走 SSE 远程部署文件读写走 stdio 本地拉起客户端里同时维护两种 session 即可。接入所需的 Key 和文档入口API Keyshttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite模型对话验证https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite最后留一个实操建议先把三个服务端都用 SSE 跑起来确认工具列表和调用链路没问题再把其中不需要远程共享的改成 stdio。这样排障时变量最少不会一上来就被到底是通信方式问题还是工具本身问题绕进去。