1. 为什么我要自己写 MCP Server:从工具碎片化到统一插座
如果你正在做 AI Agent 或者智能助手类产品,大概率遇到过这样的场景:模型要读本地文件,你写一套函数调用;模型要查数据库,你又写一套;换个模型供应商,之前那套工具描述和参数格式全部推倒重来。MCP Server 就是来解决这个问题的——它把「AI 能调用的能力」抽象成一个标准化的服务端,用 JSON-RPC 2.0 作为通信协议,通过 stdio 或 SSE 传输,任何支持 MCP 的客户端都能即插即用。
MCP Server 本质上是一个遵循 Model Context Protocol 的进程,它向 AI 客户端暴露三类能力:Tools(可调用的函数)、Resources(可读取的数据)、Prompts(预定义模板)。其中 Tools 最常用,也是本文的重点。适合谁?适合所有需要给 AI 挂载自定义能力的 Python 开发者,尤其是那些不想为每个模型平台重复写集成代码的人。
我试过用 Function Calling 给三个不同平台分别写工具描述,维护成本高得离谱。后来把逻辑收敛到一个 MCP Server 里,客户端换了一圈,Server 代码一行没改。这篇文章就带你从零写一个能跑、能验证、能接入 TaoToken 统一通道的 Python MCP Server,包含完整骨架、配置文件片段和本地 stdio 联调步骤。
核心检索词先明确:MCP Server 是什么?它是一个用 JSON-RPC 与 stdio 通信、可被 AI 工具调用的自定义工具箱服务端。能做什么?把文件读写、数据库查询、HTTP 请求等能力标准化暴露给 AI。适合谁?做 Agent 开发、需要统一工具接口的 Python 工程师。
2. 前置准备:Python 环境、MCP SDK 与 TaoToken 统一 Key 通道
在写代码之前,先把环境和账号通道准备好。这一章不涉及复杂配置,但每一步都会影响后面能不能跑通。
2.1 Python 版本与 MCP SDK 安装
MCP 的 Python SDK 要求 Python 3.10 及以上。先确认版本:
python3 --version # 期望输出类似:Python 3.11.9然后安装 SDK:
pip install mcp # 期望输出:Successfully installed mcp-1.27.0验证安装:
python3 -c "import mcp; print(mcp.__version__)" # 期望输出:1.27.0版本差异要留意。MCP SDK 在 1.x 系列里 API 有过调整,如果你装到的是更老的版本,建议升级到较新的稳定版:
pip install --upgrade mcp2.2 为什么需要 TaoToken 统一 Key 通道
自己写的 MCP Server 负责「工具能力」,但工具背后往往还要调用大模型来做推理、总结或决策。如果每个模型供应商都单独配一套 Key、一套 Base URL,管理起来很乱。TaoToken 提供统一 Key 和 API 通道,把模型调用收敛到一个入口,MCP Server 里只需要配置一次就能切换不同模型。
TaoToken 官网入口:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
API 基础地址(不加 UTM):https://taotoken.net/api
你需要先去控制台创建一个 API Key,后面在 MCP Server 的配置里会用到。创建 Key 的入口在控制台的 API Keys 页面,模型对话调试可以在模型对话页面完成,长期编码或 Agent 场景可以看 Coding Plan。
2.3 目录与测试数据准备
为了让后面的文件系统 Server 有东西可读,先建一个受控目录和测试文件:
mkdir -p ~/allowed_files/notes echo "# 2026-07-23 周会记录" > ~/allowed_files/notes/meeting.md echo "MCP Server 联调测试" > ~/allowed_files/README.md这个~/allowed_files就是后面 Server 的安全边界根目录,所有文件操作都被限制在里面。
2.4 三件套概念:Base URL + Key + Model ID
不管你用哪种客户端接入 TaoToken,核心都是三件套:
| 配置项 | 值 | 说明 |
|---|---|---|
| Base URL | https://taotoken.net/api | 统一 API 入口 |
| API Key | 控制台创建 | 身份凭证 |
| Model ID | 按需选择 | 模型标识 |
后面在 MCP Server 里调用模型时,这三个值会写进配置或环境变量。记住这个组合,任何客户端接入都是围绕它展开的。
3. 可复制配置:MCP Server 骨架、config.toml 与 settings.json 片段
这一章是全文的技术核心,给出可直接复制的 Server 代码和客户端配置片段。所有路径和字段都保持可运行状态。
3.1 文件系统 MCP Server 完整骨架
新建fs_mcp_server.py:
#!/usr/bin/env python3 """fs_mcp_server.py — 文件系统 MCP Server 允许 AI 在受控目录内安全读取和搜索文件。 """ import os import json import fnmatch from mcp.server.fastmcp import FastMCP # 安全边界:所有操作限制在此目录内 ALLOWED_ROOT = os.path.expanduser("~/allowed_files") mcp = FastMCP("file-system-server") def _is_path_safe(requested_path: str) -> bool: """确保请求路径没有逃出允许的根目录""" abs_path = os.path.abspath(os.path.join(ALLOWED_ROOT, requested_path)) return abs_path.startswith(ALLOWED_ROOT) @mcp.tool(description="读取指定文件的内容。当你需要查看本地文档、笔记或配置时调用。参数 path 是相对于受控根目录的路径,例如 'notes/meeting.md'。返回文件文本内容。") async def read_file(path: str) -> str: if not _is_path_safe(path): return f"错误:路径 '{path}' 超出允许范围" abs_path = os.path.abspath(os.path.join(ALLOWED_ROOT, path)) if not os.path.isfile(abs_path): return f"错误:文件不存在 '{path}'" try: with open(abs_path, "r", encoding="utf-8") as f: return f.read() except Exception as e: return f"读取失败:{e}" @mcp.tool(description="搜索受控目录下的文件,支持通配符。当你需要按名称模式查找文件时调用。参数 pattern 如 '*.md',参数 root_dir 是可选子目录。返回匹配文件的相对路径 JSON 数组。") async def search_files(pattern: str, root_dir: str = "") -> str: search_root = os.path.join(ALLOWED_ROOT, root_dir) if root_dir else ALLOWED_ROOT if not search_root.startswith(ALLOWED_ROOT): return f"错误:目录 '{root_dir}' 超出允许范围" results = [] for dirpath, _, filenames in os.walk(search_root): for fn in filenames: if fnmatch.fnmatch(fn, pattern): rel_path = os.path.relpath(os.path.join(dirpath, fn), ALLOWED_ROOT) results.append(rel_path) return json.dumps(results, indent=2, ensure_ascii=False) if __name__ == "__main__": os.makedirs(ALLOWED_ROOT, exist_ok=True) mcp.run(transport="stdio")这段代码的关键点有三个:_is_path_safe做路径逃逸防护,@mcp.tool的 description 写清楚调用场景,mcp.run(transport="stdio")走标准输入输出通信。
3.2 带模型调用的 MCP Server 配置片段
如果你的 MCP Server 内部还要调用大模型,把 TaoToken 三件套写进配置。以config.toml为例:
[mcp] name = "file-system-server" transport = "stdio" [llm] base_url = "https://taotoken.net/api" api_key = "sk-你的TaoToken密钥" model_id = "claude-3-5-sonnet"对应的settings.json片段(适用于支持 JSON 配置的客户端):
{ "mcpServers": { "file-system-server": { "command": "python3", "args": ["/home/user/fs_mcp_server.py"], "env": { "TAOTOKEN_BASE_URL": "https://taotoken.net/api", "TAOTOKEN_API_KEY": "sk-你的TaoToken密钥", "TAOTOKEN_MODEL_ID": "claude-3-5-sonnet" } } } }注意command和args的路径要换成你机器上的真实路径。env里的三个变量就是 Base URL、Key、Model ID 三件套,缺一不可。
3.3 SQLite 查询 Server 骨架
再给一个更贴近生产的例子,sqlite_mcp_server.py:
#!/usr/bin/env python3 """sqlite_mcp_server.py — 只读 SQLite 查询 MCP Server""" import sqlite3 import json from pathlib import Path from mcp.server.fastmcp import FastMCP mcp = FastMCP("sqlite-query-server") READ_ONLY_DB = str(Path.home() / "data/app.db") def _validate_query(sql: str) -> bool: stripped = sql.strip().upper() forbidden = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER"] return stripped.startswith("SELECT") and not any(w in stripped for w in forbidden) @mcp.tool(description="执行 SQLite SELECT 查询并返回 JSON 结果。当你需要查询用户数据、订单记录或产品信息时调用。仅支持只读查询,禁止写操作。") async def query_database(sql: str) -> str: if not _validate_query(sql): return "错误:仅支持 SELECT 查询" try: conn = sqlite3.connect(READ_ONLY_DB) conn.row_factory = sqlite3.Row cursor = conn.execute(sql) rows = [dict(row) for row in cursor.fetchall()] conn.close() return json.dumps(rows, indent=2, ensure_ascii=False, default=str) except Exception as e: return f"查询失败:{e}" @mcp.tool(description="列出数据库中所有表和视图。当你需要了解数据库结构时调用。返回表名和类型的 JSON 数组。") async def list_tables() -> str: try: conn = sqlite3.connect(READ_ONLY_DB) cursor = conn.execute( "SELECT name, type FROM sqlite_master WHERE type IN ('table', 'view')" ) tables = [dict(row) for row in cursor.fetchall()] conn.close() return json.dumps(tables, indent=2, ensure_ascii=False) except Exception as e: return f"查询失败:{e}" if __name__ == "__main__": mcp.run(transport="stdio")_validate_query是安全核心,只放行 SELECT,把写操作全部拦掉。AI 生成的 SQL 不可信,这层校验必须有。
4. 验证请求:stdio 联调与成功结果确认
代码写完不算完,必须验证 Server 真的能被调用。这一章给出完整的联调步骤和预期输出。
4.1 用 MCP CLI 直接测试
MCP SDK 自带命令行工具,可以直接跑 Server:
python3 -m mcp run fs_mcp_server.py如果 Server 正常启动,会进入等待 JSON-RPC 消息的状态。你可以手动发一条初始化请求测试:
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}' | python3 fs_mcp_server.py期望返回类似:
{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2024-11-05","capabilities":{"tools":{}},"serverInfo":{"name":"file-system-server","version":"1.0"}}}看到serverInfo里有你的 Server 名字,说明 stdio 通道通了。
4.2 调用工具验证
继续发一条工具调用请求,测试read_file:
echo '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"read_file","arguments":{"path":"notes/meeting.md"}}}' | python3 fs_mcp_server.py期望返回文件内容:
{"jsonrpc":"2.0","id":2,"result":{"content":[{"type":"text","text":"# 2026-07-23 周会记录"}]}}再测路径逃逸防护:
echo '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"read_file","arguments":{"path":"../../etc/passwd"}}}' | python3 fs_mcp_server.py期望返回:
{"jsonrpc":"2.0","id":3,"result":{"content":[{"type":"text","text":"错误:路径 '../../etc/passwd' 超出允许范围"}]}}安全边界生效,说明_is_path_safe工作正常。
4.3 挂载到客户端验证
把settings.json片段写进客户端的 MCP 配置里,重启客户端。在对话里让 AI 读取notes/meeting.md,如果 AI 能返回文件内容,说明整条链路打通:客户端 → stdio → MCP Server → 文件系统。
实测下来,stdio 模式最大的好处就是可以直接用命令行调试,不用先部署 HTTP 服务。开发阶段效率高很多。
5. 常见报错排查:401、local proxy failed、reading choices 与 OAuth
联调过程中最容易卡在几个典型报错上,这一章逐个对照排查。
5.1 401 Unauthorized
现象:调用模型接口时返回 401。
原因通常是 API Key 没配、配错或过期。检查config.toml或settings.json里的api_key字段,确认和 TaoToken 控制台里创建的一致。注意 Key 不要有多余空格或换行。
[llm] base_url = "https://taotoken.net/api" api_key = "sk-你的TaoToken密钥" # 确认这里没有引号包裹错误 model_id = "claude-3-5-sonnet"如果 Key 确认无误还是 401,去控制台重新生成一个再试。
5.2 local proxy failed
现象:客户端报local proxy failed或连接被拒绝。
这个报错通常和网络配置有关。检查 Base URL 是否写成了https://taotoken.net/api,不要多加斜杠或路径。同时确认本机没有异常的本地代理设置干扰请求。如果客户端有代理配置项,清空或设为直连。
5.3 reading choices 报错
现象:返回体解析时报reading 'choices'或类似字段缺失。
这通常说明返回的不是标准模型响应格式,可能是 Base URL 配错导致请求打到了非预期端点。确认base_url是https://taotoken.net/api,且model_id是有效模型标识。如果用的是 OpenAI 兼容格式,检查客户端是否开启了对应的兼容模式。
5.4 OAuth 相关报错
现象:客户端提示 OAuth 认证失败或 token 无效。
部分客户端默认走 OAuth 流程,但 TaoToken 用的是 API Key 认证。在客户端设置里把认证方式切换为 API Key,填入三件套。如果客户端强制 OAuth,检查是否有「使用 API Key」的选项。
5.5 三件套检查清单
出现任何接入问题,先对照这张表:
| 检查项 | 正确值 | 常见错误 |
|---|---|---|
| Base URL | https://taotoken.net/api | 多写路径、少写 /api |
| API Key | 控制台创建 | 过期、空格、引号 |
| Model ID | 有效模型标识 | 拼写错误、用了不存在的模型 |
CC Switch、Cline MCP、Codex auth.json 这类客户端配置,核心都是把这三件套填对。以 Codex 的auth.json为例:
{ "base_url": "https://taotoken.net/api", "api_key": "sk-你的TaoToken密钥", "model": "claude-3-5-sonnet" }字段名可能因客户端而异,但值就是这三样。
6. 接入 TaoToken:统一 Key 通道与后续扩展
Server 跑通之后,最后一步是把它接入 TaoToken 的统一通道,让模型调用也走同一个入口。
6.1 统一 Key 通道的价值
自己写 MCP Server 时,工具逻辑和模型调用是两件事。工具逻辑用 JSON-RPC 暴露,模型调用用 TaoToken 的 API。把模型调用收敛到 TaoToken 之后,切换模型只需要改model_id,Base URL 和 Key 都不用动。这对需要频繁对比不同模型效果的场景特别有用。
6.2 接入步骤
第一步,去 TaoToken 控制台创建 API Key。入口在 API Keys 页面。
第二步,把三件套写进 MCP Server 的配置或环境变量。参考第 3 章的config.toml和settings.json片段。
第三步,在 MCP Server 内部调用模型时,用统一的 Base URL:
import os import httpx BASE_URL = os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api") API_KEY = os.environ.get("TAOTOKEN_API_KEY") MODEL_ID = os.environ.get("TAOTOKEN_MODEL_ID", "claude-3-5-sonnet") async def call_model(prompt: str) -> str: async with httpx.AsyncClient() as client: resp = await client.post( f"{BASE_URL}/v1/chat/completions", headers={"Authorization": f"Bearer {API_KEY}"}, json={ "model": MODEL_ID, "messages": [{"role": "user", "content": prompt}], }, timeout=60, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"]第四步,验证调用。跑一次call_model("你好"),能返回文本就说明通道通了。
6.3 后续扩展方向
Server 骨架有了,统一通道也有了,接下来可以往几个方向扩展。一是把 stdio 换成 SSE,部署成远程服务,挂到反向代理后面。二是写一个 Gateway 把多个 MCP Server 聚合起来,统一暴露给客户端。三是除了 Tools,再暴露 Resources 和 Prompts,让 AI 能读取结构化数据和套用模板。四是在生产环境加上 API Key 校验和调用日志。
模型对话调试可以在模型对话页面做,长期编码或 Agent 场景可以看 Coding Plan,接入文档在文档页面,创建 Key 在 API Keys 页面。
6.4 一个实用技巧
开发阶段用python3 -m mcp run直接测 Server,不要急着挂客户端。等 JSON-RPC 请求响应都正常了,再写settings.json挂上去。这样出问题能快速定位是 Server 逻辑问题还是客户端配置问题。另外,Tool 的 description 一定要写详细,说清楚「什么场景调用」「返回什么」「有什么限制」,模型才会在正确的时候调用你的工具。描述太简略,模型会直接忽略。