1. 为什么我要自己写一个 MCP ServerMCP Server 这个词最近在 AI 圈子里出现得越来越频繁。简单说它是一个遵循 Model Context Protocol 的服务端程序能让 Claude、Cursor、Cline 这类支持 MCP 的客户端调用你自定义的工具、读取你指定的资源、复用你写好的提示词模板。你可以把它理解成给 AI 装了一个「外挂工具箱」——AI 本身不会查你公司的数据库、不会读你本地的日志文件但只要你把这些能力包装成一个 MCP Server客户端就能在对话里直接调用。它适合谁三类人最值得动手一是想把内部 API 暴露给 AI 助手的前后端工程师二是想让 AI 帮自己操作本地脚本、文件、数据库的运维和数据分析同学三是正在做 AI Agent 产品、需要标准化工具接入层的开发者。这三类人有一个共同点不想每次换客户端就重写一遍工具逻辑而 MCP 正好把「工具定义」和「客户端」解耦了。我这次要带你做的是一个用 Python SDK 从零实现的 MCP Server覆盖两种传输方式STDIO标准输入输出本地进程通信和 SSEServer-Sent EventsHTTP 长连接推送。同时我会把服务端调用大模型时的 endpoint 和鉴权统一改到 TaoToken 的 Key/API 通道上这样你本地调试和线上部署用的是同一套凭证不用来回改配置。整篇的节奏是先讲清楚 MCP Server 到底在解决什么问题再准备 TaoToken 的 Key 和 Base URL然后给你一份可以直接复制的 server 骨架代码接着分别用 STDIO 和 SSE 启动并验证连通性最后把几个我踩过的报错摊开讲。你跟着敲一遍大概四十分钟能跑通第一个能被客户端调用的 MCP Server。需要提前说明的是MCP 协议本身还在演进Python SDK 的接口在不同版本间会有细微差异。我下面用的写法以官方mcp包的主流版本为准如果你装的是更新的版本个别导入路径可能要微调我会在排障章节里点出来。2. TaoToken 统一 Key 与 API 通道准备在写代码之前先把「AI 能力从哪来」这件事定下来。MCP Server 本身只是工具层但很多工具比如让 AI 总结一段文本、生成 SQL需要调用大模型。如果每个工具都单独配一套 Key代码里会散落一堆凭证换环境时非常痛苦。我的做法是统一走 TaoToken 的 API 通道一个 Key 覆盖多个模型。第一步打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册并登录。登录后进入控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 在左侧找到 API Keys 菜单点进去创建一个新的 Key。创建时给它起个能认出来的名字比如mcp-local-dev方便以后区分是本地调试还是线上服务。创建完成后你会拿到一串以sk-开头的密钥。这串东西只显示一次复制下来存到安全的地方。我一般会把它写进项目根目录的.env文件并且把.env加进.gitignore避免误提交。第二步确认 Base URL。TaoToken 的 API 入口是 https://taotoken.net/api 注意这里不带任何查询参数就是干净的 API 根路径。在 OpenAI 兼容的 SDK 里你通常需要填的是https://taotoken.net/api/v1这种带版本号的地址具体以你所用 SDK 的文档为准。我下面代码里会把它抽成环境变量方便切换。第三步选一个 Model ID。如果你只是做本地验证选一个响应快、成本低的对话模型即可。Model ID 的准确写法在控制台的模型列表里能看到直接复制不要凭记忆手敲大小写和连字符错一个字符就会报模型不存在。把这三样东西整理成环境变量后面代码直接读# .env 文件内容 TAOTOKEN_API_KEYsk-你的密钥 TAOTOKEN_BASE_URLhttps://taotoken.net/api/v1 TAOTOKEN_MODEL你的模型ID注意不要把 Key 硬编码进mcp_server.py。MCP Server 经常会被客户端以子进程方式拉起硬编码的 Key 容易在日志里被打印出来一旦日志外泄就等于密钥泄露。如果你更习惯用命令行临时注入也可以这样启动export TAOTOKEN_API_KEYsk-你的密钥 export TAOTOKEN_BASE_URLhttps://taotoken.net/api/v1 export TAOTOKEN_MODEL你的模型ID到这里前置准备就完成了。你手里应该有一个可用的 Key、一个 Base URL、一个 Model ID。接下来进入代码环节。3. 可复制的 MCP Server 骨架代码这一节是全文的核心我会给你一份完整的mcp_server.py它同时支持 STDIO 和 SSE 两种传输方式并且把模型调用统一指向 TaoToken 通道。代码我拆成几块讲你可以直接整段复制。先装依赖。MCP 的 Python SDK 包名是mcp另外我们需要httpx做 HTTP 请求、python-dotenv读环境变量、uvicorn和starlette支撑 SSE 的 HTTP 服务pip install mcp[cli] httpx python-dotenv uvicorn starlette如果你用的是较老的 Python建议升到 3.10 以上因为 SDK 里用到了不少新语法。装完后可以用pip show mcp确认版本。接下来是代码。先写配置加载和模型调用部分# mcp_server.py import os import json import httpx from dotenv import load_dotenv load_dotenv() API_KEY os.getenv(TAOTOKEN_API_KEY) BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api/v1) MODEL_ID os.getenv(TAOTOKEN_MODEL) def call_model(prompt: str) - str: 统一走 TaoToken 通道调用模型 if not API_KEY: raise RuntimeError(缺少 TAOTOKEN_API_KEY请检查 .env) url f{BASE_URL}/chat/completions headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } payload { model: MODEL_ID, messages: [{role: user, content: prompt}], temperature: 0.3, } with httpx.Client(timeout60) as client: resp client.post(url, headersheaders, jsonpayload) resp.raise_for_status() data resp.json() return data[choices][0][message][content]这段的关键点在于BASE_URL和API_KEY都从环境变量读call_model里拼的是标准的/chat/completions路径。如果你的 SDK 版本对路径有要求改BASE_URL即可不用动函数体。然后是 MCP Server 的主体。用官方 SDK 的Server类注册工具from mcp.server import Server import mcp.types as types app Server(taotoken-demo-server) app.list_tools() async def list_tools() - list[types.Tool]: return [ types.Tool( namesummarize_text, description对输入文本做摘要走 TaoToken 统一通道, inputSchema{ type: object, properties: { text: {type: string, description: 待摘要的文本} }, required: [text], }, ), types.Tool( nameecho, description回显输入用于连通性验证, inputSchema{ type: object, properties: { message: {type: string} }, required: [message], }, ), ] app.call_tool() async def call_tool(name: str, arguments: dict) - list[types.TextContent]: if name echo: return [types.TextContent(typetext, textfecho: {arguments[message]})] if name summarize_text: result call_model(f请用三句话总结以下内容\n{arguments[text]}) return [types.TextContent(typetext, textresult)] raise ValueError(f未知工具: {name})list_tools负责告诉客户端「我有哪些工具」call_tool负责真正执行。echo工具不调用模型专门用来验证链路是否通summarize_text才会走 TaoToken 通道。最后是两种传输方式的启动入口import anyio from mcp.server.stdio import stdio_server async def run_stdio(): async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream, app.create_initialization_options()) def run_sse(): from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Route, Mount import uvicorn sse SseServerTransport(/messages/) async def handle_sse(request): async with sse.connect_sse( request.scope, request.receive, request._send ) as streams: await app.run( streams[0], streams[1], app.create_initialization_options() ) starlette_app Starlette( routes[ Route(/sse, endpointhandle_sse), Mount(/messages/, appsse.handle_post_message), ] ) uvicorn.run(starlette_app, host127.0.0.1, port8000) if __name__ __main__: import sys mode sys.argv[1] if len(sys.argv) 1 else stdio if mode stdio: anyio.run(run_stdio) elif mode sse: run_sse() else: print(用法: python mcp_server.py [stdio|sse])这份骨架的好处是STDIO 和 SSE 共用同一套工具注册逻辑你新增工具只需要改list_tools和call_tool两种传输方式自动都能用。模型调用也只有一个出口call_model换 Key 或换 Base URL 只改环境变量。提示SSE 模式下SseServerTransport的路径/messages/要和客户端配置里的 messages 地址保持一致否则客户端发起的工具调用会 404。4. STDIO 与 SSE 启动及连通性验证代码写完了接下来分别把两种模式跑起来确认真的能被调用。先验证 STDIO。STDIO 模式下MCP Server 是被客户端以子进程方式拉起的它通过标准输入读 JSON-RPC 消息、通过标准输出写回结果。所以你不能直接python mcp_server.py stdio然后干等那样它会在 stdin 上阻塞。正确的验证方式是写一个最小的客户端脚本或者用官方提供的调试工具。我推荐先用官方 CLI 调试器装 SDK 时带的mcp命令就能用mcp dev mcp_server.py这条命令会启动一个开发用界面自动以 STDIO 方式拉起你的 server并列出所有工具。你在界面里点echo输入hello如果返回echo: hello说明 STDIO 链路通了。再点summarize_text输入一段文字如果返回摘要说明 TaoToken 通道也通了。如果你不想用界面也可以手写一个最小客户端# test_client.py import anyio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters( commandpython, args[mcp_server.py, stdio], ) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具:, [t.name for t in tools.tools]) result await session.call_tool(echo, {message: ping}) print(echo 返回:, result.content[0].text) anyio.run(main)运行python test_client.py你应该看到工具列表和echo: ping。这一步成功说明 STDIO 模式完全可用。再验证 SSE。SSE 模式下 server 是一个常驻的 HTTP 服务客户端通过GET /sse建立事件流通过POST /messages/发送请求。启动命令python mcp_server.py sse看到 uvicorn 打印Uvicorn running on http://127.0.0.1:8000就说明起来了。先做最基础的连通性检查另开一个终端curl -N http://127.0.0.1:8000/sse-N是关闭缓冲你会看到服务端持续推送event: endpoint和data:行里面包含一个 session 相关的 endpoint 路径。这说明 SSE 通道已经建立。按 CtrlC 断开即可。然后用客户端脚本走完整流程# test_sse_client.py import anyio from mcp import ClientSession from mcp.client.sse import sse_client async def main(): async with sse_client(http://127.0.0.1:8000/sse) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(echo, {message: sse-ping}) print(SSE echo 返回:, result.content[0].text) anyio.run(main)运行后如果打印SSE echo 返回: sse-ping两种传输方式就都验证通过了。这时候你可以把summarize_text也调一次确认模型通道在 SSE 模式下同样工作。注意SSE 模式默认监听127.0.0.1只允许本机访问。如果你要部署到服务器给远程客户端用需要改成0.0.0.0并且务必加上鉴权否则任何人都能调用你的模型额度。5. 常见报错排查401、local proxy failed、reading choices这一节把我实际遇到过的几个报错摊开讲你大概率会撞上其中一两个。报错一401 Unauthorized。这个最常见出现在call_model里。原因通常是三种Key 没读到、Key 写错、Base URL 拼错。先确认.env是否被load_dotenv()正确加载可以在代码里临时打印API_KEY[:8]看前几位对不对。如果 Key 是对的检查BASE_URL是否以/v1结尾以及call_model里拼出来的完整 URL 是不是https://taotoken.net/api/v1/chat/completions。我踩过的坑是把 Base URL 写成了https://taotoken.net/api少了/v1结果请求打到了不存在的路径返回 404 而不是 401排查时容易误判。报错二local proxy failed。这个报错通常出现在客户端侧提示无法连接到本地 MCP Server。STDIO 模式下多半是command或args配错了比如客户端配置里写的是python但你的环境里只有python3。SSE 模式下多半是端口没起来或者被占用。先用curl确认端口能通再检查客户端配置里的 URL 是否和 server 实际监听地址一致。另外有些客户端对127.0.0.1和localhost的处理不同如果连不上两个都试一下。报错三reading choices 相关错误。这个报错来自模型返回体解析典型信息是KeyError: choices或者list index out of range。原因一般是模型返回了错误结构比如返回了{error: {...}}而不是标准的{choices: [...]}。这时候不要只看异常要把resp.text打印出来看原始返回。常见触发场景是 Model ID 写错服务端返回了模型不存在的错误或者请求体里messages格式不对。我建议在call_model里加一层判断data resp.json() if choices not in data: raise RuntimeError(f模型返回异常: {json.dumps(data, ensure_asciiFalse)})这样报错信息会直接告诉你服务端到底返回了什么比KeyError好排查得多。报错四OAuth 或鉴权头冲突。有些客户端在连接 MCP Server 时会自动带上自己的 OAuth 头如果你的 server 又要求另一种鉴权两边会打架。SSE 模式下尤其明显。解决办法是在 server 侧明确只认一种鉴权方式或者在客户端配置里关掉自动鉴权。如果你用的是 Cline、CC Switch 这类工具检查它们的 MCP 配置里有没有多余的headers字段。关于三件套的完整性。无论你用哪种客户端接入一个 MCP Server 本质上都要配齐三样东西Base URL或启动命令、Key或环境变量、Model ID。以 Cline 的 MCP 配置为例如果是 SSE 模式配置里要有url指向http://127.0.0.1:8000/sse如果是 STDIO 模式要有command和args。Key 和 Model ID 则通过环境变量传给 server 进程。这三样缺一个链路就断。我见过有人只配了 URL 没配环境变量结果 server 起来了但一调用模型就 401排查半天才发现是 Key 没传进去。6. 把 MCP Server 接到你的日常工作流跑通第一个 server 之后真正有价值的是把它接到你每天用的工具里。如果你用的是 Claude Code 这类编码助手可以在它的配置里注册这个 MCP Server让它在写代码时直接调用你的summarize_text或后续新增的工具。配置方式是在客户端的 MCP 设置里新增一条STDIO 模式填启动命令SSE 模式填 URL。如果你打算长期跑、并且工具会越来越多建议把模型调用统一收敛到 TaoToken 的 Coding Plan 通道上地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 这样额度管理和 Key 轮换都在一个地方不用每个 server 单独维护。日常调试模型返回是否正常可以直接用模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 手动发一条同样的 prompt对比 server 里的返回快速判断是模型问题还是代码问题。新增工具时的经验先把工具逻辑写成一个纯函数单独测试通过再包进call_tool。这样出问题时你能立刻定位是工具逻辑错了还是 MCP 协议层错了。另外inputSchema一定要写清楚required字段客户端靠它做参数校验写漏了会导致调用时参数缺失却报不出明确错误。最后一个小技巧在call_tool里加一行日志把工具名和参数打印到 stderr。STDIO 模式下 stdout 被协议占用日志必须走 stderr否则会污染 JSON-RPC 消息流导致客户端解析失败。这个坑我踩过一次现象是客户端莫名其妙断开查了半天才发现是print用错了流。