1. 从 stdio 到 SSEMCP 双传输联调到底难在哪MCPModel Context Protocol是让大模型调用本地或远程工具的一套协议FastMCP 是它官方 Python SDK 里最省事的服务端封装ClientSession 则是客户端侧负责握手、列工具、发调用的核心对象。如果你正在用 FastMCP 写服务端、用 ClientSession 写客户端大概率会卡在同一个地方stdio 模式跑通了换成 SSE 就报连接超时或者 SSE 通了stdio 又提示脚本路径不对。两种传输方式的连接参数、启动命令、URL 形态完全不同但工具定义部分几乎一模一样这种“一半相同一半不同”的结构最容易让人反复踩坑。这篇面向想一次跑通两种传输模式的开发者给出可复制的 FastMCP 服务端骨架、ClientSession 连接配置以及用 TaoToken 统一 Key/API 通道完成鉴权与调用的验证步骤。适合已经装好 Python 3.12、知道什么是 tool call、但还没把 stdio 和 SSE 两条链路都跑顺的人。我会把服务端和客户端拆开写每个传输模式单独给一份能直接运行的代码最后用同一个 TaoToken Key 把两条链路都验证一遍。先说结论stdio 适合本地脚本、单机调试、进程内通信SSE 适合服务常驻、多客户端连接、跨机器调用。两者在 FastMCP 里只差mcp.run(transport...)一行在 ClientSession 里只差 transport 的构造方式。把这两处吃透剩下的工具逻辑可以完全复用。2. TaoToken 前置统一 Key 与 API 通道准备在写代码之前先把鉴权和 API 通道准备好。MCP 本身不绑定某一家模型但客户端要调用大模型来决定“该调哪个工具、传什么参数”所以你需要一个稳定的 OpenAI 兼容入口。TaoToken 提供的就是这个入口一个 Key 同时覆盖模型对话和后续的 coding 场景省得在多个平台之间来回切。第一步打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册并登录。第二步进入控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建 API Key。第三步如果你后面要长期跑编码类 Agent可以顺手看一下 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它和按量调用是两条不同的计费路径按自己的使用频率选。拿到 Key 之后记下两个东西Base URL 是https://taotoken.net/apiKey 形如sk-xxxx。这两个值会同时出现在 stdio 和 SSE 两种模式的客户端里因为模型调用这一层和传输层是解耦的——不管 MCP 用哪种 transport客户端调模型都走同一个 OpenAI 兼容接口。注意API 地址不要加 UTM 参数直接写https://taotoken.net/api即可UTM 只用于官网和 deep link 的跳转统计。环境准备部分建议单独建一个 conda 环境避免和系统里的包打架conda create -n mcp python3.12 -y conda activate mcp pip install mcp openai如果你要测试 SSE还需要确认端口没被占用测试 stdio 则要确认 server 脚本路径是绝对路径或相对当前工作目录正确。这两点后面排障会反复用到。3. 可复制配置FastMCP 服务端骨架stdio SSE服务端结构分三块创建 FastMCP 实例、用mcp.tool()注册工具、最后mcp.run()指定传输方式。先给一个最小可运行骨架工具只保留一个写文件的方便你验证链路跑通后再往里加自己的逻辑。# server.py from mcp.server.fastmcp import FastMCP # stdio 模式不需要 portSSE 模式需要 port mcp FastMCP(my_server, version1.0.0, port8000) mcp.tool() def write_to_txt_page(file_name: str, text: str) - None: 向指定本地文件追加文本。 Args: file_name: 要写入的文件名 text: 要写入的文本内容 try: with open(file_name, a, encodingutf-8) as f: f.write(text) except Exception as e: raise Exception(f写入文件时出错: {e}) if __name__ __main__: # 切换这里即可在 stdio 和 SSE 之间切换 mcp.run(transportstdio) # mcp.run(transportsse)关键点有三个。第一FastMCP初始化时给port只在 SSE 模式下有意义stdio 模式传了也不影响但为了清晰建议 SSE 才写。第二工具函数的 docstring 会被 ClientSession 通过list_tools()读出来作为模型判断“何时调用”的依据所以参数说明要写清楚别偷懒。第三mcp.run(transport...)是唯一的传输开关stdio走标准输入输出sse会在0.0.0.0:8000起一个 HTTP 服务SSE 端点默认是/sse。如果你要同时保留两种模式可以拆成两个入口文件或者用环境变量控制import os if __name__ __main__: transport os.getenv(MCP_TRANSPORT, stdio) mcp.run(transporttransport)这样启动时MCP_TRANSPORTsse python server.py就走 SSE不设就走 stdio部署时更灵活。4. ClientSession 连接配置stdio 与 SSE 的差异点客户端是重点因为两种传输的差异全在这里。整体结构分三块初始化 OpenAI 兼容客户端指向 TaoToken、连接 MCP 服务器、处理用户请求并完成 tool call 回填。先看 stdio 版本的连接部分。# client_stdio.py import asyncio import json import sys from typing import Optional from contextlib import AsyncExitStack from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from openai import OpenAI class MCPClient: def __init__(self): self.session: Optional[ClientSession] None self.exit_stack AsyncExitStack() # 统一走 TaoToken 的 OpenAI 兼容入口 self.client OpenAI( api_keysk-你的TaoTokenKey, base_urlhttps://taotoken.net/api ) self.model gpt-4o-mini # 按需替换 async def connect_to_server(self, server_script_path: str): if not server_script_path.endswith((.py, .js)): raise ValueError(服务器脚本必须是 .py 或 .js 文件) command python if server_script_path.endswith(.py) else node server_params StdioServerParameters( commandcommand, args[server_script_path], envNone ) stdio_transport await self.exit_stack.enter_async_context( stdio_client(server_params) ) self.stdio, self.write stdio_transport self.session await self.exit_stack.enter_async_context( ClientSession(self.stdio, self.write) ) await self.session.initialize() response await self.session.list_tools() print(已连接, 可用工具:, [t.name for t in response.tools])SSE 版本只改connect_to_server和main里的入参# client_sse.py 片段 from mcp.client.sse import sse_client async def connect_to_server(self, url: str): sse_transport await self.exit_stack.enter_async_context( sse_client(url) ) self.session await self.exit_stack.enter_async_context( ClientSession(*sse_transport) ) await self.session.initialize() response await self.session.list_tools() print(已连接, 可用工具:, [t.name for t in response.tools]) async def main(): client MCPClient() try: await client.connect_to_server(http://0.0.0.0:8000/sse) await client.chat_loop() finally: await client.cleanup()对比一下就很清楚stdio 传的是StdioServerParameters由客户端负责拉起子进程SSE 传的是一个 URL服务端必须已经在跑。ClientSession的构造在 stdio 下是ClientSession(stdio, write)在 SSE 下是ClientSession(*sse_transport)因为sse_client返回的就是一个二元组。这是最容易写错的地方参数顺序或解包方式错了会直接抛类型错误。处理用户请求的逻辑两种模式完全一致核心是把list_tools()的结果转成 OpenAI 的tools格式拿到tool_calls后调session.call_tool再把结果作为role: tool的消息回填发起第二轮模型调用async def process_query(self, query: str) - str: messages [{role: user, content: query}] response await self.session.list_tools() available_tools [{ type: function, function: { name: t.name, description: t.description, parameters: t.inputSchema } } for t in response.tools] resp self.client.chat.completions.create( modelself.model, messagesmessages, toolsavailable_tools ) msg resp.choices[0].message final_text [] if msg.tool_calls: tool_call msg.tool_calls[0] tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) result await self.session.call_tool(tool_name, tool_args) final_text.append(f[调用工具 {tool_name}, 参数 {tool_args}]) messages.append({ role: assistant, content: , tool_calls: [tool_call] }) messages.append({ role: tool, content: result.content[0].text if result.content[0].text else , name: tool_name, tool_call_id: tool_call.id }) resp2 self.client.chat.completions.create( modelself.model, messagesmessages, toolsavailable_tools ) final_text.append(resp2.choices[0].message.content) else: final_text.append(msg.content) return \n.join(final_text)注意inputSchema转成parameters这一步字段名不一样写错模型就识别不到工具。另外tool_call_id必须原样回填否则第二轮调用会被服务端拒绝。5. 验证请求与成功结果两条链路各跑一遍先验证 stdio。终端 A 不需要单独起服务端客户端会自己拉起子进程python client_stdio.py server.py看到已连接, 可用工具: [write_to_txt_page]就说明握手成功。输入把 hello mcp 写到 test.txt模型会返回一个 tool_call客户端执行后你会在当前目录看到test.txt内容追加了hello mcp。整个过程服务端进程由客户端管理退出客户端时子进程也会被回收。再验证 SSE。终端 A 先起服务端MCP_TRANSPORTsse python server.py或者直接把mcp.run(transportsse)写死。看到类似Uvicorn running on http://0.0.0.0:8000就说明 SSE 服务起来了。终端 B 跑客户端python client_sse.py连接成功后同样输入写文件指令效果和 stdio 一致。区别在于此时服务端是常驻的你可以再开一个终端 B2 同时连上去两个客户端共享同一个服务端进程工具调用互不干扰——这是 SSE 相对 stdio 最实际的优势。如果你想用网页端快速看工具列表可以装 CLI 后跑pip install mcp[cli] mcp dev server.py它会起一个本地调试页面stdio 和 SSE 都能在这里手动触发工具适合排查“工具到底注册上没有”这类问题。6. 本篇常见错排查报错一FileNotFoundError: [Errno 2] No such file or directorystdio原因基本是 server 脚本路径不对。stdio 模式下客户端用command args拉起子进程args[server_script_path]是相对当前工作目录解析的。解决传绝对路径或确认你在 server.py 所在目录执行客户端。报错二Connection refused或ConnectErrorSSE服务端没起或者 URL 写错。SSE 客户端连的是http://0.0.0.0:8000/sse注意末尾的/sse不能少少了会 404。另外0.0.0.0是监听地址客户端连本机建议写http://127.0.0.1:8000/sse跨机器则写服务端实际 IP。报错三TypeError: coroutine object is not subscriptableClientSession(*sse_transport)写成了ClientSession(sse_transport)或者 stdio 下忘了解包self.stdio, self.write stdio_transport。SSE 的 transport 是二元组必须解包。报错四模型不调用工具直接闲聊检查available_tools里parameters字段是否来自t.inputSchema以及工具 docstring 是否为空。docstring 为空时模型没有判断依据会倾向于不调用。另外确认 TaoToken 的 Key 有余额、base_url写的是https://taotoken.net/api而不是官网地址。报错五tool_call_id不匹配第二轮消息里tool_call_id必须和第一轮tool_call.id完全一致name也要和工具名一致。复制粘贴时容易漏改建议直接引用变量而不是手写字符串。报错六SSE 端口被占用换port8001重新起同时客户端 URL 同步改。stdio 模式不存在这个问题因为它不占端口。7. 下一步把 Key 和传输方式固定下来两条链路都跑通之后建议做两件事。第一把 TaoToken 的 Key 从代码里挪到环境变量客户端初始化改成api_keyos.getenv(TAOTOKEN_API_KEY)避免提交到仓库。第二把传输方式也做成配置项本地调试用 stdio部署到服务器用 SSE服务端和客户端各读一个环境变量即可代码零改动。如果你后面要接 Claude Code 或做长期编码 Agent可以看 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 需要新建或轮换 Key 去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 接入细节和字段说明看文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。想先在网页里验证模型能不能正常返回 tool_call用模型对话 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 最快。实测下来stdio 和 SSE 的坑九成集中在连接参数上工具逻辑本身几乎不用改。把第 4 节那两段connect_to_server对照着看几遍比反复重启服务端有效得多。