1. 为什么我要手写一个 MCP Server
MCP Server 说白了就是给大模型装"手"的进程：模型想查数据库、调内部 API、读本地文件，都通过它暴露的 Tools 和 Resources 完成。现在很多人用npx mcp-server-xxx一把梭，本地跑得挺欢，一旦要排查线上问题、或者把公司内部系统封装成 MCP Server，就抓瞎了——因为不知道 JSON-RPC 消息长什么样、stdio 和 Streamable HTTP 到底差在哪、握手失败该看哪一行日志。
这篇就干一件事：不依赖任何 MCP 框架，用标准库把协议跑通。你会看到 JSON-RPC 2.0 的四个字段怎么在 stdio 和 HTTP 两条传输层上流动，然后手写一个能用的 MCP Server 和一个最小 Client，最后用 curl 验证 Streamable HTTP 会话。适合需要自建 MCP Server 并接入 AI 工具的开发者，尤其是想把内部系统安全暴露给 Agent 的那批人。
我试过直接照官方 SDK 抄，结果被"日志污染协议流"坑了一下午，所以下面会把踩过的坑单独拎出来讲。
2. TaoToken 统一 Key 的前置准备
自建 MCP Server 之后，你大概率要把它接到某个 Host（Cursor、Claude Code、自研 Agent）上，而 Host 侧调用模型需要 Key。如果每个工具、每个环境各配一套 Key，管理成本会爆炸。TaoToken 的思路是统一入口：一个 Key 覆盖模型对话、Coding Plan、API 调用，MCP Server 侧只需要在配置里引用同一个环境变量即可。
你需要先拿到 Key，入口在这里：
- 控制台（创建/管理 Key）：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite
- API Keys 页面：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite
- 接入文档（协议与端点说明）：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
API 基地址统一用https://taotoken.net/api（注意这个地址不带 UTM 参数，写进代码里就用它）。Key 建议放环境变量，别硬编码：
export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"注意：MCP Server 本身不直接调模型，它只暴露工具；真正调模型的是 Host。所以 Key 配在 Host 侧，Server 侧只在需要回调模型（比如工具内部做二次推理）时才用得上。这个边界先分清，后面配置才不会乱。
3. 可复制配置：config.toml 与 settings.json 骨架
不同 Host 的配置文件格式不一样，这里给两份最常用的骨架。核心都是三件事：启动命令、环境变量、传输方式。
先看config.toml（适合自研 Host 或支持 TOML 的工具）：
[mcp] # 传输方式：stdio 或 streamable-http transport = "stdio" [mcp.server.minimal] command = "python3" args = ["/opt/mcp/minimal_mcp_server.py"] env = { TAOTOKEN_API_KEY = "${TAOTOKEN_API_KEY}", TAOTOKEN_BASE_URL = "https://taotoken.net/api" } [mcp.server.remote] transport = "streamable-http" url = "http://127.0.0.1:8765/mcp" headers = { Authorization = "Bearer ${TAOTOKEN_API_KEY}" }再看settings.json（Claude Code / Cursor 这类 Host 常用）：
{ "mcpServers": { "minimal": { "command": "python3", "args": ["/opt/mcp/minimal_mcp_server.py"], "env": { "TAOTOKEN_API_KEY": "sk-你的key", "TAOTOKEN_BASE_URL": "https://taotoken.net/api" } }, "remote": { "type": "streamable-http", "url": "http://127.0.0.1:8765/mcp", "headers": { "Authorization": "Bearer sk-你的key" } } } }参数对照表，方便你按需改：
| 字段 | 作用 | stdio 必填 | HTTP 必填 |
|---|---|---|---|
| command | 启动 Server 的可执行文件 | 是 | 否 |
| args | 启动参数数组 | 是 | 否 |
| env | 注入进程的环境变量 | 是 | 否 |
| url | Streamable HTTP 端点 | 否 | 是 |
| headers | 鉴权/会话头 | 否 | 是 |
| transport/type | 传输类型标识 | 是 | 是 |
提示：
env里引用${TAOTOKEN_API_KEY}是否生效取决于 Host 是否支持变量展开。不确定就直接写值，但别把带 Key 的配置文件提交到 Git。
4. 手写 MCP Server：从 JSON-RPC 到 stdio
协议层所有消息都是 JSON-RPC 2.0，结构永远是jsonrpc / id / method / params四件套。握手消息长这样：
{ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {"tools": {"listChanged": true}}, "clientInfo": {"name": "my-host", "version": "1.0.0"} } }下面是不依赖任何框架的 stdio 版 Server，核心就三点：stdout 发消息、stderr 打日志、按 method 分发。
#!/usr/bin/env python3 """minimal_mcp_server.py — 纯标准库实现的 MCP Server（stdio 传输）""" import json import sys from typing import Any def send(msg: dict) -> None: """MCP 走 stdout，每行一个 JSON 对象""" sys.stdout.write(json.dumps(msg, ensure_ascii=False) + "\n") sys.stdout.flush() def log(msg: str) -> None: """调试日志必须走 stderr，不能污染协议流""" sys.stderr.write(f"[server] {msg}\n") TOOLS = [ { "name": "add", "description": "计算两个整数之和", "inputSchema": { "type": "object", "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}}, "required": ["a", "b"], }, }, { "name": "get_discount", "description": "查询商品今日折扣", "inputSchema": { "type": "object", "properties": {"sku": {"type": "string"}}, "required": ["sku"], }, }, ] def call_tool(name: str, args: dict) -> Any: if name == "add": return {"result": args["a"] + args["b"]} if name == "get_discount": # 真实场景这里会查数据库/调内部 API return {"sku": args["sku"], "discount": 0.85} raise ValueError(f"unknown tool: {name}") def handle(msg: dict) -> None: method = msg.get("method") mid = msg.get("id") if method == "initialize": send({ "jsonrpc": "2.0", "id": mid, "result": { "protocolVersion": "2025-06-18", "capabilities": {"tools": {}}, "serverInfo": {"name": "minimal-server", "version": "0.1.0"}, }, }) elif method == "notifications/initialized": log("client initialized, ready") elif method == "tools/list": send({"jsonrpc": "2.0", "id": mid, "result": {"tools": TOOLS}}) elif method == "tools/call": params = msg.get("params", {}) try: r = call_tool(params["name"], params.get("arguments", {})) send({ "jsonrpc": "2.0", "id": mid, "result": { "content": [{"type": "text", "text": json.dumps(r, ensure_ascii=False)}], "isError": False, }, }) except Exception as e: send({ "jsonrpc": "2.0", "id": mid, "result": { "content": [{"type": "text", "text": str(e)}], "isError": True, }, }) elif method == "ping": send({"jsonrpc": "2.0", "id": mid, "result": {}}) else: log(f"unhandled method: {method}") if __name__ == "__main__": for line in sys.stdin: line = line.strip() if not line: continue handle(json.loads(line))一个能跑的 MCP Server 骨架就是这么薄。注意notifications/initialized没有id，它是通知不是请求，Server 不需要回包。
5. 最小 Client：理解 Host 侧的握手
再看 Host 侧怎么跟 Server 对话。一个最小 Client 需要：启动子进程 →initialize→ 发initialized通知 → 调工具。
#!/usr/bin/env python3 """minimal_mcp_client.py — 最小 MCP Client，演示完整握手""" import json import subprocess import sys proc = subprocess.Popen( [sys.executable, "minimal_mcp_server.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, ) def request(method: str, params: dict, mid: int) -> dict: proc.stdin.write(json.dumps( {"jsonrpc": "2.0", "id": mid, "method": method, "params": params} ) + "\n") proc.stdin.flush() return json.loads(proc.stdout.readline()) # 1. 握手：协商协议版本与能力 resp = request("initialize", { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "minimal-client", "version": "0.1.0"}, }, mid=1) assert resp["result"]["protocolVersion"] == "2025-06-18" print("握手成功:", resp["result"]["serverInfo"]) # 2. 通知 Server 初始化完成（通知没有 id） proc.stdin.write(json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}) + "\n") proc.stdin.flush() # 3. 拉取工具清单 tools = request("tools/list", {}, mid=2)["result"]["tools"] print("工具:", [t["name"] for t in tools]) # 4. 调用工具 r = request("tools/call", {"name": "add", "arguments": {"a": 40, "b": 2}}, mid=3) print("add(40,2) =", r["result"]["content"][0]["text"])跑起来输出：
握手成功: {'name': 'minimal-server', 'version': '0.1.0'} 工具: ['add', 'get_discount'] add(40,2) = {"result": 42}整个协议没有魔法，就是协商 → 通知 → 请求/响应三次交互，和普通 RPC 没有本质区别。
6. Streamable HTTP：无状态化的关键设计
本地用 stdio，云端就得上 HTTP。2025-06-18 规范把 SSE 升级为 Streamable HTTP：普通请求走 POST 立即返回 JSON；需要流式时服务端用Content-Type: text/event-stream推事件，客户端拿到sessionId后在后续请求头里带上Mcp-Session-Id保持会话。
生产环境最关键的一条：POST 请求必须幂等、无状态，这样前面挂多少个 Nginx/LB 都不怕。典型请求长这样：
POST /mcp HTTP/1.1 Host: mcp.example.com Content-Type: application/json Accept: application/json, text/event-stream Mcp-Session-Id: a1b2c3d4e5 {"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_discount","arguments":{"sku":"SKU-001"}}}服务端流式响应：
HTTP/1.1 200 OK Content-Type: text/event-stream event: message data: {"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"{\"sku\": \"SKU-001\", \"discount\": 0.85}"}]}}用 curl 验证握手和会话，先发 initialize：
curl -i -X POST http://127.0.0.1:8765/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"curl","version":"1.0"}}}'响应头里会带Mcp-Session-Id，把它记下来，后续请求带上：
curl -i -X POST http://127.0.0.1:8765/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -H "Mcp-Session-Id: a1b2c3d4e5" \ -d '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}'如果返回text/event-stream，你会看到event: message加一行data:，这就是流式通道在工作。这也是为什么 MCP 网关（Gateway）在企业里成为标配——它把散落的 stdio Server 统一转换成 HTTP 出口，还能顺手做鉴权、限流、审计。
7. 本篇常见错排查
协议污染：stdio 模式下任何多余输出（print调试、第三方库的日志）都会让 Client 解析崩溃。所有日志走 stderr，这是线上事故第一高发点。我踩过的坑就是某个依赖库默认往 stdout 打 banner，排查了半天。
超时不泄漏：流式模式下 SSE 长连接要设置空闲超时，Client 用完必须释放，否则服务端连接数只涨不跌。典型的生产事故，建议在网关层加连接数上限。
握手版本不匹配：Client 发2025-06-18，Server 回了个旧版本，assert直接挂。排查时先看initialize的响应体，别急着看工具逻辑。
Session 丢失：Streamable HTTP 下忘了带Mcp-Session-Id，服务端会当成新会话，工具状态全丢。curl 验证时务必把响应头里的 session id 复制到下一个请求。
安全边界：MCP 统一了"接线"却不会自动装"保险丝"。生产环境必须做到最小权限账号、只读优先、root 目录限定、OAuth Token 短期化、高危操作默认禁止。工具内部如果要回调模型做二次推理，Key 从环境变量读，别写进代码。
8. 接入与验证：把 Server 挂到 Host 上
Server 跑通后，把它挂到 Host 上验证。stdio 版直接把settings.json里的command/args指向你的脚本；HTTP 版填url和headers。验证模型侧是否正常，可以用模型对话页面发一条消息，确认 Host 能列出你的工具：
- 模型对话（验证工具是否被正确识别）：https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=chat&utm_campaign=rewrite
- Coding Plan（长期编码/Agent 场景，统一 Key 覆盖）：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite
- 接入文档（协议细节与端点）：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
如果你在 Claude Code 里接，Anthropic 兼容入口在这里：https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude-code-anthropic&utm_campaign=rewrite
排障时优先看 API Keys 和接入文档两页，Key 权限、端点格式、协议版本对不上，九成问题都出在这。
