1. 为什么我要手搓一个 MCP 服务MCPModel Context Protocol是 Anthropic 提出的开放协议它定义了 AI 客户端Claude Desktop、OpenCode、Trae CN 这类工具和外部数据源之间的标准交互方式。你可以把它理解成「AI 界的 USB-C」客户端只管按协议发请求服务端只管按协议回数据两边不用互相认识。MCP 服务能做什么简单说就是把你的本地文档、数据库、内部 API 包装成 AI 能直接调用的工具和资源。适合谁适合手上有私有数据、想让 AI 真正读进去、又不想把数据传到第三方平台的开发者。我这次的目标很具体用 FastAPI 从零实现一个 MCP 服务走 JSON-RPC 2.0 协议、SSE 传输最后通过 TaoToken 的统一 Key 通道接进 AI 工具跑通一次完整的tools/list和tools/call调用。为什么不用现成 SDK因为我想把协议每一层都摸清楚——消息怎么分发、会话怎么管理、SSE 双通道怎么保活这些只有自己写一遍才真正理解。下面这份记录里你可以直接复制config.toml、settings.json骨架和启动命令跟着做就能跑通。2. TaoToken 前置统一 Key 与 API 通道准备在动手写服务之前先把 AI 侧的接入通道准备好。我选择 TaoToken 作为统一入口原因是它把模型对话、Coding Plan、API Key 管理收敛到一个控制台里MCP 服务调试时不用在多个平台之间来回切 Key。你需要做三件事注册账号、创建 API Key、确认接入文档里的请求格式。2.1 创建 API Key登录控制台后进入 API Keys 页面新建一个 Key 并复制保存。这个 Key 后面会同时用在两个地方一是 MCP 服务自身的 Bearer Token 认证保护你的服务端二是 AI 客户端调用模型时的鉴权。建议开发阶段用两个不同的 Key避免混淆。控制台入口https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentmcp_fastapiutm_campaignrewriteAPI Key 管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentmcp_fastapiutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentmcp_fastapiutm_campaignrewrite2.2 确认 API 通道地址TaoToken 的 API 基础地址是https://taotoken.net/api兼容 OpenAI 风格的/v1/chat/completions调用。MCP 服务本身不直接调模型但你的 AI 客户端比如 OpenCode需要配置这个地址来发对话请求。把下面这段先记下来第 4 节的settings.json会用到# config.toml —— AI 客户端模型通道配置骨架 [model] provider openai-compatible base_url https://taotoken.net/api api_key sk-你的TaoToken密钥 model claude-sonnet-4-20250514如果你打算长期跑编码类 Agent 任务可以了解下 Coding Plan它按周期计费比单次调用更适合高频场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentmcp_fastapiutm_campaignrewrite3. FastAPI 搭建 MCP 服务JSON-RPC 与 SSE 双通道MCP over SSE 的核心架构是「GET 连 SSE 收消息POST 发命令」的双通道模式。服务端只暴露两个路由GET /mcp维持长连接推送事件POST /mcp接收 JSON-RPC 消息。初学者最容易搞混的就是这一点——以为所有通信都走 SSE其实客户端请求是普通 HTTP POST。3.1 项目结构与依赖pip install fastapi uvicorn sse-starlette pydantic目录结构建议这样组织后面排查问题时定位快mcp-server/ ├── server.py # FastAPI 入口 路由 ├── sse_manager.py # SSE 连接与会话管理 ├── handler.py # JSON-RPC 方法分发 ├── uri_parser.py # 资源 URI 解析 └── config.toml # 服务配置3.2 两个核心路由# server.py from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from sse_starlette.sse import EventSourceResponse from sse_manager import SSEConnectionManager from handler import MCPSSEHandler app FastAPI(titleUni-Index MCP Server) sse_manager SSEConnectionManager() handler MCPSSEHandler(sse_manager) app.get(/mcp) async def sse_endpoint(request: Request): # 长连接分配 session_id推送 endpoint 事件 session_id sse_manager.create_session() async def event_generator(): yield {event: endpoint, data: f/mcp?session_id{session_id}} async for msg in sse_manager.listen(session_id): yield {event: message, data: msg} return EventSourceResponse(event_generator()) app.post(/mcp) async def jsonrpc_endpoint(request: Request, session_id: str None): body await request.json() if session_id and session_id in sse_manager.active_sessions: # SSE 模式异步处理通过 SSE 推送响应 import asyncio asyncio.create_task(handler.process_sse_request(session_id, body)) return {status: processing, session_id: session_id} # 无状态模式直接返回 JSON-RPC 响应 return await handler.handle_message(body, session_id, sse_manager) app.get(/health) async def health(): return {status: ok}3.3 JSON-RPC 消息结构与 Pydantic 模型所有 MCP 请求都遵循 JSON-RPC 2.0。一个tools/call请求长这样{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: search_document, arguments: {query: JSON-RPC, max_results: 5} } }字段里jsonrpc必须严格等于2.0少一个字符就返回-32600错误id用于请求响应匹配通知类消息可以为空。用 Pydantic 定义模型时我踩过一个坑不同客户端传参格式不统一有的用params.name有的用params.tool有的把参数塞在params.args里。所以我在 handler 里做了兼容层统一归一化成namearguments再分发。3.4 方法分发路由表handler.handle_message()里用字典做 method 分发比一长串 if-else 清晰得多ROUTES { ping: handle_ping, initialize: handle_initialize, notifications/initialized: handle_initialized, tools/list: handle_tools_list, tools/call: handle_tools_call, resources/templates/list: handle_templates_list, resources/read: handle_resources_read, prompts/list: handle_prompts_list, prompts/get: handle_prompts_get, }resources/list我故意没实现返回-32601方法不存在。原因是文档数量可能很大全量枚举没意义改成用resources/templates/list告诉客户端 URI 模板再通过search_document工具定位具体文档最后用resources/read精读——这更符合「先搜索、再精读」的使用模式。3.5 会话状态机与保活会话有三个状态NOT_INITIALIZED→AWAITING_INIT→ACTIVE。所有方法在处理前检查状态未就绪就返回-32000。保活用双重策略SSE 每 30 秒发一行注释心跳防代理超时服务端空闲 60 秒主动发ping并等待 30 秒响应。这里有个细节asyncio.Queue.get()默认无限阻塞没法做心跳得配合asyncio.Eventasyncio.wait_for实现可超时等待。4. 可复制配置config.toml 与 settings.json 骨架配置分两份config.toml管 MCP 服务自身settings.json管 AI 客户端怎么连过来。这两份骨架你可以直接抄改掉 Key 和路径就能用。4.1 config.toml# MCP 服务端配置 [server] host 0.0.0.0 port 8080 api_key uni-index-dev-key-2026 # Bearer Token客户端需带上 session_timeout 300 # 僵尸连接清理阈值秒 ping_interval 60 # 服务端主动 ping 间隔 heartbeat_interval 30 # SSE 注释心跳间隔 [index] doc_root ./docs # 本地文档根目录 uri_scheme uni-index # 资源 URI 前缀 [model] provider openai-compatible base_url https://taotoken.net/api api_key sk-你的TaoToken密钥 model claude-sonnet-4-202505144.2 settings.jsonAI 客户端侧以 OpenCode / Claude Desktop 这类支持 MCP 的客户端为例配置里声明一个 MCP server指向你的 FastAPI 服务{ mcpServers: { uni-index: { url: http://127.0.0.1:8080/mcp, headers: { Authorization: Bearer uni-index-dev-key-2026 } } }, model: { baseUrl: https://taotoken.net/api, apiKey: sk-你的TaoToken密钥, model: claude-sonnet-4-20250514 } }注意url只配一个/mcpGET 和 POST 共用这个路径靠 HTTP Method 区分。这是我把官方推荐的/sse/messages双路径合并后的设计好处是客户端配置极简代价是路由逻辑依赖 Method 判断。4.3 启动命令# 开发模式带热重载 uvicorn server:app --host 0.0.0.0 --port 8080 --reload # 生产模式 uvicorn server:app --host 0.0.0.0 --port 8080 --workers 2启动后先访问http://127.0.0.1:8080/health返回{status:ok}说明服务起来了。这一步别跳过我见过好几次服务没起就急着配客户端结果排查半天发现是端口占用。5. 验证请求跑通一次完整 JSON-RPC 调用配置就绪后用 curl 手动走一遍握手和工具调用确认链路通了再接客户端。整个过程分四步SSE 连接拿 session_id、initialize 握手、tools/list 列工具、tools/call 调工具。5.1 建立 SSE 连接curl -N -H Authorization: Bearer uni-index-dev-key-2026 \ -H Accept: text/event-stream \ http://127.0.0.1:8080/mcp你会看到服务端先推一个endpoint事件里面带着session_idevent: endpoint data: /mcp?session_idabc-1235.2 initialize 握手拿到 session_id 后用 POST 发 initializecurl -X POST http://127.0.0.1:8080/mcp?session_idabc-123 \ -H Authorization: Bearer uni-index-dev-key-2026 \ -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}}}响应会通过 SSE 通道推回来包含协议版本和服务器能力列表。接着发notifications/initialized通知会话进入 ACTIVE 状态。5.3 tools/list 列工具curl -X POST http://127.0.0.1:8080/mcp?session_idabc-123 \ -H Authorization: Bearer uni-index-dev-key-2026 \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:2,method:tools/list}预期返回两个工具get_server_status和search_document。5.4 tools/call 实际调用curl -X POST http://127.0.0.1:8080/mcp?session_idabc-123 \ -H Authorization: Bearer uni-index-dev-key-2026 \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:3,method:tools/call,params:{name:search_document,arguments:{query:JSON-RPC,max_results:5}}}成功的话SSE 通道会推回搜索结果格式类似event: message data: {jsonrpc:2.0,id:3,result:{content:[{type:text,text:找到 3 条结果: [docs_concepts:42]协议基础 (匹配度: 1.20)...}]}}看到这个返回说明从 FastAPI 服务到 JSON-RPC 分发到 SSE 推送的整条链路都通了。接下来把settings.json配进 AI 客户端就能在对话里直接让模型调用你的工具。想先在网页端验证模型通道是否正常可以打开模型对话页面发一条测试消息https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmcp_fastapiutm_campaignrewrite6. 常见报错排查6.1 CORS preflight 被 401 拦截浏览器发 OPTIONS 预检请求时被 Auth 中间件拦下返回 401。原因是中间件注册顺序错了。FastAPI 里后注册的中间件先执行所以 CORSMiddleware 要放在最后注册让它最先处理 OPTIONSapp.add_middleware(MCPAuthMiddleware, api_keyapi_key) # 内层 app.add_middleware(CORSMiddleware, allow_origins[*]) # 外层6.2 ping 响应和客户端请求混淆服务端发 ping 后客户端返回{jsonrpc:2.0,id:srv-ping-xxx,result:{}}这个结构和一个无 method 的正常请求长得一样。我用两层校验区分第一层看字段特征无 method 有 result/error 有 id第二层查_pending_pings集合确认这个 ping_id 确实发过。只有两层都过才判定为 ping 响应否则忽略。6.3 SSE 连接空闲被代理切断asyncio.Queue.get()无限阻塞导致没法发心跳。改成asyncio.Eventasyncio.wait_for组合超时后先发心跳再继续等。同时每 30 秒发一行 SSE 注释: heartbeat很多反向代理看到有数据流动就不会断连。6.4 资源 URI 行号非法客户端可能传start end、start 0或非数字。在 URI 解析阶段就用正则加校验拦住返回规范的 JSON-RPC 错误别让非法参数流到业务层if start 1: raise UriParseError(f起始行号必须为正整数: {start}) if start end: raise UriParseError(f起始行号 {start} 不能大于结束行号 {end})6.5 僵尸连接堆积SSE 没有断开通知机制客户端异常退出后服务端不知道。加一个后台定时任务每 60 秒扫描所有会话last_heartbeat超过 300 秒的直接清理。这个阈值别设太小网络抖动时容易误杀正常连接。6.6 客户端参数格式不统一有的客户端传params.name有的传params.tool参数有的在arguments有的在args。在 handler 入口做一层归一化别在每个工具函数里各写一套兼容逻辑否则维护起来很痛苦。7. 接入 AI 工具与后续方向服务跑通后把settings.json放进 AI 客户端的配置目录重启客户端在对话里问一句「列出你可用的工具」如果模型能报出search_document说明 MCP 接入成功。这时候你的本地文档就真正变成 AI 能调用的上下文了。如果你打算把这个服务长期挂在后台给团队用建议走 Coding Plan 通道按周期计费比单次调用省心适合高频 Agent 场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentmcp_fastapiutm_campaignrewrite后续我准备把传输层从 SSE 迁到 Streamable HTTP。SSE 的长连接对代理不友好而 Streamable HTTP 用短连接按需响应在POST /mcp上加个?streamtrue参数就能兼容新旧客户端实现复杂度还更低。另外通知机制和 Sampling 也在规划里。手搓 MCP 最有意思的地方在于你不是在调 API而是在理解一种协议为什么这么设计——JSON-RPC 为什么用 id、SSE 双通道为什么比纯 WebSocket 更适合 AI 场景、资源为什么用 URI 模板而不是枚举列表。这些问题的答案只有自己写一遍才会真正明白。