1. 从零手写 MCP Server为什么本地工具接进统一通道才是关键MCP Server 是什么简单说它是一个遵循 Model Context Protocol 的进程对外暴露若干 Tool工具和 Resource资源让 AI 客户端能像调用函数一样调用你本地的能力。能做什么把查数据库、读文件、跑脚本这些本地操作包装成模型可发现、可调用的标准接口。适合谁适合已经会用 Python、想让自己的本地工具被 AI Agent 直接调用的开发者。前两讲我们把 Transport、Tool、Resource 的概念过了一遍也看了一段最小 Server 的代码。但概念归概念真正动手时你会发现几个绕不开的问题工具怎么声明才能被正确发现stdio 模式下为什么启动后终端一片空白本地跑通了怎么把它接到统一的 Key/API 通道上而不是每个客户端配一遍这一讲就解决这些。我会带你手写一个能用的 MCP Server包含两个真实工具query_database和read_file。写完你手上会有一个可以被任何 MCP Client 连接的工具服务器并且知道怎么把它接入 TaoToken 的统一通道用一套 Key 管理所有模型调用。先说清楚整体链路。MCP Server 本身不负责调用大模型它只负责提供工具。真正决定调哪个工具的是 MCP Client 背后的 Agent。所以我们的 Server 要做两件事第一通过list_tools告诉 Client 我有哪些工具第二通过call_tool接收调用请求并返回结果。这两件事做对了工具就能被正确发现和调用。我试过把工具描述写得含糊结果 Agent 该调query_database的时候去调了read_file排查半天才发现是 description 没写清楚适用场景。所以下面每个工具的 description 我都会写明白适用场景和注意事项这不是凑字数是直接影响调用准确率的东西。环境准备上你需要 Python 3.10 以上以及 mcp SDK 1.x。先确认版本pip install mcp --upgrade python -c import mcp; print(mcp.__version__)输出应该是 1.x比如 1.2.0。如果低于 1.0升级一下老版本的 API 签名和新版不一致照抄代码会报错。项目结构保持简单mcp-server-demo/ ├── server.py # MCP Server 主文件 ├── setup_db.py # 数据库初始化脚本 ├── demo.db # SQLite 数据库自动生成 └── sample.txt # 示例文件手动创建sample.txt内容随便写几行比如这是一个示例文件。 MCP Server 可以通过 read_file 工具读取它。 第3讲手写第一个 MCP Server。到这里准备工作就完成了。接下来先建数据库再写 Server 主体。数据库不是必须的但用一个带真实数据的库来测试比空表更能验证查询逻辑对不对。2. TaoToken 前置准备统一 Key 与 API 通道怎么配在写 Server 之前先把 TaoToken 这一侧准备好。为什么要先做这步因为 MCP Server 写完后你要验证它能不能被真实 Agent 调用而 Agent 背后需要模型。TaoToken 提供统一的 Key 和 API 通道你申请一次后面所有客户端都用同一套配置不用每个工具单独配。先注册并拿到 API Key。打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 进入控制台创建 Key。控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite Key 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。创建后复制那串以sk-开头的 Key只显示一次丢了就重新建。Base URL 用 https://taotoken.net/api 注意这个地址不加 UTM 参数直接填。Model ID 按你实际要用的模型填比如claude-sonnet-4-5或gpt-4o这类具体以控制台模型列表为准。这里有个关键点MCP Server 本身不直接调模型它只提供工具。所以 TaoToken 的 Key 是配在 MCP Client 那一侧的不是配在 Server 里。很多人第一次做会搞混把 Key 写进 server.py结果发现根本用不上。记住分工Server 管工具Client 管模型和 Key。那为什么这一讲要提前讲 TaoToken因为验证环节需要一个能调工具的 Agent。你可以用 Claude Code、Cline 这类支持 MCP 的客户端把它们的模型通道指向 TaoToken这样工具调用和模型调用走的是同一套体系。Claude Code 的接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有 Base URL、Key、Model ID 三件套的填法。如果你只是想先验证 Server 能不能跑不接模型也行用后面的 Python Client 脚本就能测。但要做成能被 Agent 发现并调用的完整链路TaoToken 这一侧必须先通。我建议的顺序是先拿 Key再写 Server最后用 Client 验证。关于 Coding Plan如果你打算长期做编码类 Agent可以看 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它适合需要持续调用、多工具协作的场景。短期验证用按量 Key 就够了。模型对话入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 想先确认模型能不能正常返回可以在这里发一条消息试试确认 Key 有效再往下走。3. 可复制配置server.py 骨架与依赖清单这一节是核心直接给可复制的代码。先建数据库再写 Server。3.1 初始化 SQLite 测试库新建setup_db.pyimport sqlite3 def setup_database(): conn sqlite3.connect(demo.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS employees ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, department TEXT NOT NULL, salary REAL NOT NULL, hire_date TEXT NOT NULL ) ) cursor.execute(DELETE FROM employees) employees_data [ (1, 张三, 技术部, 25000, 2022-03-15), (2, 李四, 市场部, 20000, 2023-01-10), (3, 王五, 技术部, 28000, 2021-07-01), (4, 赵六, 人事部, 22000, 2023-09-20), (5, 孙七, 技术部, 32000, 2020-05-12), (6, 周八, 市场部, 18000, 2024-02-28), ] cursor.executemany( INSERT INTO employees VALUES (?, ?, ?, ?, ?), employees_data ) cursor.execute( CREATE TABLE IF NOT EXISTS sales ( id INTEGER PRIMARY KEY, product TEXT NOT NULL, amount REAL NOT NULL, sale_date TEXT NOT NULL, employee_id INTEGER, FOREIGN KEY (employee_id) REFERENCES employees(id) ) ) cursor.execute(DELETE FROM sales) sales_data [ (1, 笔记本电脑, 8999, 2025-01-15, 1), (2, 显示器, 2499, 2025-01-16, 3), (3, 键盘, 599, 2025-02-01, 5), (4, 笔记本电脑, 7999, 2025-02-10, 1), (5, 鼠标, 299, 2025-02-15, 3), (6, 显示器, 2199, 2025-03-01, 5), (7, 笔记本电脑, 8999, 2025-03-10, 1), (8, 键盘, 499, 2025-03-20, 3), ] cursor.executemany( INSERT INTO sales VALUES (?, ?, ?, ?, ?), sales_data ) conn.commit() conn.close() print(数据库初始化完成) if __name__ __main__: setup_database()运行python setup_db.py看到数据库初始化完成即可。3.2 依赖清单新建requirements.txtmcp1.0.0安装pip install -r requirements.txt。mcp SDK 自带 asyncio 支持不需要额外装。3.3 Server 骨架新建server.py完整代码如下import sqlite3 import os import json import asyncio from mcp.server import Server from mcp.server.stdio import stdio_server import mcp.types as types server Server(demo-server) server.list_tools() async def handle_list_tools() - list[types.Tool]: return [ types.Tool( namequery_database, description执行只读 SQL 查询返回 JSON 格式的结果。 适用场景 - 查询员工信息、部门分布、薪资统计 - 查询销售记录、产品销售排行 - 任何 SELECT 查询 注意仅支持 SELECT 语句不支持 INSERT/UPDATE/DELETE。 返回格式JSON 数组每个元素是一条记录。, inputSchema{ type: object, properties: { sql: { type: string, description: SQL 查询语句例如SELECT * FROM employees LIMIT 5 } }, required: [sql] } ), types.Tool( nameread_file, description读取指定文件的内容。 适用场景 - 读取配置文件、日志文件 - 读取代码文件 - 读取文本格式的文档 注意仅能读取当前目录及其子目录下的文件。 支持格式.txt .py .json .yaml .yml .md .csv .log, inputSchema{ type: object, properties: { path: { type: string, description: 文件路径相对于当前工作目录 } }, required: [path] } ), ] server.call_tool() async def handle_call_tool(name: str, arguments: dict) - list[types.TextContent]: if name query_database: return await _query_database(arguments) elif name read_file: return await _read_file(arguments) else: return [types.TextContent(typetext, textf未知工具{name})] async def _query_database(arguments: dict) - list[types.TextContent]: sql arguments.get(sql, ) sql_trimmed sql.strip().upper() if not sql_trimmed.startswith(SELECT): return [types.TextContent(typetext, text错误仅支持 SELECT 查询语句)] dangerous_keywords [DROP, DELETE, INSERT, UPDATE, ALTER, CREATE, EXEC] for kw in dangerous_keywords: if kw in sql_trimmed: return [types.TextContent(typetext, textf错误查询中包含被禁止的关键字 {kw})] try: conn sqlite3.connect(demo.db) conn.row_factory sqlite3.Row cursor conn.cursor() cursor.execute(sql) columns [desc[0] for desc in cursor.description] rows cursor.fetchall() conn.close() result [] for row in rows: result.append(dict(zip(columns, row))) return [types.TextContent( typetext, textjson.dumps(result, ensure_asciiFalse, indent2) )] except Exception as e: return [types.TextContent(typetext, textf查询失败{str(e)})] async def _read_file(arguments: dict) - list[types.TextContent]: path arguments.get(path, ) if .. in path: return [types.TextContent(typetext, text错误不允许访问上级目录)] allowed_extensions (.txt, .py, .json, .yaml, .yml, .md, .csv, .log, .ini, .cfg) if not any(path.endswith(ext) for ext in allowed_extensions): return [types.TextContent( typetext, textf错误不支持读取该文件类型。支持的格式{, .join(allowed_extensions)} )] try: if not os.path.exists(path): return [types.TextContent(typetext, textf错误文件不存在{path})] with open(path, r, encodingutf-8) as f: content f.read() return [types.TextContent(typetext, textcontent)] except Exception as e: return [types.TextContent(typetext, textf读取失败{str(e)})] async def main(): async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options() ) if __name__ __main__: asyncio.run(main())几个设计点说明。list_tools返回的 description 里我特意写了适用场景和注意这是给 Agent 看的写得越具体工具选择越准。call_tool用 name 分发未知工具返回提示而不是抛异常避免 Client 侧直接崩。_query_database做了两层防护只允许 SELECT 开头且禁止危险关键字防止 Agent 误生成写操作。_read_file禁止..路径穿越并限制扩展名白名单。3.4 接入 TaoToken 的配置片段如果你用 Claude Code 或 Cline 这类客户端MCP Server 的注册配置通常长这样以 JSON 为例路径按你实际项目改{ mcpServers: { demo-server: { command: python, args: [/absolute/path/to/mcp-server-demo/server.py], env: {} } } }注意args里用绝对路径stdio 模式下相对路径容易因为工作目录不同而找不到文件。模型侧的配置Base URL、Key、Model ID填在客户端自己的设置里Base URL 用 https://taotoken.net/api Key 用你申请的那串Model ID 按控制台列表填。这三件套分开填别混进 MCP 配置里。4. 验证请求确认工具能被发现与调用Server 写完了怎么确认它真的能用两种方式先手动启动看行为再用 Client 脚本做完整验证。4.1 手动启动观察python server.py正常情况下终端不会有任何输出会卡住——这是对的因为 Server 在等 Client 通过 stdio 连进来。按 CtrlC 停止。如果你看到它立刻退出说明main()里少了async with或者asyncio.run没包住。4.2 用 Python Client 做完整验证新建test_server.pyimport asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def test(): server_params StdioServerParameters( commandpython, args[server.py] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(f可用工具 ({len(tools.tools)} 个):) for tool in tools.tools: print(f - {tool.name}: {tool.description[:50]}...) print(\n--- 测试 query_database ---) result await session.call_tool(query_database, { sql: SELECT name, department, salary FROM employees ORDER BY salary DESC LIMIT 3 }) print(result.content[0].text) print(\n--- 测试 read_file ---) result await session.call_tool(read_file, { path: sample.txt }) print(result.content[0].text) print(\n--- 测试安全限制应该被拒绝---) result await session.call_tool(query_database, { sql: DROP TABLE employees }) print(result.content[0].text) if __name__ __main__: asyncio.run(test())运行python test_server.py预期输出可用工具 (2 个): - query_database: 执行只读 SQL 查询返回 JSON 格式的结果... - read_file: 读取指定文件的内容... --- 测试 query_database --- [ { name: 孙七, department: 技术部, salary: 32000 }, { name: 王五, department: 技术部, salary: 28000 }, { name: 张三, department: 技术部, salary: 25000 } ] --- 测试 read_file --- 这是一个示例文件。 MCP Server 可以通过 read_file 工具读取它。 第3讲手写第一个 MCP Server。 --- 测试安全限制应该被拒绝--- 错误查询中包含被禁止的关键字 DROP看到这个输出说明三件事都对了工具被正确发现list_tools 返回 2 个、工具被正确调用查询和读文件都返回了数据、安全限制生效DROP 被拦下。4.3 用 MCP Inspector 可视化验证如果你想要图形界面SDK 自带一个 Web 调试工具npm install -g modelcontextprotocol/inspector npx modelcontextprotocol/inspector python server.py浏览器打开 http://localhost:5173 能看到 Server 连接状态、Tools 列表点进工具填参数就能 Call Tool。这个方式适合快速看工具 schema 长什么样。4.4 接入 TaoToken 后的端到端验证前面是本地验证。要验证接入 TaoToken 统一通道这条链路把 MCP Server 注册到支持 MCP 的客户端里客户端的模型通道指向 TaoToken。然后在对话里让它查一下技术部薪资最高的三个人观察它是否自动调用了query_database。如果调用了并返回正确结果说明工具发现、调用、模型通道三件事全通了。这一步如果模型没调工具通常是 description 不够明确或者客户端没正确加载 MCP 配置。先确认客户端日志里有没有 demo-server connected 之类的记录。5. 常见报错排查401、local proxy failed、reading choices 逐个解决这一节按真实报错来。下面这些是我和读者都踩过的对照着查。报错一401 Unauthorized现象客户端调用模型时返回 401。原因TaoToken 的 Key 没填、填错或者填到了 MCP 配置的 env 里而不是模型设置里。解决确认 Key 以sk-开头填在客户端的模型 API Key 字段Base URL 是 https://taotoken.net/api 。MCP Server 的配置里不需要放这个 Key。报错二local proxy failed / connection refused现象Client 连 Server 时报连接失败。原因stdio 模式下 Client 用 TCP 方式去连或者 Server 路径写错。解决stdio 不走网络Client 必须用stdio_client启动子进程。检查 MCP 配置里的command和argsargs用绝对路径。如果 Server 手动启动就立刻退出回到 4.1 检查async with stdio_server()。报错三reading choices / 返回结构解析失败现象调用工具后客户端报解析错误。原因call_tool返回的不是list[types.TextContent]或者返回了 None。解决确认每个分支都返回 TextContent 列表包括错误分支。上面代码里未知工具、查询失败、文件不存在都返回了 TextContent不要图省事直接return。报错四工具调用返回空结果现象query_database返回[]。原因SQL 执行成功但表里没数据或者连的库不对。解决先手动sqlite3 demo.db SELECT * FROM employees LIMIT 3;确认数据在。注意sqlite3.connect(demo.db)是相对路径Server 的工作目录如果不是项目目录会连到别的库或新建空库。稳妥做法是用绝对路径拼数据库位置。报错五中文乱码现象返回的 JSON 里中文变成\u5f20\u4e09。原因json.dumps默认ensure_asciiTrue。解决改成json.dumps(data, ensure_asciiFalse, indent2)上面代码已经处理。报错六OAuth / 认证相关报错现象客户端提示需要 OAuth 或认证失败。原因某些客户端对远程 MCP 有 OAuth 要求但本地 stdio Server 不需要。解决确认你用的是 stdio 模式而不是 HTTP/SSE 模式。本地调试阶段不需要任何 OAuth 配置把客户端的远程 MCP 选项关掉。报错七ModuleNotFoundError: No module named mcp现象启动 Server 报找不到 mcp。原因装到了别的 Python 环境。解决which python和pip -V确认是同一个环境或者用python -m pip install mcp显式指定。排查顺序建议先手动启动 Server 看是否卡住再用 test_server.py 本地验证最后接客户端。每一步都过了再往下不要跳步否则报错来源分不清是 Server 还是 Client。6. 把工具接进统一通道下一步怎么走到这里你手上有一个能跑的 MCP Server两个工具都能被正确发现和调用安全限制也生效了。接下来把它接到 TaoToken 统一通道用一套 Key 管理模型调用。具体动作在支持 MCP 的客户端里注册这个 Server模型侧填 TaoToken 的三件套——Base URL 用 https://taotoken.net/api Key 从 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 拿Model ID 按控制台列表填。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有各客户端的详细填法。想先确认模型通道正常去 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 发条消息试试。长期做编码 Agent 的话https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 更合适。一个实用技巧给 Server 加工具时description 里把什么时候该用我写清楚比写我能做什么更重要。Agent 选工具靠的是场景匹配不是功能罗列。另外call_tool的错误分支一定要返回 TextContent 而不是抛异常否则 Client 侧会直接断连排查起来很痛苦。下一步可以试着给 Server 加第三个工具比如get_table_schema(table_name)返回表结构或者把read_file的扩展名白名单加上.html和.css。加完用 test_server.py 再跑一遍确认新工具出现在 list_tools 里、能被 call_tool 正确分发。这个循环跑顺了你就能把任何本地能力包装成 MCP 工具接进统一通道。
