MCP协议开发实战：用TaoToken统一Key搭建AI Agent工具链的Client-Server骨架

1. 为什么你的 AI Agent 工具链总是拼不起来如果你正在做 AI Agent大概率遇到过这种局面模型能聊天但一让它“查个天气再写进文件”代码就开始散架。工具定义写死在 Agent 主逻辑里换一个模型供应商要改一遍加一个数据库查询又要改一遍最后 Agent 主体变成一坨谁都不敢动的胶水代码。MCPModel Context Protocol模型上下文协议想解决的就是这件事。它把“工具”从 Agent 里拆出来变成独立的 Server 进程用 JSON-RPC 标准化通信。Agent 只负责调度工具只负责执行两边通过 Client-Server 骨架对接。你可以把它理解成 AI 世界的 USB-C工具开发者实现一次 MCP Server任何支持 MCP Client 的 Agent 都能即插即用。这篇要带你跑通的是最小闭环本地起一个 MCP Server 暴露工具用 TaoToken 统一 Key 作为模型通道写一个 MCP Client 完成一次“请求-工具调用-响应”的完整链路。适合已经写过基础 Agent、想把手上的工具调用整理成可扩展骨架的开发者。全程可复制配置文件和 SDK 初始化片段都会给全。2. TaoToken 在 MCP 工具链里的位置MCP 协议本身只管工具怎么暴露、怎么调用它不负责模型推理。Agent 的“大脑”仍然需要一个大模型来决策当前用户请求要不要调工具、调哪个、参数是什么。这一步就需要一个稳定的模型 API 通道。TaoToken 在这里扮演的是统一 Key 的角色。你不需要在 Agent 代码里为每个模型供应商维护一套鉴权和 base_url而是通过一个 API 通道接入模型对话、工具调用决策都走同一个入口。对 MCP 工具链来说这意味着 Client 侧的 LLM 初始化可以收敛成一处配置后面换模型只改配置不改调度逻辑。具体接入时模型对话走https://taotoken.net/apiKey 在控制台的 API Keys 页面生成。如果你后面要做长期编码类 Agent可以关注 Coding Plan只是验证模型工具调用能力用模型对话入口就够。官网入口是https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI Keys 管理在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_campaignrewrite接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_campaignrewrite。注意MCP Server 是本地进程不要把它写成直连生产数据库的通道。工具权限要单独收口后面排障章节会讲。3. 可复制骨架config.toml 与 settings.json先把工程结构定下来。MCP 的 Client-Server 骨架建议拆成三块Client 主体、Server 工具目录、配置文件。配置文件我习惯用config.toml管模型通道用settings.json管 MCP Server 注册两者职责分开改一个不影响另一个。3.1 config.toml模型通道与运行参数# config.toml [llm] provider taotoken base_url https://taotoken.net/api api_key sk-你的TaoTokenKey model claude-3-5-sonnet timeout 60 max_tokens 4096 [agent] max_iterations 10 tool_timeout 30 log_level INFO [mcp] settings_path ./settings.json这里base_url固定指向 TaoToken 的 API 通道api_key从控制台拿。max_iterations是 Agent 循环上限防止工具调用死循环tool_timeout是单个工具执行超时避免某个 Server 卡死拖垮整条链。3.2 settings.jsonMCP Server 注册表{ mcpServers: { weather: { command: python, args: [./mcp_servers/weather_server/server.py], env: { PYTHONUNBUFFERED: 1 } }, file: { command: python, args: [./mcp_servers/file_server/server.py], env: { WORKSPACE: ./workspace } } } }command和args决定 Client 怎么拉起 Server 进程env用来传工具自己的环境变量。注意file服务里我加了WORKSPACE限制工具只能在这个目录下读写这是最小权限原则的落地方式。3.3 SDK 初始化片段Client 侧初始化分两步先读配置再建立 MCP 会话。下面这段是核心骨架可以直接放进mcp_client/agent.py。# mcp_client/agent.py import asyncio import json import tomllib from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from openai import AsyncOpenAI class MCPAgent: def __init__(self, config_path: str ./config.toml): with open(config_path, rb) as f: self.config tomllib.load(f) llm_cfg self.config[llm] self.llm AsyncOpenAI( base_urlllm_cfg[base_url], api_keyllm_cfg[api_key], ) self.model llm_cfg[model] self.max_iterations self.config[agent][max_iterations] with open(self.config[mcp][settings_path], r) as f: self.mcp_settings json.load(f)[mcpServers] self.sessions {} self.tools [] async def initialize(self): for name, cfg in self.mcp_settings.items(): params StdioServerParameters( commandcfg[command], argscfg[args], envcfg.get(env, {}), ) read, write await stdio_client(params).__aenter__() session await ClientSession(read, write).__aenter__() await session.initialize() tool_list await session.list_tools() self.sessions[name] session for t in tool_list.tools: self.tools.append({ type: function, function: { name: f{name}__{t.name}, description: t.description, parameters: t.inputSchema, }, }) print(f[MCP] {name} 已连接工具数 {len(tool_list.tools)})这里有个细节工具名我做了server__tool的前缀拼接因为不同 Server 可能有同名工具加前缀后路由不会冲突。inputSchema直接来自 MCP Server 的声明不需要手写。4. 一次请求-响应验证从用户提问到工具执行骨架搭好后最关键的是验证链路真的通了。我设计一个最小任务用户问“郑州天气怎么样把结果写到 weather.txt”。这条任务需要两个工具协作能同时验证工具发现、参数传递、结果回填三个环节。4.1 Agent 主循环async def run(self, user_query: str) - str: messages [{role: user, content: user_query}] for i in range(self.max_iterations): resp await self.llm.chat.completions.create( modelself.model, messagesmessages, toolsself.tools, tool_choiceauto, ) msg resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) for call in msg.tool_calls: server_name, tool_name call.function.name.split(__, 1) args json.loads(call.function.arguments) session self.sessions.get(server_name) if not session: result f错误未找到服务 {server_name} else: raw await session.call_tool(tool_name, args) result raw.content[0].text if raw.content else messages.append({ role: tool, tool_call_id: call.id, content: result, }) print(f[Tool] {server_name}.{tool_name} - {result[:80]}) return 达到最大迭代次数任务未完成4.2 天气 Server 的最小实现# mcp_servers/weather_server/server.py import asyncio from mcp.server import Server from mcp.server.stdio import stdio_server app Server(weather-server) WEATHER { 郑州: 郑州 当前气温24℃多云微风, 北京: 北京 当前气温22℃晴东北风2级, } app.tool( nameget_weather, description查询指定城市的实时天气, inputSchema{ type: object, properties: { city: {type: string, description: 城市名称} }, required: [city], }, ) async def get_weather(city: str) - str: return WEATHER.get(city, f{city} 天气信息暂不可用) async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: asyncio.run(main())4.3 启动与验证# 终端 1不需要单独启动 ServerClient 会自动拉起 python -m mcp_client.agent预期输出类似[MCP] weather 已连接工具数 1 [MCP] file 已连接工具数 2 [Tool] weather.get_weather - 郑州 当前气温24℃多云微风 [Tool] file.write_file - 已写入 ./workspace/weather.txt 任务结果郑州当前天气为24℃多云已写入 weather.txt看到[Tool]两行说明链路通了模型先决策调天气工具拿到结果后再决策调文件工具最后汇总回复。整个过程 Agent 主体没有硬编码任何工具逻辑全靠 MCP 声明和模型决策。5. 本篇常见错排查5.1 Server 启动即退出Client 报连接失败最常见的原因是 Server 脚本里stdio_server没包在async with里或者app.run的参数不对。检查main()是否完整另外PYTHONUNBUFFERED1建议加上否则 stdout 缓冲会让 Client 读不到初始化消息。5.2 工具列表为空如果list_tools返回 0 个工具先确认装饰器用的是app.tool而不是app.list_tools。另一个坑是inputSchema写成了input_schemaMCP 的字段名是驼峰写错不会报错但工具不会注册。5.3 模型不调工具直接编答案这通常是tool_choice没设成auto或者工具描述太模糊。把description写清楚“查询指定城市的实时天气”比“天气工具”有效得多。如果模型仍然不调检查tools数组是否真的传进了请求有时候配置读错路径会导致空数组。5.4 工具调用超时单个 Server 卡住会拖垮整个循环。在call_tool外面包一层asyncio.wait_for超时时间从config.toml的tool_timeout读。另外文件类工具一定要限制工作目录避免误操作大文件。5.5 Key 鉴权失败如果模型请求返回 401先确认config.toml里的base_url是https://taotoken.net/api不要带多余路径。Key 从 API Keys 页面重新复制一次注意不要带空格。接入细节可以对照接入文档核对请求头格式。6. 把骨架跑成你自己的工具链到这里最小闭环已经通了。接下来扩展的方向很明确每加一个能力就写一个独立 MCP Server在settings.json里注册一行Client 不需要改代码。比如加一个 SQL 查询 Server只暴露只读查询工具参数做白名单校验加一个 HTTP 请求 Server限制目标域名。工具之间通过 Agent 的循环自然串联上一个工具的输出可以作为下一个工具的输入。如果你要长期跑编码类 Agent建议把 Coding Plan 纳入通道规划避免频繁切换 Key。验证模型工具调用能力时用模型对话入口快速试接入和排障阶段对照接入文档能省不少时间。骨架代码建议先跑通再重构别一上来就追求抽象MCP 的价值在于工具解耦不在于 Client 写得多优雅。