1. 从零实现 MCP Server 到底解决什么问题MCP Server 是什么一句话说清它把你的数据库、搜索引擎、外部 API 包装成 Agent 能直接调用的“工具”让模型不再只会聊天而是能真正动手干活。适合谁适合已经跑通基础对话、想让 Agent 接入真实数据源的开发者尤其是做数据分析、自动化报告、内部工具链的同学。我见过太多人卡在同一个地方Agent 能写代码但拿不到真实数据。你问它“上个月订单量多少”它只能编。你让它“查一下竞品最新动态”它只能瞎猜。根因不是模型不行而是它没有一条安全、可控、可审计的通道去触碰外部世界。MCPModel Context Protocol就是这条通道的标准协议。这一讲的目标很明确从零搭一个可运行的 MCP Server注册数据库查询、搜索引擎、文件读写、图表生成四类工具再通过 TaoToken 统一 Key 和 API 通道让 Agent 一次配置就能调用全部能力。全程本地可复现不需要复杂基础设施。先说清楚架构分层。最底层是数据源SQLite 数据库、搜索 API、本地文件系统。中间层是 MCP Server它用server.tool()装饰器把每个能力注册成标准工具附带清晰的描述和参数 schema。最上层是 Agent通过 MCP 客户端连接 Server拿到工具列表后由模型自主决策调用哪个。TaoToken 在这里的角色是统一模型入口——你不需要为每个模型单独配 Key一个通道搞定对话、工具调用、流式返回。为什么不用传统的 Function Calling 直接写因为 MCP 把工具定义和 Agent 运行时解耦了。你写一次 ServerClaude Code、Cline、Codex 都能接。工具描述、参数校验、错误格式全部标准化换模型不用重写。这是工程上的关键收益。本篇会给出完整项目结构、可复制代码、TaoToken 配置片段、本地启动命令、端到端验证流程以及真实会遇到的报错排查。跟着做30 分钟内你能得到一个能查库、能搜索、能写文件、能画图的 Agent 工具链。2. TaoToken 前置准备与统一 Key 配置在写 Server 之前先把模型通道打通。TaoToken 提供统一的 API 入口兼容 OpenAI 风格的调用方式Agent 侧只需要配 Base URL、API Key、Model ID 三件套。这样你的 MCP Server 专注做工具模型调用交给统一通道职责清晰。第一步拿到 API Key。访问 https://taotoken.net/api-keys 创建密钥。建议按项目命名比如mcp-super-server方便后续审计和轮换。Key 只在创建时完整显示一次复制后存到环境变量不要硬编码进代码。第二步确认 Base URL。TaoToken 的 API 根地址是 https://taotoken.net/api 所有兼容 OpenAI 协议的客户端都填这个。注意末尾不要多加/v1具体路径由 SDK 拼接。第三步选 Model ID。做 Agent 工具调用建议选支持 function calling 的模型。你可以在 https://taotoken.net/models 查看可用列表也可以在模型对话页 https://taotoken.net/chat 先手动试一轮确认模型能正确理解工具描述。配置写入环境变量Linux/macOS 用export TAOTOKEN_API_KEYsk-你的密钥 export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL你的模型IDWindows PowerShell$env:TAOTOKEN_API_KEYsk-你的密钥 $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api $env:TAOTOKEN_MODEL你的模型ID如果你用 Claude Code 或 Cline 这类工具配置方式略有不同。以 Claude Code 为例settings 文件里需要写全三件套。Cline 的 MCP 配置则在cline_mcp_settings.json中声明 Server 启动命令。Codex 用户注意auth.json里同样要填 Base URL 和 Key缺一不可。注意Key 不要提交到 Git。用.env文件加.gitignore或者直接用系统环境变量。生产环境建议走密钥管理服务。验证通道是否通跑一段最小请求import os from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL] ) resp client.chat.completions.create( modelos.environ[TAOTOKEN_MODEL], messages[{role: user, content: 回复 OK 两个字母}] ) print(resp.choices[0].message.content)返回OK就说明通道正常。这一步别跳过后面 Agent 调不通八成是这里没配好。如果你更想先体验模型能力再动手写代码可以去 https://taotoken.net/chat 直接对话确认模型对工具描述的理解程度。3. 可复制的 MCP Server 项目结构与配置片段项目结构先定下来后面所有代码往里填mcp_super_server/ ├── server.py # MCP Server 主文件 ├── config.py # 配置读取 ├── tools/ │ ├── database.py # 数据库工具 │ ├── search.py # 搜索工具 │ ├── filesystem.py # 文件工具 │ └── chart.py # 图表工具 ├── requirements.txt └── .env依赖清单requirements.txtmcp1.0.0 openai1.30.0 matplotlib3.8.0 python-dotenv1.0.0config.py负责读取环境变量统一出口import os from dotenv import load_dotenv load_dotenv() TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY) TAOTOKEN_BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) TAOTOKEN_MODEL os.getenv(TAOTOKEN_MODEL) DB_PATH os.getenv(DB_PATH, data.db)主 Server 文件server.py用server.tool()注册工具。先看数据库部分from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import asyncio, json, sqlite3 server Server(super-tool-server) server.tool() async def query_database(sql: str, db_path: str data.db) - list[TextContent]: 执行SQL查询仅支持SELECT。当用户需要查询数据库中的结构化数据时使用。 Args: sql: SQL查询语句如 SELECT * FROM users LIMIT 10 db_path: 数据库文件路径默认data.db if not sql.strip().upper().startswith(SELECT): return [TextContent(typetext, textjson.dumps( {success: False, error: 仅允许SELECT查询}))] try: conn sqlite3.connect(db_path) cursor conn.execute(sql) columns [d[0] for d in cursor.description] if cursor.description else [] rows [dict(zip(columns, row)) for row in cursor.fetchall()] conn.close() return [TextContent(typetext, textjson.dumps( {success: True, columns: columns, data: rows, total: len(rows)}, ensure_asciiFalse, indent2))] except Exception as e: return [TextContent(typetext, textjson.dumps( {success: False, error: str(e)}))]搜索工具接入外部 API这里给出接口结构实际 Key 从环境变量读import os, httpx server.tool() async def web_search(query: str, max_results: int 5) - list[TextContent]: 搜索互联网获取实时信息。当需要查询实时数据、新闻或不确定的事实时使用。 Args: query: 搜索关键词 max_results: 最大结果数默认5 api_key os.getenv(SEARCH_API_KEY) if not api_key: return [TextContent(typetext, textjson.dumps( {success: False, error: 未配置SEARCH_API_KEY}))] async with httpx.AsyncClient(timeout15) as client: resp await client.get( https://api.example-search.com/search, params{q: query, limit: max_results}, headers{Authorization: fBearer {api_key}} ) data resp.json() return [TextContent(typetext, textjson.dumps( {query: query, results: data.get(items, [])}, ensure_asciiFalse))]文件工具和图表工具按同样模式注册。文件读写要限制路径防止越权import os ALLOWED_ROOT os.path.abspath(os.getenv(WORK_DIR, .)) def _safe_path(p: str) - str: full os.path.abspath(p) if not full.startswith(ALLOWED_ROOT): raise ValueError(路径越界) return full server.tool() async def read_file(file_path: str) - list[TextContent]: 读取文本文件内容。当需要查看本地文件时使用。 Args: file_path: 文件路径 try: full _safe_path(file_path) with open(full, r, encodingutf-8) as f: content f.read() return [TextContent(typetext, textjson.dumps( {success: True, path: file_path, content: content, size: len(content)}, ensure_asciiFalse))] except Exception as e: return [TextContent(typetext, textjson.dumps( {success: False, error: str(e)}))]图表工具用 matplotlib 无头模式import matplotlib matplotlib.use(Agg) import matplotlib.pyplot as plt server.tool() async def create_chart(chart_type: str, data: str, title: str Chart, output_path: str chart.png) - list[TextContent]: 生成数据图表。支持line折线图、bar柱状图、pie饼图。 Args: chart_type: 图表类型 line/bar/pie data: JSON格式数据如 {labels:[A,B],values:[10,20]} title: 图表标题 output_path: 输出图片路径 chart_data json.loads(data) labels chart_data.get(labels, []) values chart_data.get(values, []) fig, ax plt.subplots(figsize(10, 6)) if chart_type line: ax.plot(labels, values, markero, linewidth2) elif chart_type bar: ax.bar(labels, values, colorskyblue) elif chart_type pie: ax.pie(values, labelslabels, autopct%1.1f%%) ax.set_title(title) plt.tight_layout() plt.savefig(output_path, dpi150) plt.close() return [TextContent(typetext, textjson.dumps( {success: True, path: output_path, type: chart_type}))]启动入口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())Agent 侧接入时TaoToken 三件套写进配置。以 Cline 的 MCP 配置为例cline_mcp_settings.json{ mcpServers: { super-tool-server: { command: python, args: [/绝对路径/mcp_super_server/server.py], env: { TAOTOKEN_API_KEY: sk-你的密钥, TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_MODEL: 你的模型ID, DB_PATH: /绝对路径/data.db } } } }Claude Code 的 settings 片段{ mcpServers: { super-tool-server: { command: python, args: [/绝对路径/mcp_super_server/server.py], env: { TAOTOKEN_API_KEY: sk-你的密钥, TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_MODEL: 你的模型ID } } } }Codex 的auth.json同样要填全 Base URL、Key、Model ID缺一项都会导致工具调用阶段鉴权失败。三件套是硬性要求不是可选项。4. 本地启动与端到端调用验证代码写完先本地启动验证。用官方 MCP Inspector 最直观npx modelcontextprotocol/inspector python server.py它会打开一个 Web 界面左侧列出所有注册的工具右侧可以手动填参数调用。先点list_tables参数db_path填你的数据库路径看返回的 JSON 里tables数组是否正确。再试query_databaseSQL 填SELECT * FROM users LIMIT 5确认data字段有内容。如果 Inspector 里工具列表是空的说明server.tool()装饰器没生效检查 mcp 版本是否 1.0.0。如果调用返回error先看错误信息常见的是路径不对或 SQL 语法问题。再用 Python 客户端做一次程序化验证这个脚本能直接复用到 CIimport asyncio, json from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def test(): params StdioServerParameters(commandpython, args[server.py]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具, [t.name for t in tools.tools]) r await session.call_tool(list_tables, {db_path: test.db}) print(数据库表, r.content[0].text) r await session.call_tool(write_file, { file_path: output/test.txt, content: Hello MCP! }) print(写文件, r.content[0].text) chart_data json.dumps({ labels: [Jan, Feb, Mar], values: [100, 200, 150] }) r await session.call_tool(create_chart, { chart_type: bar, data: chart_data, title: Monthly Sales, output_path: output/sales.png }) print(生成图表, r.content[0].text) asyncio.run(test())预期输出工具列表包含query_database、list_tables、web_search、read_file、write_file、create_chart等数据库表返回 JSON写文件返回bytes_written图表返回path和type。四项都通过说明 Server 本身没问题。最后做 Agent 端到端验证。用 LangGraph 的 ReAct Agent 接 MCP 工具from langchain_openai import ChatOpenAI from langchain_mcp_adapters.tools import load_mcp_tools from langgraph.prebuilt import create_react_agent from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import asyncio, os async def main(): params StdioServerParameters(commandpython, args[server.py]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await load_mcp_tools(session) model ChatOpenAI( modelos.environ[TAOTOKEN_MODEL], api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], temperature0 ) agent create_react_agent(model, tools) result await agent.ainvoke({ messages: [(user, 列出数据库里所有表然后查users表前5条数据)] }) print(result[messages][-1].content) asyncio.run(main())跑通后你会看到 Agent 先调list_tables再调query_database最后把结果整理成自然语言返回。这就是完整的工具链闭环。如果模型没调工具而是直接回答说明工具描述不够清晰回去改 docstring把“什么时候用”写具体。5. 本篇常见报错排查401 Unauthorized。最常见九成是 Key 没配对。检查三处环境变量是否真的导出echo $TAOTOKEN_API_KEY、Base URL 是否写成https://taotoken.net/api而不是带/v1、Model ID 是否在可用列表里。Cline 用户注意cline_mcp_settings.json的env块是否被正确读取有时候 JSON 格式错一个逗号就整块失效。local proxy failed / connection refused。MCP Server 启动失败或端口占用。先单独跑python server.py看有没有 import 错误。常见的是mcp包版本太旧server.tool()不存在升级到 1.0.0 以上。如果是 stdio 模式确认没有其他进程占用标准输入输出。reading choices 报错 / 返回结构解析失败。模型返回的 tool_calls 格式和客户端预期不一致。检查 Model ID 是否支持 function calling部分纯对话模型不支持工具调用。换一个明确支持工具调用的模型再试。另外确认langchain_mcp_adapters版本和mcp版本兼容。OAuth 相关报错。如果你接的是需要 OAuth 的远程 MCP Server本地 stdio 模式不涉及。但如果你把 Server 部署成 HTTP 模式鉴权头要带对。本地开发建议先用 stdio跑通再上远程。工具调用返回空 / Agent 不调工具。两个原因工具描述太模糊模型不知道何时用或者参数 schema 有歧义。把 docstring 里的“当用户需要…时使用”写清楚参数类型标注明确。实测下来描述质量直接决定调用准确率。路径越界错误。文件工具报“路径越界”说明WORK_DIR环境变量没设或设错。把它设成你允许操作的根目录绝对路径所有文件操作都会限制在这个范围内。图表生成失败。matplotlib 在无显示环境需要Agg后端代码里已经设了。如果还报错检查output_path的目录是否存在savefig不会自动建目录。先os.makedirs再保存。排查顺序建议先单独跑 Server再跑 Python 客户端最后接 Agent。每层验证通过再往上走不要一上来就调 Agent出错时定位困难。6. 把工具链跑起来之后Server 跑通只是起点。真正决定 Agent 好不好用的是工具描述的设计。我试过把query_database的描述从“执行SQL”改成“当用户询问具体数据、统计、筛选时使用仅支持SELECT”调用准确率明显提升。模型选工具靠的就是这段文字别偷懒。下一步可以做的给搜索工具加结果缓存避免重复请求给数据库工具加连接池高频查询时响应更快给文件工具加操作日志方便审计。这些优化不影响主流程但生产环境必备。如果你想让 Agent 长期跑编码任务或复杂 Agent 流程Coding Plan 更适合配额和稳定性针对长会话优化。日常验证模型能力模型对话页足够。接入文档在 https://taotoken.net/doc 有完整协议说明和示例。工具链的价值在于复用。你写一次 MCP ServerClaude Code、Cline、Codex 都能接换模型不用重写工具层。这才是 MCP 协议真正的工程意义。把今天这套跑通下一讲我们继续往工具箱里加代码执行和更复杂的组合工具。