)
1. 为什么我要自己写一个 MCP Server模型上下文协议Model Context Protocol简称 MCP这两年在 Agent 开发圈里出现得越来越频繁。简单说它是一套让 AI 客户端和外部工具之间用统一方式对话的规范客户端负责把用户问题转成工具调用服务端负责执行工具并把上下文回传。你可以把它理解成 Agent 世界的 USB-C 接口——不管对面是哪种模型、哪个客户端只要双方都按 MCP 说话工具就能即插即用。它适合谁如果你正在做大模型开发想让自己的 Agent 调用本地脚本、查数据库、读文件、跑命令又不想为每个客户端单独写一套适配层那 MCP Server 就是那个值得投入的中间层。我试过把几个内部工具直接硬编码进 Agent 的 prompt 里工具一多就乱成一锅粥后来改成 MCP Server 统一暴露客户端换了好几个都不用改工具代码。这篇会带你从零搭一个能跑的 MCP Server重点讲清三件事SSE 传输通道怎么建、工具怎么注册、Agent 怎么通过统一 Key 通道调用它。最后我们会用一次真实的工具调用验证协议握手和上下文回传是否成功。全程用可复制的配置和代码不堆概念。2. 前置准备用 TaoToken 做统一 Key 与 API 通道在动手写 Server 之前先把「模型侧」的通道打通。MCP Server 本身只负责工具执行真正发起工具调用决策的是模型。所以我们需要一个稳定的 API 入口让本地服务和模型对话时不用来回换 Key。TaoToken 在这里扮演的角色是统一 Key/API 通道你申请一个 Key就能在本地 MCP 服务、编码工具、Agent 脚本里复用同一套凭证省去每个客户端单独配置的麻烦。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 这个地址不加 UTM 参数配置里直接写它。具体操作分三步。第一步进控制台创建 API Key地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建后复制那串 sk- 开头的 Key只显示一次记得存好。第二步如果你要验证模型对话是否正常可以去模型对话页 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 发一条消息试试确认 Key 有额度、能返回。第三步如果你打算长期跑编码类 Agent建议看下 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合高频调用的场景。注意API Key 属于敏感凭证不要写进会提交到公开仓库的配置文件里。本地调试可以用环境变量或者放在 .gitignore 覆盖的 config 文件中。Key 拿到后我们把它作为环境变量注入后面 MCP Server 和客户端配置都会引用它export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api这两行是后面所有配置的基础。如果你用的是 Windows PowerShell把 export 换成$env:TAOTOKEN_API_KEYsk-你的Key即可。3. 可复制配置config.toml 与 settings.json 骨架MCP 生态里不同客户端的配置文件格式不太一样但核心字段就那几个服务名、启动命令、传输方式、环境变量。下面给两份骨架你可以直接抄。先看config.toml适合支持 TOML 配置的客户端或自建服务# config.toml - MCP Server 本地配置骨架 [mcp] server_name local-tools transport sse host 127.0.0.1 port 8765 sse_path /sse message_path /message [mcp.model] provider taotoken base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model claude-sonnet [[mcp.tools]] name get_time description 返回当前服务器时间 enabled true [[mcp.tools]] name echo_message description 原样回传输入文本用于验证上下文回传 enabled true再看settings.json适合 Claude Code、部分 IDE 插件这类用 JSON 描述 MCP 服务的客户端{ mcpServers: { local-tools: { transport: sse, url: http://127.0.0.1:8765/sse, env: { TAOTOKEN_API_KEY: ${TAOTOKEN_API_KEY}, TAOTOKEN_BASE_URL: https://taotoken.net/api } } } }这两份配置的对应关系可以用一张表说清配置项config.tomlsettings.json作用服务名server_namemcpServers 的键客户端识别服务的标识传输方式transporttransport固定为 sse连接地址host port sse_pathurl客户端建立 SSE 长连接消息端点message_path由服务端 endpoint 事件下发客户端 POST 请求地址凭证api_key_envenv注入统一 Key提示settings.json 里的${TAOTOKEN_API_KEY}是变量引用写法实际客户端会从环境变量读取。如果你的客户端不支持变量展开就手动填 Key但别提交到仓库。配置写好后客户端启动时会去连http://127.0.0.1:8765/sse服务端返回一个 endpoint 事件告诉它往哪 POST握手就开始了。下一节我们看服务端怎么实现这套逻辑。4. 从零实现SSE 传输、工具注册与握手流程MCP 的通信可以概括成「两通道、四步骤」。两通道是SSE 长连接负责服务端往客户端推消息HTTP POST 负责客户端往服务端发请求。四步骤是连建立 SSE、取拿到 POST 端点、握initialize 握手、用tools/list 和 tools/call。4.1 SSE 端点与 POST 端点先建 SSE 端点。客户端 GET 过来后服务端要立刻下发一个endpoint事件把后续 POST 地址告诉它# server.py - SSE 端点与 POST 端点 import json import time from flask import Flask, Response, request, jsonify app Flask(__name__) sessions {} app.route(/sse, methods[GET]) def sse_endpoint(): session_id request.args.get(clientId, fclient-{int(time.time())}) def event_stream(): # 第一步下发 endpoint 事件告知 POST 地址 endpoint_uri fhttp://127.0.0.1:8765/message/{session_id} yield fevent: endpoint\ndata: {endpoint_uri}\n\n # 保持连接定期心跳 while True: time.sleep(15) yield event: ping\ndata: {\type\:\ping\}\n\n return Response(event_stream(), mimetypetext/event-stream) app.route(/message/session_id, methods[POST]) def message_endpoint(session_id): req request.get_json(forceTrue) method req.get(method) req_id req.get(id) # 分发到协议处理逻辑 resp handle_request(method, req.get(params, {}), req_id) if resp is not None: push_to_client(session_id, resp) return jsonify({status: accepted})这里有个容易踩的坑SSE 响应头必须是text/event-stream而且不能有缓存。Flask 的Response配合mimetype参数基本够用但如果你前面挂了反向代理记得关掉缓冲否则事件会被攒着一起发握手就卡住了。4.2 工具注册与协议处理工具注册的本质就是维护一张「名字 → 执行函数」的表tools/list返回清单tools/call按名字找函数执行# tools.py - 工具注册表 TOOL_REGISTRY {} def register_tool(name, description, func): TOOL_REGISTRY[name] { name: name, description: description, handler: func, } def get_time(): return {content: [{type: text, text: time.strftime(%Y-%m-%d %H:%M:%S)}]} def echo_message(text): return {content: [{type: text, text: fecho: {text}}]} register_tool(get_time, 返回当前服务器时间, get_time) register_tool(echo_message, 原样回传输入文本, echo_message)协议处理函数负责把 JSON-RPC 请求映射到具体动作# protocol.py - 握手与工具调用 def handle_request(method, params, req_id): if method initialize: return { jsonrpc: 2.0, id: req_id, result: { protocolVersion: 2024-11-05, capabilities: {tools: {listChanged: True}}, serverInfo: {name: local-tools, version: 1.0.0}, }, } if method notifications/initialized: return None # 通知无需响应 if method tools/list: tools [{name: t[name], description: t[description]} for t in TOOL_REGISTRY.values()] return {jsonrpc: 2.0, id: req_id, result: {tools: tools}} if method tools/call: name params.get(name) args params.get(arguments, {}) tool TOOL_REGISTRY.get(name) if not tool: return {jsonrpc: 2.0, id: req_id, error: {code: -32601, message: ftool not found: {name}}} result tool[handler](**args) return {jsonrpc: 2.0, id: req_id, result: result} return {jsonrpc: 2.0, id: req_id, error: {code: -32601, message: fmethod not found: {method}}}push_to_client就是把响应通过之前保存的 SSE 连接推回去。注意initialize必须返回协议版本和能力声明客户端拿到后才会发notifications/initialized确认握手完成。少了这一步后面的tools/list会被客户端拒绝。5. 验证请求一次工具调用跑通握手与上下文回传服务端跑起来后用 curl 模拟客户端走一遍完整流程。先启动服务python server.py # 输出Running on http://127.0.0.1:8765然后开一个终端建立 SSE 连接观察服务端下发的事件curl -N http://127.0.0.1:8765/sse?clientIdtest-01 # 预期输出 # event: endpoint # data: http://127.0.0.1:8765/message/test-01拿到 endpoint 后另开终端发 initialize 请求curl -X POST http://127.0.0.1:8765/message/test-01 \ -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 终端应该能看到服务端推回的 initialize 响应里面包含serverInfo和capabilities。接着发 initialized 通知和 tools/listcurl -X POST http://127.0.0.1:8765/message/test-01 \ -H Content-Type: application/json \ -d {jsonrpc:2.0,method:notifications/initialized} curl -X POST http://127.0.0.1:8765/message/test-01 \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:2,method:tools/list}SSE 终端会推回工具清单包含get_time和echo_message。最后调用一次工具验证上下文回传curl -X POST http://127.0.0.1:8765/message/test-01 \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:3,method:tools/call,params:{name:echo_message,arguments:{text:hello mcp}}}预期在 SSE 流里看到{jsonrpc:2.0,id:3,result:{content:[{type:text,text:echo: hello mcp}]}}看到这行就说明整条链路通了SSE 建连成功、endpoint 下发正确、握手完成、工具注册可查、调用结果通过 SSE 回传。如果你在客户端里接的是 TaoToken 的模型通道模型会拿到这个 result 继续生成回答上下文就串起来了。6. 本篇常见错排查握手卡住tools/list 报未初始化。最常见的原因是客户端没发notifications/initialized或者服务端把它当成普通请求返回了响应。通知类消息的 id 字段缺失服务端必须返回 None不能推任何东西回去否则客户端会认为协议错乱。SSE 连接建立后收不到 endpoint 事件。检查响应头是不是text/event-stream以及中间有没有反向代理开了缓冲。Nginx 默认会缓冲 SSE需要在 location 里加proxy_buffering off;。另外事件格式必须是event: xxx\ndata: xxx\n\n结尾两个换行不能少。POST 请求返回 404。多半是 endpoint 事件里下发的地址和实际路由对不上。注意 session_id 要一致SSE 连接用的 clientId 和 POST 路径里的 session_id 必须是同一个否则服务端找不到对应的连接消息推不回去。工具调用报 method not found。检查tools/call的 params 结构name 和 arguments 是平级的别嵌套错。另外工具注册时函数签名要和 arguments 的键对应echo_message(text...)对应{text: ...}键名不一致会抛 TypeError。Key 相关报错。如果模型侧调用返回鉴权失败先确认环境变量TAOTOKEN_API_KEY在当前 shell 里生效可以用echo $TAOTOKEN_API_KEY检查。如果要在客户端里管理 Key去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 重新生成一个接入细节参考文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。Claude Code 这类工具接入可以看 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite 。7. 把 MCP Server 接进你的日常开发流服务跑通只是第一步真正省时间的是把它接进日常工具链。如果你主要用编码类 Agent把上面那份 settings.json 放进客户端的 MCP 配置目录重启后它就会自动连你的本地 Server。之后你在对话里说「现在几点」Agent 会自己决定调get_time你不用手动敲 curl。想让工具更实用可以按同样的注册模式加几个读本地日志文件的read_log、查 SQLite 的query_db、调内部 HTTP 接口的call_api。每个工具就是一个 Python 函数加一行 register扩展成本很低。唯一要注意的是别把生产库直连暴露成工具MCP 的定位是给模型提供上下文不是绕过权限的通道。统一 Key 这块TaoToken 的价值在于你换客户端、换模型时不用重新配一遍凭证。本地 MCP Server 读环境变量编码工具读同一份配置Agent 脚本也引用同一个 Key维护面就收窄到一处。长期跑下来这种「一处配置、多处复用」的省心感比省下的那点配置时间更值钱。