构建基于 SSE 协议通信的 MCP Server 和 Client：把 endpoint 改到 TaoToken

1. 从 stdio 到 SSEMCP 通信为什么要换条路如果你跟着前面的教程写过 MCP Server大概率用的是 stdio 模式客户端subprocess拉起服务器进程双方通过 stdin/stdout 交换 JSON-RPC 消息。这个模式跑本地小工具没问题但一旦你想让多个客户端共用同一个 Server或者把 Server 放到另一台机器上stdio 就卡住了——它天生是 1:1 的进程耦合客户端一退出服务器进程也跟着没了。SSEServer-Sent Events解决的正是这个解耦问题。它基于普通 HTTP服务器通过一条长连接持续向客户端推送事件流客户端则用 POST 把消息发回去。MCP 官方 SDK 把这条链路封装成了SseServerTransport你只需要挂两个路由/sse负责建立事件流/messages/负责接收客户端上行消息。服务器从此变成一个独立常驻进程谁都能连连完随时断互不影响。这篇文章要做的是在这套 SSE 链路上把 endpoint 统一改到 TaoToken 的 API 通道。适合谁看已经写过 stdio 版 MCP Server、想升级成多客户端架构的开发者或者手里有一堆本地工具查数据库、读文件、调内部接口想通过一个统一入口暴露给 LLM 的人。核心检索词就三个SSE、MCP Server、MCP Client全文围绕这条通信链路展开每一步都能复制去跑。我试过把 Server 和 Client 分别跑在两台机器上中间只靠一个 Base URL 串起来实测下来只要握手参数对消息往返完全稳定。下面从环境准备开始一步步把链路搭起来最后用 curl 验证握手是否真的成功。2. TaoToken 前置准备把 API 通道接进 MCP 链路在动手写 SSE 代码之前先把 TaoToken 这条 API 通道准备好。MCP Server 本身不直接调模型它负责暴露工具真正需要模型能力的是 MCP Client 那一侧——当 Client 把用户问题转成工具调用、或者把工具返回结果整理成自然语言时背后要有一个兼容 OpenAI 接口的模型服务。TaoToken 提供的正是这个统一入口Base URL 固定为https://taotoken.net/api模型 ID 按你实际开通的填。第一步去控制台拿 Key。打开https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite登录后在 API Keys 页面创建一个新 Key复制出来形如sk-xxxx的字符串。这个 Key 只显示一次建议先存到密码管理器里。如果你还没决定用哪个模型可以先去模型对话页面https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite试几条 prompt确认响应风格符合预期再写进配置。第二步把 Key 写进环境变量别硬编码在代码里。在项目根目录建一个.env文件# .env TAOTOKEN_API_KEYsk-你的真实key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_IDgpt-4o-mini这里TAOTOKEN_MODEL_ID填你在控制台看到的可用模型 ID不同账号开通的模型可能不一样以实际列表为准。用python-dotenv加载from dotenv import load_dotenv import os load_dotenv() API_KEY os.getenv(TAOTOKEN_API_KEY) BASE_URL os.getenv(TAOTOKEN_BASE_URL) MODEL_ID os.getenv(TAOTOKEN_MODEL_ID)第三步确认 Python 环境和依赖。用 uv 管理项目最省事uv init mcp-sse-demo cd mcp-sse-demo uv add mcp[cli] elasticsearch python-dotenv uvicorn openai starlettemcp[cli]里已经带了SseServerTransport和sse_client不用额外装。starlette和uvicorn负责把 SSE 服务跑起来openai用来在 Client 侧调 TaoToken 的兼容接口。第四步如果你要复现 Elasticsearch 那套工具用 docker-compose 起一个本地集群docker-compose up -d浏览器打开http://localhost:5601进 Kibana用户名elastic密码test123。然后在 Dev Tools 里建两个索引并插数据POST /student/_doc { name: Alice, age: 20, major: Computer Science } POST /teacher/_doc { name: Tom, subject: English, yearsOfExperience: 10 }数据不用多几条就够验证工具调用是否真的打到了后端。到这里TaoToken 的 Key、Base URL、Model ID 三件套齐了Elasticsearch 也活了可以进入 SSE 服务端的配置环节。3. 可复制配置SSE Server 与 Client 的完整参数这一节给出能直接复制运行的配置片段。先看 Server 侧核心是用SseServerTransport把 MCP 服务器挂到 Starlette 上。新建server.pyimport argparse import uvicorn from mcp.server import Server from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Route, Mount from starlette.requests import Request def create_starlette_app(mcp_server: Server, *, debug: bool False) - Starlette: sse SseServerTransport(/messages/) async def handle_sse(request: Request) - None: async with sse.connect_sse( request.scope, request.receive, request._send, ) as (read_stream, write_stream): await mcp_server.run( read_stream, write_stream, mcp_server.create_initialization_options(), ) return Starlette( debugdebug, routes[ Route(/sse, endpointhandle_sse), Mount(/messages/, appsse.handle_post_message), ], ) if __name__ __main__: mcp_server Server(demo-server) parser argparse.ArgumentParser(descriptionRun MCP SSE-based server) parser.add_argument(--host, default0.0.0.0) parser.add_argument(--port, typeint, default18080) args parser.parse_args() app create_starlette_app(mcp_server, debugTrue) uvicorn.run(app, hostargs.host, portargs.port)两个路由的分工要记牢/sse是客户端建立事件流的入口/messages/是客户端 POST 上行消息的入口。SseServerTransport(/messages/)里的路径必须和Mount的路径一致否则客户端发消息会 404。再看 Client 侧关键是sse_client的 URL 要指向 Server 的/sse端点同时把模型调用指向 TaoToken。新建client.pyimport asyncio import sys from contextlib import AsyncExitStack from mcp import ClientSession from mcp.client.sse import sse_client from openai import AsyncOpenAI import os from dotenv import load_dotenv load_dotenv() class MCPClient: def __init__(self): self.session None self.exit_stack AsyncExitStack() self.llm AsyncOpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), ) async def connect_to_sse_server(self, server_url: str): self._streams_context sse_client(urlserver_url) streams await self._streams_context.__aenter__() self._session_context ClientSession(*streams) self.session await self._session_context.__aenter__() await self.session.initialize() async def list_tools(self): response await self.session.list_tools() return response.tools async def cleanup(self): await self._session_context.__aexit__(None, None, None) await self._streams_context.__aexit__(None, None, None) async def main(): if len(sys.argv) 2: print(Usage: uv run client.py http://localhost:18080/sse) sys.exit(1) client MCPClient() try: await client.connect_to_sse_server(sys.argv[1]) tools await client.list_tools() print(Connected to server with tools:, [t.name for t in tools]) finally: await client.cleanup() if __name__ __main__: asyncio.run(main())如果你用 Claude Code 或 Cline 这类工具接入配置写法不一样。以 Claude Code 的settings.json为例MCP Server 的 SSE 接入片段{ mcpServers: { taotoken-sse-demo: { type: sse, url: http://localhost:18080/sse, env: { TAOTOKEN_API_KEY: sk-你的真实key, TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_MODEL_ID: gpt-4o-mini } } } }Cline 的 MCP 配置在cline_mcp_settings.json里结构类似把type设为sse、url指向/sse即可。Codex 的auth.json则是另一套格式如果你用 Codex CLI把 Base URL 和 Key 写进~/.codex/auth.json{ api_key: sk-你的真实key, base_url: https://taotoken.net/api }三件套Base URL Key Model ID在任何一种接入方式里都不能少缺一个就会在握手或首次调用时报错。配置写完后先别急着跑 Client用下一节的 curl 确认 Server 的握手是否正常。4. 验证请求用 curl 确认握手与消息往返Server 启动命令uv run server.py --host 0.0.0.0 --port 18080看到Uvicorn running on http://0.0.0.0:18080就说明起来了。先验证/sse端点是否返回事件流。开一个终端执行curl -N -H Accept: text/event-stream http://localhost:18080/sse-N关闭缓冲让你实时看到推送。正常输出类似event: endpoint data: /messages/?session_idxxxxxxxxxxxxxxxx这行endpoint事件就是握手成功的标志它告诉客户端后续上行消息要 POST 到哪个地址。session_id是本次连接的唯一标识每个客户端连上来都会拿到不同的值。注意 curl 会一直挂着不退出这是 SSE 长连接的特性按 CtrlC 结束即可。拿到session_id后验证消息往返。另开一个终端用 POST 发一条 JSON-RPC 初始化请求curl -X POST http://localhost:18080/messages/?session_idxxxxxxxxxxxxxxxx \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:curl-test,version:1.0}}}如果 Server 正常第一个终端挂着/sse的那个会实时打印出响应事件event: message data: {jsonrpc:2.0,id:1,result:{protocolVersion:2024-11-05,capabilities:{tools:{}},serverInfo:{name:demo-server,version:1.0}}}看到result里有serverInfo说明握手和消息往返都通了。这一步很关键——很多人 Client 连不上其实是 Server 的/messages/路由没挂对用 curl 能快速定位是 Server 的问题还是 Client 的问题。再验证工具列表。发一条tools/listcurl -X POST http://localhost:18080/messages/?session_idxxxxxxxxxxxxxxxx \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:2,method:tools/list,params:{}}第一个终端应该打印出工具数组包含你在 Server 里注册的list_indices、get_index等。到这一步SSE 链路本身已经确认跑通剩下的就是 Client 侧把模型调用接上 TaoToken。启动 Clientuv run client.py http://localhost:18080/sse预期输出Connected to server with tools: [list_indices, get_index]如果这里报连接错误回到 curl 那步确认 Server 是否还在跑、session_id是否过期。SSE 连接断开后session_id会失效Client 每次重连都会拿新的。5. 常见报错排查401、local proxy failed 与 reading choices链路跑通之前大概率会撞上几个典型报错。这一节按真实错误信息对照排查。401 Unauthorized。这个几乎都出在 TaoToken 的 Key 上。检查.env里TAOTOKEN_API_KEY是否完整、有没有多余空格确认TAOTOKEN_BASE_URL是https://taotoken.net/api末尾不要多加/v1或斜杠。如果 Key 是从控制台复制的注意别把前后引号也带进去。还有一种情况是 Key 被禁用或额度耗尽去控制台 API Keys 页面确认状态。local proxy failed / connection refused。这个报错通常出现在 Client 连 Server 的阶段不是连 TaoToken。检查三件事Server 是否真的在18080端口监听lsof -i:18080Client 传的 URL 是不是http://localhost:18080/sse路径必须是/sse不是/messages/如果 Server 和 Client 不在同一台机器localhost要换成 Server 的实际 IP且 Server 启动时--host要是0.0.0.0而不是127.0.0.1。Error reading choices / choices 字段为空。这个报错来自模型调用侧说明 TaoToken 返回的响应结构里没有choices。常见原因有两个一是TAOTOKEN_MODEL_ID填错了填了一个不存在的模型名接口返回错误对象而不是正常 completion二是请求体格式不对比如把messages写成了prompt。用openaiSDK 时确认调用方式是response await self.llm.chat.completions.create( modelos.getenv(TAOTOKEN_MODEL_ID), messages[{role: user, content: 你好}], ) print(response.choices[0].message.content)如果这行单独跑都报reading choices那就是模型 ID 或 Key 的问题跟 MCP 无关。OAuth / 认证跳转异常。有些 MCP 客户端在连 SSE 时会尝试走 OAuth 流程如果你的 Server 没配认证客户端可能卡在跳转。检查客户端配置里是否误开了 OAuth 选项Claude Code 的settings.json里 SSE 类型不需要 OAuth 字段删掉即可。Cline 的 MCP 配置同理type: sse只认url和env。session_id 失效导致 404。SSE 连接断开后旧session_id作废如果 Client 重连逻辑没更新 sessionPOST 到/messages/会返回 404。sse_client内部会自动处理重连并拿新 session但如果你自己手写了 HTTP 请求需要监听endpoint事件动态更新 URL。工具调用返回空结果。链路通了但工具没返回数据先确认 Elasticsearch 是否在跑、.env里的ELASTIC_HOST是否正确。用 curl 直接打 EScurl -k -u elastic:test123 https://localhost:9200/_cat/indices能看到student和teacher两个索引说明后端没问题问题在 MCP 工具实现里。排查顺序建议固定先 curl 验/sse握手再 curl 验/messages/往返再跑 Client 验工具列表最后验模型调用。每一步单独确认比一上来就跑完整链路容易定位。6. 把 endpoint 固定到 TaoToken长期编码与 Agent 场景的接入建议链路跑通之后真正要长期用的是把 endpoint 稳定指向 TaoToken让 MCP Client 每次启动都走同一条 API 通道。这里给几个实操建议。第一把三件套集中管理。不要在每个脚本里重复写 Base URL 和 Key统一放.env用load_dotenv()加载。如果团队多人协作.env加进.gitignore另建.env.example提交模板# .env.example TAOTOKEN_API_KEYsk-your-key-here TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_IDgpt-4o-mini第二Client 侧的重连参数要显式配置。sse_client默认会自动重连但重连间隔和超时可以通过参数调整。如果你跑的是长时间 Agent 任务建议把超时设长一点避免网络抖动导致会话中断self._streams_context sse_client( urlserver_url, timeout30, sse_read_timeout300, )sse_read_timeout控制读取事件流的等待时间Agent 任务里工具执行可能耗时较久设成 300 秒比较稳妥。第三如果你用 Coding Plan 跑长期编码任务MCP Server 的工具列表最好按需精简。工具太多会让模型在tools/list阶段消耗大量 token而且容易选错工具。把高频工具留下低频的拆到独立 Server按需连接。Coding Plan 的入口在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite适合需要持续跑 Agent、对调用稳定性有要求的场景。第四接入文档放在手边。TaoToken 的接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite里面列了各语言 SDK 的 Base URL 写法和常见错误码。遇到 401 或模型 ID 问题时先翻文档比盲目改代码快。第五API Keys 定期轮换。在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite页面可以创建多个 Key给不同项目分配不同 Key方便按项目排查调用来源也方便某个 Key 泄露时单独吊销而不影响其他服务。最后说一个实际踩过的坑SSE 连接在容器环境里容易被负载均衡器掐断因为默认空闲超时可能只有 60 秒。如果你的 Server 部署在网关后面确认网关的proxy_read_timeout大于sse_read_timeout否则长连接会被静默断开Client 侧表现为工具调用到一半没响应。本地开发没这个问题上生产前记得检查。