企业官网建设流程全解析

1. 从 stdio 到 SSE：MCP 通信为什么要换条路

如果你跟着前面的教程写过 MCP Server，大概率用的是 stdio 模式：客户端subprocess拉起服务器进程，双方通过 stdin/stdout 交换 JSON-RPC 消息。这个模式跑本地小工具没问题，但一旦你想让多个客户端共用同一个 Server，或者把 Server 放到另一台机器上，stdio 就卡住了——它天生是 1:1 的进程耦合，客户端一退出，服务器进程也跟着没了。

SSE（Server-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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite，登录后在 API Keys 页面创建一个新 Key，复制出来形如sk-xxxx的字符串。这个 Key 只显示一次，建议先存到密码管理器里。如果你还没决定用哪个模型，可以先去模型对话页面https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite试几条 prompt，确认响应风格符合预期再写进配置。

第二步，把 Key 写进环境变量，别硬编码在代码里。在项目根目录建一个.env文件：

# .env TAOTOKEN_API_KEY=sk-你的真实key TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL_ID=gpt-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 starlette

mcp[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.py：

import 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( debug=debug, routes=[ Route("/sse", endpoint=handle_sse), Mount("/messages/", app=sse.handle_post_message), ], ) if __name__ == "__main__": mcp_server = Server("demo-server") parser = argparse.ArgumentParser(description="Run MCP SSE-based server") parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=18080) args = parser.parse_args() app = create_starlette_app(mcp_server, debug=True) uvicorn.run(app, host=args.host, port=args.port)

两个路由的分工要记牢：/sse是客户端建立事件流的入口，/messages/是客户端 POST 上行消息的入口。SseServerTransport("/messages/")里的路径必须和Mount的路径一致，否则客户端发消息会 404。

再看 Client 侧，关键是sse_client的 URL 要指向 Server 的/sse端点，同时把模型调用指向 TaoToken。新建client.py：

import 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_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), ) async def connect_to_sse_server(self, server_url: str): self._streams_context = sse_client(url=server_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_id=xxxxxxxxxxxxxxxx

这行endpoint事件就是握手成功的标志，它告诉客户端后续上行消息要 POST 到哪个地址。session_id是本次连接的唯一标识，每个客户端连上来都会拿到不同的值。注意 curl 会一直挂着不退出，这是 SSE 长连接的特性，按 Ctrl+C 结束即可。

拿到session_id后，验证消息往返。另开一个终端，用 POST 发一条 JSON-RPC 初始化请求：

curl -X POST "http://localhost:18080/messages/?session_id=xxxxxxxxxxxxxxxx" \ -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/list：

curl -X POST "http://localhost:18080/messages/?session_id=xxxxxxxxxxxxxxxx" \ -H "Content-Type: application/json" \ -d '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}'

第一个终端应该打印出工具数组，包含你在 Server 里注册的list_indices、get_index等。到这一步，SSE 链路本身已经确认跑通，剩下的就是 Client 侧把模型调用接上 TaoToken。

启动 Client：

uv 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:18080）；Client 传的 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( model=os.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 重连逻辑没更新 session，POST 到/messages/会返回 404。sse_client内部会自动处理重连并拿新 session，但如果你自己手写了 HTTP 请求，需要监听endpoint事件动态更新 URL。

工具调用返回空结果。链路通了但工具没返回数据，先确认 Elasticsearch 是否在跑、.env里的ELASTIC_HOST是否正确。用 curl 直接打 ES：

curl -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_KEY=sk-your-key-here TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL_ID=gpt-4o-mini

第二，Client 侧的重连参数要显式配置。sse_client默认会自动重连，但重连间隔和超时可以通过参数调整。如果你跑的是长时间 Agent 任务，建议把超时设长一点，避免网络抖动导致会话中断：

self._streams_context = sse_client( url=server_url, timeout=30, sse_read_timeout=300, )

sse_read_timeout控制读取事件流的等待时间，Agent 任务里工具执行可能耗时较久，设成 300 秒比较稳妥。

第三，如果你用 Coding Plan 跑长期编码任务，MCP Server 的工具列表最好按需精简。工具太多会让模型在tools/list阶段消耗大量 token，而且容易选错工具。把高频工具留下，低频的拆到独立 Server，按需连接。Coding Plan 的入口在https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite，适合需要持续跑 Agent、对调用稳定性有要求的场景。

第四，接入文档放在手边。TaoToken 的接入文档在https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite，里面列了各语言 SDK 的 Base URL 写法和常见错误码。遇到 401 或模型 ID 问题时先翻文档，比盲目改代码快。

第五，API Keys 定期轮换。在https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite页面可以创建多个 Key，给不同项目分配不同 Key，方便按项目排查调用来源，也方便某个 Key 泄露时单独吊销而不影响其他服务。

最后说一个实际踩过的坑：SSE 连接在容器环境里容易被负载均衡器掐断，因为默认空闲超时可能只有 60 秒。如果你的 Server 部署在网关后面，确认网关的proxy_read_timeout大于sse_read_timeout，否则长连接会被静默断开，Client 侧表现为工具调用到一半没响应。本地开发没这个问题，上生产前记得检查。