☰
MCP Server 从零到生产:用 Python 搭建自己的 AI 工具箱并接入 TaoToken
2026/10/2 17:06:54 网站建设 项目流程

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 mcp

2.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 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", 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 URLhttps://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 一定要写详细,说清楚「什么场景调用」「返回什么」「有什么限制」,模型才会在正确的时候调用你的工具。描述太简略,模型会直接忽略。

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

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

立即咨询