1. 为什么我要自己写一个 MCP Server
MCP Server 这个词最近在 AI 圈子里出现得越来越频繁。简单说，它是一个遵循 Model Context Protocol 的服务端程序，能让 Claude、Cursor、Cline 这类支持 MCP 的客户端调用你自定义的工具、读取你指定的资源、复用你写好的提示词模板。你可以把它理解成给 AI 装了一个「外挂工具箱」——AI 本身不会查你公司的数据库、不会读你本地的日志文件，但只要你把这些能力包装成一个 MCP Server，客户端就能在对话里直接调用。
它适合谁？三类人最值得动手：一是想把内部 API 暴露给 AI 助手的前后端工程师；二是想让 AI 帮自己操作本地脚本、文件、数据库的运维和数据分析同学；三是正在做 AI Agent 产品、需要标准化工具接入层的开发者。这三类人有一个共同点：不想每次换客户端就重写一遍工具逻辑，而 MCP 正好把「工具定义」和「客户端」解耦了。
我这次要带你做的，是一个用 Python SDK 从零实现的 MCP Server，覆盖两种传输方式：STDIO（标准输入输出，本地进程通信）和 SSE（Server-Sent Events，HTTP 长连接推送）。同时，我会把服务端调用大模型时的 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 注册并登录。登录后进入控制台，地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，在左侧找到 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_KEY=sk-你的密钥 TAOTOKEN_BASE_URL=https://taotoken.net/api/v1 TAOTOKEN_MODEL=你的模型ID注意：不要把 Key 硬编码进
mcp_server.py。MCP Server 经常会被客户端以子进程方式拉起，硬编码的 Key 容易在日志里被打印出来，一旦日志外泄就等于密钥泄露。
如果你更习惯用命令行临时注入，也可以这样启动：
export TAOTOKEN_API_KEY="sk-你的密钥" export TAOTOKEN_BASE_URL="https://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": f"Bearer {API_KEY}", "Content-Type": "application/json", } payload = { "model": MODEL_ID, "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, } with httpx.Client(timeout=60) as client: resp = client.post(url, headers=headers, json=payload) 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( name="summarize_text", description="对输入文本做摘要，走 TaoToken 统一通道", inputSchema={ "type": "object", "properties": { "text": {"type": "string", "description": "待摘要的文本"} }, "required": ["text"], }, ), types.Tool( name="echo", 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(type="text", text=f"echo: {arguments['message']}")] if name == "summarize_text": result = call_model(f"请用三句话总结以下内容：\n{arguments['text']}") return [types.TextContent(type="text", text=result)] 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", endpoint=handle_sse), Mount("/messages/", app=sse.handle_post_message), ] ) uvicorn.run(starlette_app, host="127.0.0.1", port=8000) 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( command="python", 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 通道已经建立。按 Ctrl+C 断开即可。
然后用客户端脚本走完整流程：
# 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_ascii=False)}")这样报错信息会直接告诉你服务端到底返回了什么，比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_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ，这样额度管理和 Key 轮换都在一个地方，不用每个 server 单独维护。日常调试模型返回是否正常，可以直接用模型对话页面 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 手动发一条同样的 prompt，对比 server 里的返回，快速判断是模型问题还是代码问题。
新增工具时的经验：先把工具逻辑写成一个纯函数，单独测试通过，再包进call_tool。这样出问题时你能立刻定位是工具逻辑错了还是 MCP 协议层错了。另外，inputSchema一定要写清楚required字段，客户端靠它做参数校验，写漏了会导致调用时参数缺失却报不出明确错误。
最后一个小技巧：在call_tool里加一行日志，把工具名和参数打印到 stderr。STDIO 模式下 stdout 被协议占用，日志必须走 stderr，否则会污染 JSON-RPC 消息流，导致客户端解析失败。这个坑我踩过一次，现象是客户端莫名其妙断开，查了半天才发现是print用错了流。
