1. 为什么你的 Agent 每加一个工具就要改代码先说一个我踩过的坑。去年做客服 Agent 的时候工具从 3 个涨到 17 个每加一个工具就要动三处代码写 handler、改 TOOLS 字典、调 prompt 里的工具描述。最要命的是上线之后发现某个工具参数写错了得重新发版。这种耦合度在生产环境里就是灾难。MCPModel Context Protocol要解决的就是这件事。它把工具从 Agent 代码里彻底剥离出来变成一个独立的 Server 进程Agent 启动时通过 JSON-RPC 2.0 协议去问 Server「你有哪些工具」拿到工具列表后自动注册到自己的工具系统里。整个过程 Agent 主逻辑一行不改。你可以把 MCP 理解成 Agent 的「应用商店」每个 MCP Server 就是一个 App里面装着若干工具Agent 是手机开机时扫描已安装的 App把里面的功能挂到桌面上。想加新功能装个新 App 就行不用刷机。这套协议适合谁三类人一是自建 Agent 框架、工具数量超过 5 个的开发者二是想让多个 Agent 共享同一套工具的后端团队三是想把内部系统数据库、工单、监控包装成标准工具对外暴露的平台方。如果你还在用硬编码的 function calling 列表工具数一超过 10 个就会开始难受。这一篇我会从零实现一个可运行的 MCP Server Client包含 JSON-RPC 通信层、工具注册、自动发现、热插拔验证脚本最后用 TaoToken 统一 Key 打通多工具调用的连通性验证。代码可以直接复制跑。2. TaoToken 前置准备统一 Key 与 API 通道在写 MCP 之前得先把模型调用这条链路理顺。MCP 解决的是「工具有哪些、怎么调」但工具调用的决策还是模型做的——模型得能看懂工具描述、能返回结构化的 tool_call。所以你需要一个稳定的模型 API 通道。我用 TaoToken 做统一入口原因是它把多个模型的 Key 收敛成一个Agent 侧只配一个 Base URL 和一个 Key切换模型不用改代码。对 MCP 场景特别友好工具描述会随 Server 变化模型得频繁重新理解工具集用统一通道省去多 Key 管理的麻烦。先拿 Key。打开 https://taotoken.net/api-keys 登录后创建一个 API Key复制出来形如sk-xxxxxxxx。这个 Key 后面会写进环境变量不要硬编码到代码里。然后确认你的调用地址。TaoToken 的 API 入口是https://taotoken.net/api兼容 OpenAI 的/v1/chat/completions格式。也就是说你原来用 openai SDK 的代码只改base_url和api_key两个字段就能跑。模型 ID 这块要注意MCP 场景下模型必须支持 function calling / tool use否则工具描述传过去它也不认。选模型时优先挑带 tool 能力的具体可用列表在 https://taotoken.net/models 查。我实测下来工具数量在 20 个以内时主流模型的工具选择准确率都够用。环境变量这样配export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL你的模型ID如果你用 Claude Code 或者 Cline 这类客户端配置方式略有不同但核心三件套不变Base URL、API Key、Model ID。这三样凑齐模型通道就通了。MCP 的 Server 和 Client 都跑在本地模型调用走 TaoToken整条链路就完整了。有一点提醒MCP Server 本身不调模型它只负责执行工具。模型调用发生在 Agent 主循环里。所以 TaoToken 的 Key 是配在 Agent 侧的不是配在 MCP Server 里的。这个边界要分清不然后面调试会绕晕。3. 可复制的 MCP Server 配置与工具注册现在进入正题。MCP 的通信层是 JSON-RPC 2.0标准方法有四个initialize握手、tools/list列工具、tools/call调工具、ping心跳。我先把协议层封装出来。import json, uuid from typing import Callable, Optional class JSONRPC: VERSION 2.0 staticmethod def request(method: str, params: dict None, id: str None) - dict: return { jsonrpc: JSONRPC.VERSION, method: method, params: params or {}, id: id or str(uuid.uuid4())[:8], } staticmethod def response(id: str, result: dict) - dict: return {jsonrpc: JSONRPC.VERSION, result: result, id: id} staticmethod def error(id: str, code: int, message: str, data: dict None) - dict: return { jsonrpc: JSONRPC.VERSION, error: {code: code, message: message, data: data or {}}, id: id, } class MCPMethods: INITIALIZE initialize TOOLS_LIST tools/list TOOLS_CALL tools/call PING ping这段是协议骨架没什么花活。id用 uuid 前 8 位保证请求响应能对上。生产环境如果走 HTTP建议用完整 uuid 避免碰撞。接下来是 Server 本体。核心是一个tools字典key 是工具名value 存定义和 handler。register_tool负责往字典里塞handle_request负责路由。class MCPServer: def __init__(self, name: str, version: str 1.0.0): self.name name self.version version self.tools: dict[str, dict] {} self._handlers { MCPMethods.INITIALIZE: self._handle_initialize, MCPMethods.TOOLS_LIST: self._handle_tools_list, MCPMethods.TOOLS_CALL: self._handle_tools_call, MCPMethods.PING: self._handle_ping, } def register_tool(self, name: str, description: str, parameters: list[dict], handler: Callable): self.tools[name] { definition: { name: name, description: description, inputSchema: { type: object, properties: { p[name]: { type: p.get(type, string), description: p.get(description, ), } for p in parameters }, required: [p[name] for p in parameters if p.get(required, True)], }, }, handler: handler, } def handle_request(self, raw: dict) - dict: method raw.get(method, ) req_id raw.get(id, ) handler self._handlers.get(method) if not handler: return JSONRPC.error(req_id, -32601, fMethod not found: {method}) try: return JSONRPC.response(req_id, handler(raw.get(params, {}))) except Exception as e: return JSONRPC.error(req_id, -32000, str(e)) def _handle_initialize(self, params: dict) - dict: return { protocolVersion: 2024-11-05, serverInfo: {name: self.name, version: self.version}, capabilities: {tools: {}}, } def _handle_tools_list(self, params: dict) - dict: return {tools: [t[definition] for t in self.tools.values()]} def _handle_tools_call(self, params: dict) - dict: tool self.tools.get(params.get(name, )) if not tool: raise ValueError(fTool not found: {params.get(name)}) result tool[handler](**params.get(arguments, {})) return {content: [{type: text, text: str(result)}], isError: False} def _handle_ping(self, params: dict) - dict: return {status: ok}注意inputSchema的结构这是给模型看的工具描述格式必须和 OpenAI function calling 的 parameters 对齐否则模型理解不了参数。required字段从参数列表里自动推导标了required: False的就不进必填项。现在注册两个 Server 做演示。一个是搜索服务一个是通知服务。search_server MCPServer(search-service) search_server.register_tool( nameweb_search, description搜索互联网获取信息用于查询最新新闻、事实、人物。, parameters[ {name: query, type: string, description: 搜索关键词, required: True}, {name: max_results, type: number, description: 结果数默认5, required: False}, ], handlerlambda query, max_results5: f搜索[{query}]返回{max_results}条结果, ) notify_server MCPServer(notification-service) notify_server.register_tool( namesend_email, description发送邮件用于通知、报告分发。, parameters[ {name: to, type: string, description: 收件人邮箱, required: True}, {name: subject, type: string, description: 邮件主题, required: True}, {name: body, type: string, description: 邮件正文, required: True}, ], handlerlambda to, subject, body: f邮件已发送至{to}主题{subject}, )到这里 Server 侧就完成了。两个 Server 各自独立互不感知这就是「工具服务化」——每个 Server 是一个沙箱一个挂了不影响另一个。4. MCP Client 自动发现与热插拔验证Client 是 Agent 侧的东西职责是连 Server、拉工具列表、把远程工具包装成本地可调用的函数。关键在_register_to_agent这一步它把发现的工具动态注入 Agent 的工具注册表。class MCPClient: def __init__(self, agent): self.agent agent self.servers: dict[str, MCPServer] {} self.discovered_tools: dict[str, dict] {} def connect_server(self, server: MCPServer): self.servers[server.name] server self._send_and_receive(server, MCPMethods.INITIALIZE, {}) resp self._send_and_receive(server, MCPMethods.TOOLS_LIST, {}) for tool_def in resp.get(result, {}).get(tools, []): self.discovered_tools[tool_def[name]] { server: server.name, definition: tool_def, } print(f[DISCOVER] {tool_def[name]} from {server.name}) self._register_to_agent() def _send_and_receive(self, server: MCPServer, method: str, params: dict) - dict: return server.handle_request(JSONRPC.request(method, params)) def call_tool(self, tool_name: str, arguments: dict) - str: info self.discovered_tools.get(tool_name) if not info: return fError: Tool {tool_name} not found server self.servers[info[server]] resp self._send_and_receive(server, MCPMethods.TOOLS_CALL, { name: tool_name, arguments: arguments, }) if error in resp: return fMCP Error: {resp[error][message]} content resp.get(result, {}).get(content, []) return content[0].get(text, str(content)) if content else str(resp) def _register_to_agent(self): for name, info in self.discovered_tools.items(): schema info[definition].get(inputSchema, {}) props schema.get(properties, {}) def make_wrapper(tool_name): def wrapper(**kwargs): return self.call_tool(tool_name, kwargs) return wrapper self.agent.register_tool( namename, descriptioninfo[definition][description], parameters[ {name: pn, type: pi.get(type, string), description: pi.get(description, ), required: pn in schema.get(required, [])} for pn, pi in props.items() ], handlermake_wrapper(name), )make_wrapper这里有个闭包陷阱要注意如果直接写lambda **kw: self.call_tool(name, kw)循环里name会被最后一次迭代覆盖。用工厂函数把tool_name固定住才对。这个坑我在第一版实现时踩过所有工具都调到了最后一个。现在跑热插拔验证。核心思路是先连一个 Server看工具列表再连第二个 Server看工具列表是否自动增长最后调一个远程工具确认能通。class DummyAgent: def __init__(self): self.tools {} def register_tool(self, name, description, parameters, handler): self.tools[name] {description: description, handler: handler} agent DummyAgent() mcp MCPClient(agent) mcp.connect_server(search_server) print(连接 search 后工具:, list(agent.tools.keys())) mcp.connect_server(notify_server) print(连接 notify 后工具:, list(agent.tools.keys())) print(调用 web_search:, agent.tools[web_search][handler](queryMCP协议, max_results3)) print(调用 send_email:, agent.tools[send_email][handler]( tobossexample.com, subject日报, body今日完成MCP实现))预期输出[DISCOVER] web_search from search-service 连接 search 后工具: [web_search] [DISCOVER] send_email from notification-service 连接 notify 后工具: [web_search, send_email] 调用 web_search: 搜索[MCP协议]返回3条结果 调用 send_email: 邮件已发送至bossexample.com主题日报看到工具列表从 1 个变成 2 个且调用都返回正常结果热插拔就验证通过了。整个过程 Agent 的DummyAgent类没有任何改动工具是运行时注入的。如果你要接真实模型把agent.tools里的工具定义转成 OpenAI tools 格式连同用户消息一起发给 TaoToken 的/v1/chat/completions模型返回tool_calls后你根据function.name找到对应的 handler 执行把结果作为tool角色消息回传。这就是完整的 Agent 工具调用循环。5. 常见报错排查401、local proxy failed、reading choicesMCP 落地时踩的坑集中在两类模型通道问题和协议层问题。我按真实报错逐个说。401 Unauthorized。这个基本是 Key 的问题。先确认环境变量有没有生效echo $TAOTOKEN_API_KEY如果为空说明 export 没在当前 shell 生效。再确认 Key 有没有多余空格复制时容易带上换行。还有一种情况是 Base URL 写成了https://taotoken.net/api/v1多加了/v1正确写法是https://taotoken.net/apiSDK 会自己拼/v1/chat/completions。如果用的是 Claude Code 或 Cline检查配置文件里的ANTHROPIC_BASE_URL或OPENAI_BASE_URL是否指向正确地址。local proxy failed / connection refused。这个报错通常出现在客户端配置了本地代理端口但代理没起来。MCP 场景下如果你用 stdio 方式启动 ServerClient 是通过子进程管道通信的不走网络不会出现这个错。出现这个错说明你在 HTTP 模式下配了http://127.0.0.1:xxxx但服务没监听。检查 Server 是否真的启动了端口是否被占用。另外有些客户端默认会读系统代理环境变量如果HTTP_PROXY指向一个不存在的地址也会报这个。清掉代理环境变量再试。reading choices of undefined。这是 OpenAI SDK 的经典报错意思是响应体里没有choices字段。原因通常是一、模型 ID 写错了服务端返回了错误 JSON 而不是标准响应二、Base URL 配错请求打到了非兼容端点三、请求体格式不对比如messages为空。排查方法是在代码里打印原始响应import openai client openai.OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) try: resp client.chat.completions.create( modelos.environ[TAOTOKEN_MODEL], messages[{role: user, content: ping}], ) print(resp.choices[0].message.content) except Exception as e: print(原始错误:, e)如果这里就报错说明模型通道没通跟 MCP 无关。先把这个最小请求跑通再往上叠 MCP。OAuth / token expired。有些客户端比如 Claude Code走的是 OAuth 流程配置里如果混用了 OAuth token 和 API Key 会冲突。用 TaoToken 的 Key 时确保配置项是 API Key 字段而不是 OAuth 字段。CC Switch 这类工具切换配置时注意 Base URL、Key、Model ID 三件套要一起换只换其中一个会导致鉴权失败。工具调用返回 tool not found。这是 MCP 协议层的错不是模型通道的错。检查discovered_tools里有没有这个工具名大小写是否一致。MCP 的工具名是大小写敏感的web_search和Web_Search是两个工具。另外如果 Server 重启了但 Client 没重连discovered_tools里还是旧列表调新工具就会找不到。生产环境建议加个心跳检测Server 挂了自动重连并刷新工具列表。模型不返回 tool_calls。工具描述传过去了但模型就是不用工具直接编答案。这通常是工具描述写得太模糊。description要写清楚「什么时候用这个工具」而不是「这个工具是什么」。比如web_search的描述写成「搜索互联网获取信息用于查询最新新闻、事实、人物」比「搜索工具」强十倍。参数描述同理query要写「搜索关键词」而不是「输入」。6. 把 MCP 接进你的 Agent 主循环前面验证的是工具发现和调用现在说怎么接进真实的 Agent 循环。核心是把discovered_tools转成模型能理解的 tools 格式然后在模型返回tool_calls时分发执行。def build_tools_payload(mcp_client: MCPClient) - list[dict]: payload [] for name, info in mcp_client.discovered_tools.items(): schema info[definition][inputSchema] payload.append({ type: function, function: { name: name, description: info[definition][description], parameters: schema, }, }) return payload def run_agent_turn(mcp_client, messages, client, model): tools build_tools_payload(mcp_client) resp client.chat.completions.create( modelmodel, messagesmessages, toolstools, tool_choiceauto, ) msg resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) for call in msg.tool_calls: args json.loads(call.function.arguments) result mcp_client.call_tool(call.function.name, args) messages.append({ role: tool, tool_call_id: call.id, content: result, }) return run_agent_turn(mcp_client, messages, client, model)这段是 Agent 的核心循环模型决定调哪个工具Client 执行结果回传模型继续推理直到不再调工具为止。build_tools_payload每次从discovered_tools现算所以新连的 Server 工具会自动出现在下一轮请求里这就是热插拔在 Agent 层面的体现。接 TaoToken 的完整初始化import os, openai client openai.OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) model os.environ[TAOTOKEN_MODEL] mcp MCPClient(agent) mcp.connect_server(search_server) mcp.connect_server(notify_server) messages [{role: user, content: 搜索MCP协议最新进展把摘要发到 devexample.com}] print(run_agent_turn(mcp, messages, client, model))跑通之后你会看到模型先调web_search拿到结果后再调send_email两步工具调用自动串联。整个过程 Agent 代码没变工具是运行时注入的。想验证热插拔在connect_server之后再注册一个新工具到 Server然后重新connect_server一次discovered_tools会刷新下一轮请求模型就能看到新工具。生产环境可以把这个刷新做成定时任务或者 Server 推送通知。如果你要长期跑 Agent 任务建议用 Coding Plan 这类按量方案工具调用频繁时成本可控。模型对话入口可以用来快速验证工具描述写得对不对接入文档里有完整的参数说明。工具生态这块MCP 的价值不在协议本身而在于它让工具变成了可分发、可组合的资产——你写的 Server 别人能直接用别人写的你也能挂载。这才是「应用商店」的真正含义。 手搓生产级 AI Agent 系统（5）MCP 协议从零实现：用 TaoToken 统一 Key 打通工具热插拔与标准化生态
1. 为什么你的 Agent 每加一个工具就要改代码先说一个我踩过的坑。去年做客服 Agent 的时候工具从 3 个涨到 17 个每加一个工具就要动三处代码写 handler、改 TOOLS 字典、调 prompt 里的工具描述。最要命的是上线之后发现某个工具参数写错了得重新发版。这种耦合度在生产环境里就是灾难。MCPModel Context Protocol要解决的就是这件事。它把工具从 Agent 代码里彻底剥离出来变成一个独立的 Server 进程Agent 启动时通过 JSON-RPC 2.0 协议去问 Server「你有哪些工具」拿到工具列表后自动注册到自己的工具系统里。整个过程 Agent 主逻辑一行不改。你可以把 MCP 理解成 Agent 的「应用商店」每个 MCP Server 就是一个 App里面装着若干工具Agent 是手机开机时扫描已安装的 App把里面的功能挂到桌面上。想加新功能装个新 App 就行不用刷机。这套协议适合谁三类人一是自建 Agent 框架、工具数量超过 5 个的开发者二是想让多个 Agent 共享同一套工具的后端团队三是想把内部系统数据库、工单、监控包装成标准工具对外暴露的平台方。如果你还在用硬编码的 function calling 列表工具数一超过 10 个就会开始难受。这一篇我会从零实现一个可运行的 MCP Server Client包含 JSON-RPC 通信层、工具注册、自动发现、热插拔验证脚本最后用 TaoToken 统一 Key 打通多工具调用的连通性验证。代码可以直接复制跑。2. TaoToken 前置准备统一 Key 与 API 通道在写 MCP 之前得先把模型调用这条链路理顺。MCP 解决的是「工具有哪些、怎么调」但工具调用的决策还是模型做的——模型得能看懂工具描述、能返回结构化的 tool_call。所以你需要一个稳定的模型 API 通道。我用 TaoToken 做统一入口原因是它把多个模型的 Key 收敛成一个Agent 侧只配一个 Base URL 和一个 Key切换模型不用改代码。对 MCP 场景特别友好工具描述会随 Server 变化模型得频繁重新理解工具集用统一通道省去多 Key 管理的麻烦。先拿 Key。打开 https://taotoken.net/api-keys 登录后创建一个 API Key复制出来形如sk-xxxxxxxx。这个 Key 后面会写进环境变量不要硬编码到代码里。然后确认你的调用地址。TaoToken 的 API 入口是https://taotoken.net/api兼容 OpenAI 的/v1/chat/completions格式。也就是说你原来用 openai SDK 的代码只改base_url和api_key两个字段就能跑。模型 ID 这块要注意MCP 场景下模型必须支持 function calling / tool use否则工具描述传过去它也不认。选模型时优先挑带 tool 能力的具体可用列表在 https://taotoken.net/models 查。我实测下来工具数量在 20 个以内时主流模型的工具选择准确率都够用。环境变量这样配export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL你的模型ID如果你用 Claude Code 或者 Cline 这类客户端配置方式略有不同但核心三件套不变Base URL、API Key、Model ID。这三样凑齐模型通道就通了。MCP 的 Server 和 Client 都跑在本地模型调用走 TaoToken整条链路就完整了。有一点提醒MCP Server 本身不调模型它只负责执行工具。模型调用发生在 Agent 主循环里。所以 TaoToken 的 Key 是配在 Agent 侧的不是配在 MCP Server 里的。这个边界要分清不然后面调试会绕晕。3. 可复制的 MCP Server 配置与工具注册现在进入正题。MCP 的通信层是 JSON-RPC 2.0标准方法有四个initialize握手、tools/list列工具、tools/call调工具、ping心跳。我先把协议层封装出来。import json, uuid from typing import Callable, Optional class JSONRPC: VERSION 2.0 staticmethod def request(method: str, params: dict None, id: str None) - dict: return { jsonrpc: JSONRPC.VERSION, method: method, params: params or {}, id: id or str(uuid.uuid4())[:8], } staticmethod def response(id: str, result: dict) - dict: return {jsonrpc: JSONRPC.VERSION, result: result, id: id} staticmethod def error(id: str, code: int, message: str, data: dict None) - dict: return { jsonrpc: JSONRPC.VERSION, error: {code: code, message: message, data: data or {}}, id: id, } class MCPMethods: INITIALIZE initialize TOOLS_LIST tools/list TOOLS_CALL tools/call PING ping这段是协议骨架没什么花活。id用 uuid 前 8 位保证请求响应能对上。生产环境如果走 HTTP建议用完整 uuid 避免碰撞。接下来是 Server 本体。核心是一个tools字典key 是工具名value 存定义和 handler。register_tool负责往字典里塞handle_request负责路由。class MCPServer: def __init__(self, name: str, version: str 1.0.0): self.name name self.version version self.tools: dict[str, dict] {} self._handlers { MCPMethods.INITIALIZE: self._handle_initialize, MCPMethods.TOOLS_LIST: self._handle_tools_list, MCPMethods.TOOLS_CALL: self._handle_tools_call, MCPMethods.PING: self._handle_ping, } def register_tool(self, name: str, description: str, parameters: list[dict], handler: Callable): self.tools[name] { definition: { name: name, description: description, inputSchema: { type: object, properties: { p[name]: { type: p.get(type, string), description: p.get(description, ), } for p in parameters }, required: [p[name] for p in parameters if p.get(required, True)], }, }, handler: handler, } def handle_request(self, raw: dict) - dict: method raw.get(method, ) req_id raw.get(id, ) handler self._handlers.get(method) if not handler: return JSONRPC.error(req_id, -32601, fMethod not found: {method}) try: return JSONRPC.response(req_id, handler(raw.get(params, {}))) except Exception as e: return JSONRPC.error(req_id, -32000, str(e)) def _handle_initialize(self, params: dict) - dict: return { protocolVersion: 2024-11-05, serverInfo: {name: self.name, version: self.version}, capabilities: {tools: {}}, } def _handle_tools_list(self, params: dict) - dict: return {tools: [t[definition] for t in self.tools.values()]} def _handle_tools_call(self, params: dict) - dict: tool self.tools.get(params.get(name, )) if not tool: raise ValueError(fTool not found: {params.get(name)}) result tool[handler](**params.get(arguments, {})) return {content: [{type: text, text: str(result)}], isError: False} def _handle_ping(self, params: dict) - dict: return {status: ok}注意inputSchema的结构这是给模型看的工具描述格式必须和 OpenAI function calling 的 parameters 对齐否则模型理解不了参数。required字段从参数列表里自动推导标了required: False的就不进必填项。现在注册两个 Server 做演示。一个是搜索服务一个是通知服务。search_server MCPServer(search-service) search_server.register_tool( nameweb_search, description搜索互联网获取信息用于查询最新新闻、事实、人物。, parameters[ {name: query, type: string, description: 搜索关键词, required: True}, {name: max_results, type: number, description: 结果数默认5, required: False}, ], handlerlambda query, max_results5: f搜索[{query}]返回{max_results}条结果, ) notify_server MCPServer(notification-service) notify_server.register_tool( namesend_email, description发送邮件用于通知、报告分发。, parameters[ {name: to, type: string, description: 收件人邮箱, required: True}, {name: subject, type: string, description: 邮件主题, required: True}, {name: body, type: string, description: 邮件正文, required: True}, ], handlerlambda to, subject, body: f邮件已发送至{to}主题{subject}, )到这里 Server 侧就完成了。两个 Server 各自独立互不感知这就是「工具服务化」——每个 Server 是一个沙箱一个挂了不影响另一个。4. MCP Client 自动发现与热插拔验证Client 是 Agent 侧的东西职责是连 Server、拉工具列表、把远程工具包装成本地可调用的函数。关键在_register_to_agent这一步它把发现的工具动态注入 Agent 的工具注册表。class MCPClient: def __init__(self, agent): self.agent agent self.servers: dict[str, MCPServer] {} self.discovered_tools: dict[str, dict] {} def connect_server(self, server: MCPServer): self.servers[server.name] server self._send_and_receive(server, MCPMethods.INITIALIZE, {}) resp self._send_and_receive(server, MCPMethods.TOOLS_LIST, {}) for tool_def in resp.get(result, {}).get(tools, []): self.discovered_tools[tool_def[name]] { server: server.name, definition: tool_def, } print(f[DISCOVER] {tool_def[name]} from {server.name}) self._register_to_agent() def _send_and_receive(self, server: MCPServer, method: str, params: dict) - dict: return server.handle_request(JSONRPC.request(method, params)) def call_tool(self, tool_name: str, arguments: dict) - str: info self.discovered_tools.get(tool_name) if not info: return fError: Tool {tool_name} not found server self.servers[info[server]] resp self._send_and_receive(server, MCPMethods.TOOLS_CALL, { name: tool_name, arguments: arguments, }) if error in resp: return fMCP Error: {resp[error][message]} content resp.get(result, {}).get(content, []) return content[0].get(text, str(content)) if content else str(resp) def _register_to_agent(self): for name, info in self.discovered_tools.items(): schema info[definition].get(inputSchema, {}) props schema.get(properties, {}) def make_wrapper(tool_name): def wrapper(**kwargs): return self.call_tool(tool_name, kwargs) return wrapper self.agent.register_tool( namename, descriptioninfo[definition][description], parameters[ {name: pn, type: pi.get(type, string), description: pi.get(description, ), required: pn in schema.get(required, [])} for pn, pi in props.items() ], handlermake_wrapper(name), )make_wrapper这里有个闭包陷阱要注意如果直接写lambda **kw: self.call_tool(name, kw)循环里name会被最后一次迭代覆盖。用工厂函数把tool_name固定住才对。这个坑我在第一版实现时踩过所有工具都调到了最后一个。现在跑热插拔验证。核心思路是先连一个 Server看工具列表再连第二个 Server看工具列表是否自动增长最后调一个远程工具确认能通。class DummyAgent: def __init__(self): self.tools {} def register_tool(self, name, description, parameters, handler): self.tools[name] {description: description, handler: handler} agent DummyAgent() mcp MCPClient(agent) mcp.connect_server(search_server) print(连接 search 后工具:, list(agent.tools.keys())) mcp.connect_server(notify_server) print(连接 notify 后工具:, list(agent.tools.keys())) print(调用 web_search:, agent.tools[web_search][handler](queryMCP协议, max_results3)) print(调用 send_email:, agent.tools[send_email][handler]( tobossexample.com, subject日报, body今日完成MCP实现))预期输出[DISCOVER] web_search from search-service 连接 search 后工具: [web_search] [DISCOVER] send_email from notification-service 连接 notify 后工具: [web_search, send_email] 调用 web_search: 搜索[MCP协议]返回3条结果 调用 send_email: 邮件已发送至bossexample.com主题日报看到工具列表从 1 个变成 2 个且调用都返回正常结果热插拔就验证通过了。整个过程 Agent 的DummyAgent类没有任何改动工具是运行时注入的。如果你要接真实模型把agent.tools里的工具定义转成 OpenAI tools 格式连同用户消息一起发给 TaoToken 的/v1/chat/completions模型返回tool_calls后你根据function.name找到对应的 handler 执行把结果作为tool角色消息回传。这就是完整的 Agent 工具调用循环。5. 常见报错排查401、local proxy failed、reading choicesMCP 落地时踩的坑集中在两类模型通道问题和协议层问题。我按真实报错逐个说。401 Unauthorized。这个基本是 Key 的问题。先确认环境变量有没有生效echo $TAOTOKEN_API_KEY如果为空说明 export 没在当前 shell 生效。再确认 Key 有没有多余空格复制时容易带上换行。还有一种情况是 Base URL 写成了https://taotoken.net/api/v1多加了/v1正确写法是https://taotoken.net/apiSDK 会自己拼/v1/chat/completions。如果用的是 Claude Code 或 Cline检查配置文件里的ANTHROPIC_BASE_URL或OPENAI_BASE_URL是否指向正确地址。local proxy failed / connection refused。这个报错通常出现在客户端配置了本地代理端口但代理没起来。MCP 场景下如果你用 stdio 方式启动 ServerClient 是通过子进程管道通信的不走网络不会出现这个错。出现这个错说明你在 HTTP 模式下配了http://127.0.0.1:xxxx但服务没监听。检查 Server 是否真的启动了端口是否被占用。另外有些客户端默认会读系统代理环境变量如果HTTP_PROXY指向一个不存在的地址也会报这个。清掉代理环境变量再试。reading choices of undefined。这是 OpenAI SDK 的经典报错意思是响应体里没有choices字段。原因通常是一、模型 ID 写错了服务端返回了错误 JSON 而不是标准响应二、Base URL 配错请求打到了非兼容端点三、请求体格式不对比如messages为空。排查方法是在代码里打印原始响应import openai client openai.OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) try: resp client.chat.completions.create( modelos.environ[TAOTOKEN_MODEL], messages[{role: user, content: ping}], ) print(resp.choices[0].message.content) except Exception as e: print(原始错误:, e)如果这里就报错说明模型通道没通跟 MCP 无关。先把这个最小请求跑通再往上叠 MCP。OAuth / token expired。有些客户端比如 Claude Code走的是 OAuth 流程配置里如果混用了 OAuth token 和 API Key 会冲突。用 TaoToken 的 Key 时确保配置项是 API Key 字段而不是 OAuth 字段。CC Switch 这类工具切换配置时注意 Base URL、Key、Model ID 三件套要一起换只换其中一个会导致鉴权失败。工具调用返回 tool not found。这是 MCP 协议层的错不是模型通道的错。检查discovered_tools里有没有这个工具名大小写是否一致。MCP 的工具名是大小写敏感的web_search和Web_Search是两个工具。另外如果 Server 重启了但 Client 没重连discovered_tools里还是旧列表调新工具就会找不到。生产环境建议加个心跳检测Server 挂了自动重连并刷新工具列表。模型不返回 tool_calls。工具描述传过去了但模型就是不用工具直接编答案。这通常是工具描述写得太模糊。description要写清楚「什么时候用这个工具」而不是「这个工具是什么」。比如web_search的描述写成「搜索互联网获取信息用于查询最新新闻、事实、人物」比「搜索工具」强十倍。参数描述同理query要写「搜索关键词」而不是「输入」。6. 把 MCP 接进你的 Agent 主循环前面验证的是工具发现和调用现在说怎么接进真实的 Agent 循环。核心是把discovered_tools转成模型能理解的 tools 格式然后在模型返回tool_calls时分发执行。def build_tools_payload(mcp_client: MCPClient) - list[dict]: payload [] for name, info in mcp_client.discovered_tools.items(): schema info[definition][inputSchema] payload.append({ type: function, function: { name: name, description: info[definition][description], parameters: schema, }, }) return payload def run_agent_turn(mcp_client, messages, client, model): tools build_tools_payload(mcp_client) resp client.chat.completions.create( modelmodel, messagesmessages, toolstools, tool_choiceauto, ) msg resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) for call in msg.tool_calls: args json.loads(call.function.arguments) result mcp_client.call_tool(call.function.name, args) messages.append({ role: tool, tool_call_id: call.id, content: result, }) return run_agent_turn(mcp_client, messages, client, model)这段是 Agent 的核心循环模型决定调哪个工具Client 执行结果回传模型继续推理直到不再调工具为止。build_tools_payload每次从discovered_tools现算所以新连的 Server 工具会自动出现在下一轮请求里这就是热插拔在 Agent 层面的体现。接 TaoToken 的完整初始化import os, openai client openai.OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) model os.environ[TAOTOKEN_MODEL] mcp MCPClient(agent) mcp.connect_server(search_server) mcp.connect_server(notify_server) messages [{role: user, content: 搜索MCP协议最新进展把摘要发到 devexample.com}] print(run_agent_turn(mcp, messages, client, model))跑通之后你会看到模型先调web_search拿到结果后再调send_email两步工具调用自动串联。整个过程 Agent 代码没变工具是运行时注入的。想验证热插拔在connect_server之后再注册一个新工具到 Server然后重新connect_server一次discovered_tools会刷新下一轮请求模型就能看到新工具。生产环境可以把这个刷新做成定时任务或者 Server 推送通知。如果你要长期跑 Agent 任务建议用 Coding Plan 这类按量方案工具调用频繁时成本可控。模型对话入口可以用来快速验证工具描述写得对不对接入文档里有完整的参数说明。工具生态这块MCP 的价值不在协议本身而在于它让工具变成了可分发、可组合的资产——你写的 Server 别人能直接用别人写的你也能挂载。这才是「应用商店」的真正含义。