1. MCP 到底是什么:从 stdio 到 SSE 的 MCPServer 职责拆解
MCP 全称 Model Context Protocol,直译是「模型上下文协议」。你可以把它理解成 AI 大模型的标准化工具箱接口:模型本身只会生成文本,它没法直接读你的数据库、查你的 Git 仓库、调你的内部 API。MCP 就是给模型装上一排标准化的插座,让模型通过统一的消息格式去调用外部能力。
一个 MCPServer 本质上就是一段 Node.js 或者 Python 程序。它对外暴露若干「工具(tool)」「资源(resource)」「提示(prompt)」,模型侧通过 JSON-RPC 2.0 消息和它通信。消息长这样:
{ "jsonrpc": "2.0", "id": 129, "method": "tools/call", "params": { "name": "search_repositories", "arguments": { "query": "user:tech-shrimp" } } }MCPServer 收到这条消息后,用自己的代码逻辑或再发一次外部 API 请求完成任务,然后把结果按同样格式回传。整个过程对模型来说是透明的,它只需要知道「有个叫 search_repositories 的工具,参数是 query」。
传输层目前主流有两种:stdio 和 SSE。stdio 是标准输入输出,客户端把 MCPServer 当子进程启动,通过管道收发消息,适合本地单机场景。SSE 全称 Server-Sent Events,MCPServer 跑成一个 HTTP 服务,客户端通过GET /sse建立长连接接收事件,再通过POST /messages发送请求。SSE 的好处是跨进程、跨机器、可远程部署,也是本文实战的重点。
谁适合读这篇?如果你第一次听说 MCP,想搞清楚「协议是什么、Server 负责什么」,并且想亲手跑通一个能被本地客户端调用的 SSE 版 MCPServer,那这篇就是给你写的。我会先用 Node.js 写一个最小可用的 SSE Server,再用 Python 的官方 SDK 复刻一遍,最后讲清楚模型调用凭证怎么通过 TaoToken 统一管理,避免每个项目到处散落 Key。
先说清楚职责边界,避免后面踩坑。MCPServer 只负责「暴露能力 + 执行 + 返回结果」,它不负责模型推理,也不负责决定模型该调哪个工具——那是客户端(比如 Cherry Studio、Cline、Claude Code)和模型之间的事。所以调试 MCPServer 时,你要盯的是:连接建没建上、工具列表有没有正确返回、参数有没有按 schema 传进来、返回值格式对不对。这四点搞定了,Server 就算通了。
2. 前置准备:TaoToken 统一 Key 与 API 通道配置
在写 Server 之前,先把模型调用凭证这件事理顺。很多人的痛点是:Node.js 项目一个 Key、Python 项目又一个 Key、客户端里再填一个,改起来到处找。TaoToken 的思路是提供一个统一的 API 通道,你只需要维护一份 Key,各个项目和客户端都指向同一个 Base URL。
TaoToken 官网是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 入口是 https://taotoken.net/api (这个地址不加 UTM 参数,直接用于代码里的 Base URL)。
第一步,登录后在控制台创建 API Key。控制台地址带归因参数:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 。创建完把 Key 复制出来,形如sk-xxxx,先存到环境变量里,别硬编码进代码。
第二步,如果你用的是 Claude Code 这类编码工具,可以直接用它的接入文档配置:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。文档里会给出 Base URL 和 Key 的填法。
第三步,验证 Key 是否可用。最直接的方式是打开模型对话页面发一条消息:https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 。能正常返回就说明 Key 和通道没问题。
这里要强调一个概念:TaoToken 是模型调用的统一入口,不是 MCPServer 本身。MCPServer 里如果某个工具需要调用大模型(比如做摘要、做意图识别),它就可以用这份统一 Key 去请求 TaoToken 的 API 通道,而不是自己再维护一套凭证。这样你的 Server 代码里只需要读一个环境变量,干净很多。
把 Key 写进环境变量,Linux/macOS 下:
export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"Windows PowerShell:
$env:TAOTOKEN_API_KEY="sk-你的key" $env:TAOTOKEN_BASE_URL="https://taotoken.net/api"后面 Node.js 和 Python 的示例都会从这两个环境变量读取,保证凭证不落盘、不进 Git。如果你打算长期跑编码类 Agent,也可以了解下 Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ,它更适合高频调用场景。
3. Node.js 实现 SSE 版 MCPServer:可复制配置与启动
先建项目结构。我习惯这样组织:
mcp-node-sse/ ├── package.json ├── tsconfig.json └── src/ └── server.ts初始化并装依赖:
mkdir mcp-node-sse && cd mcp-node-sse npm init -y npm install @modelcontextprotocol/sdk express zod npm install -D typescript tsx @types/express @types/nodepackage.json里加上启动脚本和 ESM 声明:
{ "name": "mcp-node-sse", "version": "1.0.0", "type": "module", "scripts": { "dev": "tsx src/server.ts", "build": "tsc", "start": "node dist/server.js" }, "dependencies": { "@modelcontextprotocol/sdk": "^1.0.0", "express": "^4.19.2", "zod": "^3.23.8" }, "devDependencies": { "typescript": "^5.5.0", "tsx": "^4.16.0", "@types/express": "^4.17.21", "@types/node": "^20.14.0" } }tsconfig.json:
{ "compilerOptions": { "target": "ES2022", "module": "ES2022", "moduleResolution": "bundler", "outDir": "dist", "rootDir": "src", "strict": true, "esModuleInterop": true, "skipLibCheck": true }, "include": ["src"] }核心的src/server.ts,用官方 SDK 的McpServer加SSEServerTransport:
import express from "express"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; import { z } from "zod"; const app = express(); const PORT = 3001; // 每个会话对应一个 transport const transports = new Map<string, SSEServerTransport>(); function buildServer() { const server = new McpServer({ name: "demo-sse-server", version: "1.0.0", }); // 工具一:加法 server.tool( "add", "计算两个数字之和", { a: z.number(), b: z.number() }, async ({ a, b }) => ({ content: [{ type: "text", text: String(a + b) }], }) ); // 工具二:调用 TaoToken 通道做文本摘要(示例) server.tool( "summarize", "对给定文本做一句话摘要", { text: z.string() }, async ({ text }) => { const apiKey = process.env.TAOTOKEN_API_KEY; const baseUrl = process.env.TAOTOKEN_BASE_URL ?? "https://taotoken.net/api"; if (!apiKey) { return { content: [{ type: "text", text: "缺少 TAOTOKEN_API_KEY" }] }; } const resp = await fetch(`${baseUrl}/v1/chat/completions`, { method: "POST", headers: { "Content-Type": "application/json", Authorization: `Bearer ${apiKey}`, }, body: JSON.stringify({ model: "gpt-4o-mini", messages: [ { role: "system", content: "用一句话总结用户输入。" }, { role: "user", content: text }, ], }), }); const data = await resp.json(); const summary = data?.choices?.[0]?.message?.content ?? "无结果"; return { content: [{ type: "text", text: summary }] }; } ); return server; } // SSE 连接入口 app.get("/sse", async (req, res) => { const transport = new SSEServerTransport("/messages", res); transports.set(transport.sessionId, transport); res.on("close", () => transports.delete(transport.sessionId)); const server = buildServer(); await server.connect(transport); }); // 客户端 POST 消息入口 app.post("/messages", express.json(), async (req, res) => { const sessionId = req.query.sessionId as string; const transport = transports.get(sessionId); if (!transport) { res.status(400).send("unknown session"); return; } await transport.handlePostMessage(req, res); }); app.listen(PORT, () => { console.log(`MCP SSE server on http://localhost:${PORT}/sse`); });启动:
npm run dev看到MCP SSE server on http://localhost:3001/sse就说明服务起来了。注意 SSE 的路径是/sse,消息回传路径是/messages,这两个路径客户端配置时要对应上,写错一个就连不通。
4. Python 复刻与 SSE 连通性验证:从启动到工具调用成功
Python 侧用官方mcp包,先装环境。推荐用 uv 管理:
pip install uv uv init mcp-python-sse cd mcp-python-sse uv add "mcp[cli]" httpxserver.py:
import os import httpx from mcp.server.fastmcp import FastMCP mcp = FastMCP("demo-python-sse", host="0.0.0.0", port=3002) @mcp.tool() def add(a: int, b: int) -> int: """计算两个数字之和""" return a + b @mcp.tool() async def summarize(text: str) -> str: """对给定文本做一句话摘要,走 TaoToken 统一通道""" api_key = os.environ.get("TAOTOKEN_API_KEY") base_url = os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api") if not api_key: return "缺少 TAOTOKEN_API_KEY" async with httpx.AsyncClient(timeout=30) as client: resp = await client.post( f"{base_url}/v1/chat/completions", headers={"Authorization": f"Bearer {api_key}"}, json={ "model": "gpt-4o-mini", "messages": [ {"role": "system", "content": "用一句话总结用户输入。"}, {"role": "user", "content": text}, ], }, ) data = resp.json() return data["choices"][0]["message"]["content"] if __name__ == "__main__": mcp.run(transport="sse")启动:
uv run server.py默认监听 3002 端口,SSE 入口是http://localhost:3002/sse。
现在验证连通性。先确认 SSE 端点能建立连接:
curl -N http://localhost:3002/sse正常的话你会看到类似event: endpoint加一行data: /messages/?session_id=xxxx的输出,说明长连接建立成功,并且服务端告诉了你回传消息的地址。
接着在客户端里配置。以 Cherry Studio 为例,添加一个 MCP 工具,类型选 SSE,URL 填http://localhost:3002/sse。保存后客户端会自动拉取工具列表,你应该能看到add和summarize两个工具。
然后让模型调用add,输入「帮我算 3 加 5」。模型会发起tools/call,参数{"a":3,"b":5},Server 返回8。如果客户端里能看到工具调用记录和返回结果,验收就通过了。
Node.js 版同理,把 URL 换成http://localhost:3001/sse即可。两个版本我都跑过,Python 版启动更快,Node.js 版在需要复用前端生态时更顺手,按你的技术栈选。
5. 常见报错排查:401、local proxy failed 与 reading choices
调试阶段最容易卡在几个固定报错上,我按实际遇到的整理一遍。
401 Unauthorized。这个几乎都是 Key 的问题。检查三处:环境变量有没有真的 export 成功(echo $TAOTOKEN_API_KEY看输出)、请求头是不是Authorization: Bearer sk-xxx格式、Base URL 有没有多写或少写/v1。TaoToken 的 API 入口是https://taotoken.net/api,拼 chat completions 时是/api/v1/chat/completions,别把/v1漏了。
local proxy failed / connection refused。客户端连不上 Server,先确认服务真的在跑(curl -N http://localhost:3002/sse有没有输出)。如果 Server 跑在容器或远程机器上,localhost在客户端侧指向的是客户端自己,要换成实际 IP。另外 SSE 路径写错也会报这个,比如把/sse写成/sse/,某些客户端不兼容尾斜杠。
reading 'choices' of undefined。这是解析响应时data.choices不存在。原因通常是:请求根本没成功(返回的是错误对象),或者模型名写错导致返回结构不同。加一层防御:
if (!resp.ok) { const errText = await resp.text(); return { content: [{ type: "text", text: `请求失败 ${resp.status}: ${errText}` }] }; } const data = await resp.json(); const summary = data?.choices?.[0]?.message?.content ?? "无结果";Python 侧同理,先判断resp.status_code。
OAuth / 认证相关报错。如果你用的是 Claude Code 这类工具,配置时出现 OAuth 提示,说明它默认走了自己的登录流程。这时候要在配置里显式指定 Base URL 和 Key,参考接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。Claude Code 的配置通常涉及三件套:Base URL、API Key、Model ID,缺一不可。
工具列表为空。连接建上了但看不到工具,多半是server.tool()注册时 schema 写错,或者工具名重复。检查启动日志有没有报错,把 zod schema 简化成最基础的类型先跑通,再逐步加复杂参数。
SSE 连接频繁断开。长连接对超时敏感,某些反向代理会掐掉空闲连接。本地调试一般没事,如果部署到服务器,记得给 SSE 路径关掉缓冲、拉长超时。
排查顺序建议固定成:先 curl 确认 SSE 端点活着,再确认客户端 URL 和路径,再看 Key 和 Base URL,最后看响应解析。按这个顺序走,九成问题能定位。
6. 把统一 Key 用起来:模型对话、接入文档与 Coding Plan 分流
Server 跑通之后,真正省心的地方在于凭证管理。你不需要在每个 MCPServer 里各写一套 Key,而是统一从环境变量读 TaoToken 的 Key 和 Base URL。这样换 Key、换通道只改一处,所有 Server 和客户端一起生效。
如果你只是想快速验证某个模型能不能用,直接开模型对话页面发消息最快:https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 。想管理或新建 Key 就去控制台:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 。需要看具体接入参数和示例代码,翻文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。Key 列表页在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite ,方便你集中查看和轮换。
如果你主要在写代码、跑 Agent,调用频率高,Coding Plan 会更合适:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。而 Claude Code 用户可以直接看专门的接入说明:https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude-code-anthropic&utm_campaign=rewrite 。
最后留一个实用技巧:把TAOTOKEN_API_KEY和TAOTOKEN_BASE_URL写进项目的.env文件,用dotenv加载,同时把.env加进.gitignore。这样本地开发方便,又不会把 Key 提交上去。Node.js 里import "dotenv/config"一行搞定,Python 里from dotenv import load_dotenv; load_dotenv()即可。凭证管理这件事,越早规范越省事。