☰
第3讲:用 Python SDK 手写第一个 MCP Server,把本地工具接进 TaoToken
2026/10/2 6:18:23 网站建设 项目流程

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,进入控制台创建 Key。控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ,Key 管理页在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。创建后复制那串以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_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,里面有 Base URL、Key、Model ID 三件套的填法。

如果你只是想先验证 Server 能不能跑,不接模型也行,用后面的 Python Client 脚本就能测。但要做成"能被 Agent 发现并调用"的完整链路,TaoToken 这一侧必须先通。我建议的顺序是:先拿 Key,再写 Server,最后用 Client 验证。

关于 Coding Plan,如果你打算长期做编码类 Agent,可以看 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ,它适合需要持续调用、多工具协作的场景。短期验证用按量 Key 就够了。

模型对话入口在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite ,想先确认模型能不能正常返回,可以在这里发一条消息试试,确认 Key 有效再往下走。

3. 可复制配置:server.py 骨架与依赖清单

这一节是核心,直接给可复制的代码。先建数据库,再写 Server。

3.1 初始化 SQLite 测试库

新建setup_db.py:

import 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.txt:

mcp>=1.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( name="query_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( name="read_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(type="text", text=f"未知工具:{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(type="text", text="错误:仅支持 SELECT 查询语句")] dangerous_keywords = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "CREATE", "EXEC"] for kw in dangerous_keywords: if kw in sql_trimmed: return [types.TextContent(type="text", text=f"错误:查询中包含被禁止的关键字 '{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( type="text", text=json.dumps(result, ensure_ascii=False, indent=2) )] except Exception as e: return [types.TextContent(type="text", text=f"查询失败:{str(e)}")] async def _read_file(arguments: dict) -> list[types.TextContent]: path = arguments.get("path", "") if ".." in path: return [types.TextContent(type="text", 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( type="text", text=f"错误:不支持读取该文件类型。支持的格式:{', '.join(allowed_extensions)}" )] try: if not os.path.exists(path): return [types.TextContent(type="text", text=f"错误:文件不存在:{path}")] with open(path, "r", encoding="utf-8") as f: content = f.read() return [types.TextContent(type="text", text=content)] except Exception as e: return [types.TextContent(type="text", text=f"读取失败:{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 连进来。按 Ctrl+C 停止。如果你看到它立刻退出,说明main()里少了async with,或者asyncio.run没包住。

4.2 用 Python Client 做完整验证

新建test_server.py:

import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def test(): server_params = StdioServerParameters( command="python", 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和args,args用绝对路径。如果 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_ascii=True。解决:改成json.dumps(data, ensure_ascii=False, indent=2),上面代码已经处理。

报错六: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_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 拿,Model ID 按控制台列表填。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,里面有各客户端的详细填法。想先确认模型通道正常,去 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 发条消息试试。长期做编码 Agent 的话,https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 更合适。

一个实用技巧:给 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 工具接进统一通道。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询