MCP Server 从零到生产：用 Python 搭建自己的 AI 工具箱并接入 TaoToken

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然后安装 SDKpip 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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基础地址不加 UTMhttps://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 URLhttps://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, encodingutf-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, indent2, ensure_asciiFalse) if __name__ __main__: os.makedirs(ALLOWED_ROOT, exist_okTrue) mcp.run(transportstdio)这段代码的关键点有三个_is_path_safe做路径逃逸防护mcp.tool的 description 写清楚调用场景mcp.run(transportstdio)走标准输入输出通信。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, indent2, ensure_asciiFalse, defaultstr) 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, indent2, ensure_asciiFalse) except Exception as e: return f查询失败{e} if __name__ __main__: mcp.run(transportstdio)_validate_query是安全核心只放行 SELECT把写操作全部拦掉。AI 生成的 SQL 不可信这层校验必须有。4. 验证请求stdio 联调与成功结果确认代码写完不算完必须验证 Server 真的能被调用。这一章给出完整的联调步骤和预期输出。4.1 用 MCP CLI 直接测试MCP SDK 自带命令行工具可以直接跑 Serverpython3 -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_fileecho {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 URLhttps://taotoken.net/api多写路径、少写 /apiAPI 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_idBase URL 和 Key 都不用动。这对需要频繁对比不同模型效果的场景特别有用。6.2 接入步骤第一步去 TaoToken 控制台创建 API Key。入口在 API Keys 页面。第二步把三件套写进 MCP Server 的配置或环境变量。参考第 3 章的config.toml和settings.json片段。第三步在 MCP Server 内部调用模型时用统一的 Base URLimport 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: fBearer {API_KEY}}, json{ model: MODEL_ID, messages: [{role: user, content: prompt}], }, timeout60, ) 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 一定要写详细说清楚「什么场景调用」「返回什么」「有什么限制」模型才会在正确的时候调用你的工具。描述太简略模型会直接忽略。