1. 为什么 Agent 工具调用总在“最后一公里”卡住如果你正在写 Python Agent大概率遇到过这种局面模型能聊天、能写代码但一到“帮我查一下今天上海天气并画个图”就歇菜。原因不在模型本身而在于工具调用链路没有真正打通。MCPModel Context Protocol模型上下文协议就是来解决这个问题的——它把“模型如何发现工具、如何传参、如何拿回结果”标准化了让 Agent 不再为每个工具写一套胶水代码。但标准归标准落地时还有两个现实问题一是模型接入层五花八门OpenAI 格式、Anthropic 格式、各家私有格式混在一起二是 MCP Server 的传输类型不止一种stdio、sse、streamable-http 各有各的写法。这篇就聚焦一条完整链路用 Python 写一个 MCP Client把外部工具注册进来再通过 TaoToken 统一 API 通道调用模型让模型自己决定调哪个工具、传什么参数最后跑通一次端到端验证。适合谁看写过一点 Python、想给 Agent 加工具但被配置劝退的人或者已经在用 MCP 但模型接入层换来换去、想统一收口的人。下面所有代码和配置都可以直接复制改掉 Key 就能跑。2. TaoToken 前置统一 Key 与 API 通道怎么准备TaoToken 在这里扮演的角色是“模型接入层”。你的 Agent 不需要关心背后是哪个模型厂商只需要一个统一的 API 地址和一把 Key就能完成对话和工具调用。这对 MCP 场景特别友好因为工具调用的请求格式最终要转成模型能理解的 function calling 格式统一通道能省掉大量适配工作。第一步去官网注册并拿到 API Key。地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台创建 Key。控制台入口https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。Key 的管理页面在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 建议给 Agent 单独建一把方便后续轮换。第二步确认 API 基地址。TaoToken 的 API 入口是 https://taotoken.net/api 注意这个地址不带 UTM 参数代码里直接写这个就行。它兼容 OpenAI 的接口风格所以后面我们用 openai 这个 Python 包就能直接调不需要额外 SDK。第三步把 Key 放进环境变量别硬编码。Linux/macOS 下export TAOTOKEN_API_KEYsk-你的KeyWindows PowerShell$env:TAOTOKEN_API_KEYsk-你的Key注意Key 一旦泄露要立刻在控制台吊销重建。Agent 项目里建议用 .env 文件配合 python-dotenv 加载别提交到 Git。到这里前置就完成了。你手里应该有一个可用的 Key、一个 API 基地址以及一个能跑 Python 的环境。接下来进入配置骨架。3. 可复制配置settings.json 与 config.toml 骨架MCP 生态里有两套常见的配置文件风格一套是 Claude 系的 settings.json用来声明 MCP Server另一套是偏工程化的 config.toml用来放模型接入参数。我们把两者都搭起来各管各的。3.1 settings.json声明 MCP Server这个文件的作用是告诉 Client“有哪些工具可用”。格式沿用社区通用的 mcpServers 结构stdio 类型写 command 和 args远程类型写 url、type 和 headers。{ mcpServers: { mcp-server-chart: { command: npx, args: [-y, antv/mcp-server-chart] }, amap: { url: https://mcp.amap.com/sse?key你的高德Key, type: sse }, metaso: { type: http, url: https://metaso.cn/api/mcp, headers: { Authorization: Bearer 你的秘塔Key } } } }三个 Server 分别覆盖三种传输类型mcp-server-chart 是 stdio本地通过 npx 拉起amap 是 sse走远程事件流metaso 是 streamable-http走标准 HTTP。这样你的 Client 一旦支持这三种市面上绝大多数 MCP Server 都能接。3.2 config.toml模型接入参数模型接入层单独放一个 toml把 TaoToken 的地址、Key、默认模型写进去代码里读配置而不是散落各处。[llm] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model gpt-4o-mini timeout 60 [agent] max_tool_rounds 5base_url 指向 TaoToken 的 API 入口api_key_env 写环境变量名而不是 Key 本身这样配置可以进版本库。model 先填一个支持 function calling 的模型后面验证时如果发现工具调用不触发优先检查模型是否支持。3.3 依赖安装pip install mcp openai tomli python-dotenv nest_asynciotomli 用于读 tomlPython 3.11 以下需要nest_asyncio 是为了在 Jupyter 里跑异步代码不报事件循环嵌套的错。如果你只在普通脚本里跑nest_asyncio 可以不装。4. 端到端验证注册工具并让模型真正调一次配置齐了现在写核心代码。整体分四步读配置、建 MCP Client、把 MCP 工具转成 OpenAI function 格式、让模型决策并执行。4.1 MCP Client 骨架Client 要支持三种传输核心是把不同 client 的上下文统一成 read_stream/write_stream 两个流。import asyncio from contextlib import asynccontextmanager from mcp import ClientSession from mcp.client.stdio import stdio_client, StdioServerParameters from mcp.client.sse import sse_client from mcp.client.streamable_http import streamablehttp_client class MCPClient: def __init__(self, servers: dict, timeout: int 60): self.servers servers[mcpServers] self.timeout timeout asynccontextmanager async def _transport(self, name): cfg self.servers[name] if url in cfg: if cfg.get(type) sse: client sse_client(urlcfg[url], headerscfg.get(headers, {}), timeoutfloat(self.timeout)) else: client streamablehttp_client(urlcfg[url], headerscfg.get(headers, {}), timeoutfloat(self.timeout)) async with client as streams: read_stream, write_stream streams[0], streams[1] yield read_stream, write_stream else: params StdioServerParameters(commandcfg[command], argscfg[args], envcfg.get(env)) async with stdio_client(params) as (read_stream, write_stream): yield read_stream, write_stream async def alist_tools(self, name): async with self._transport(name) as (r, w): async with ClientSession(r, w) as session: await session.initialize() resp await session.list_tools() return resp.tools async def acall_tool(self, name, tool_name, argsNone): async with self._transport(name) as (r, w): async with ClientSession(r, w) as session: await session.initialize() return await session.call_tool(tool_name, args or {})这段和官方样例的差别在于把传输选择收进一个_transport上下文管理器list 和 call 复用同一套逻辑避免重复代码。streamable-http 返回的是三元组这里只取前两个流因为工具调用不需要 session id。4.2 工具格式转换MCP 的工具定义和 OpenAI 的 function calling 格式不一样需要转一层。关键是把 inputSchema 里的 properties 和 required 搬过去。def convert_tools(tools): result [] for tool in tools: schema tool.inputSchema params { type: object, properties: {}, required: schema.get(required, []), } for pname, pdef in schema.get(properties, {}).items(): ptype pdef.get(type) if not ptype and pdef.get(anyOf): ptype pdef[anyOf][0].get(type, string) params[properties][pname] {type: ptype or string, **pdef} result.append({ type: function, function: { name: tool.name, description: tool.description, parameters: params, }, }) return resultanyOf 的处理是个坑有些 MCP Server 的参数类型写成 anyOf 数组直接取 type 会拿到 None这里兜底成 string否则模型看到的参数类型是空的容易传错。4.3 让模型决策并执行工具现在把 MCP 工具喂给模型走一轮完整的“模型要工具 → 执行 → 回填结果 → 模型总结”。import os, json, tomli from openai import OpenAI with open(config.toml, rb) as f: cfg tomli.load(f) client OpenAI( base_urlcfg[llm][base_url], api_keyos.environ[cfg[llm][api_key_env]], ) async def run_agent(user_input, mcp_client, server_name): tools await mcp_client.alist_tools(server_name) openai_tools convert_tools(tools) messages [{role: user, content: user_input}] for _ in range(cfg[agent][max_tool_rounds]): resp client.chat.completions.create( modelcfg[llm][model], messagesmessages, toolsopenai_tools, ) msg resp.choices[0].message messages.append(msg) if not msg.tool_calls: return msg.content for call in msg.tool_calls: args json.loads(call.function.arguments) result await mcp_client.acall_tool(server_name, call.function.name, args) messages.append({ role: tool, tool_call_id: call.id, content: str(result.content), }) return 达到最大工具轮次未收敛max_tool_rounds 是防止模型陷入无限调工具的死循环设 5 轮足够大多数场景。每轮把模型的 tool_calls 拿出来逐个执行结果以 roletool 回填再让模型继续。4.4 跑一次验证async def main(): with open(settings.json, encodingutf-8) as f: servers json.load(f) mcp_client MCPClient(servers) answer await run_agent(搜索一下 MCP 协议是什么返回三条结果, mcp_client, metaso) print(answer) asyncio.run(main())预期结果模型先返回一个 tool_callname 是 metaso_web_searcharguments 里带 q 和 sizeClient 执行后把搜索结果回填模型再输出一段总结。如果你看到控制台先打印工具调用日志、再打印最终回答说明闭环通了。5. 本篇常见错排查工具调用跑不通八成是下面几个原因。按顺序排查效率最高。报错一ModuleNotFoundError: No module named mcp依赖没装或装错环境。确认pip show mcp有输出且和你运行脚本的 Python 是同一个。虚拟环境里装完记得激活。报错二RuntimeError: asyncio.run() cannot be called from a running event loop在 Jupyter 里直接asyncio.run会撞上事件循环嵌套。开头加两行import nest_asyncio nest_asyncio.apply()或者把asyncio.run(main())换成await main()。报错三模型不触发工具调用直接瞎编答案先确认模型支持 function calling。有些小模型或旧模型不支持 tools 参数传了也忽略。其次检查 convert_tools 后的 description 是否为空——描述为空模型不知道工具干嘛自然不调。最后看 tools 列表是否真的传进去了打印一下长度。报错四401 Unauthorized或invalid api keyTaoToken 的 Key 没读到。检查环境变量名和 config.toml 里的 api_key_env 是否一致echo $TAOTOKEN_API_KEY确认有值。base_url 必须是 https://taotoken.net/api 多写或少写路径都会 404。报错五stdio 类型 Server 启动失败npx 拉包需要 Node 环境。node -v确认已安装首次运行npx -y antv/mcp-server-chart会下载网络慢会超时可以先把包全局装好再改 command 指向本地。报错六工具执行返回内容为空部分 MCP Server 的结果在result.content里是列表结构直接 str 化可能丢信息。打印result原始对象看看结构必要时取result.content[0].text。6. 下一步把通道固定下来把工具接进来链路跑通之后真正影响长期体验的是两件事模型接入层是否稳定、工具是否好扩展。TaoToken 的统一 Key 和 API 通道解决的是前者——你换模型、加模型都不用改 Agent 代码只改 config.toml 里的 model 字段。MCP 解决的是后者——新工具只要往 settings.json 里加一段配置Client 自动发现不用动业务逻辑。如果你主要在做模型对话类的验证可以直接用模型对话页面快速试 prompt 和工具描述https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。如果你打算长期写 Agent、跑编码类任务建议看一下 Coding Plan把额度和通道固定下来https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到参数细节可以对照查。我自己的习惯是先把 settings.json 里的 Server 减到一个跑通单工具调用再逐个加。一次加三个 Server 然后报错排查成本会翻倍。另外工具描述尽量写清楚“什么时候用”模型选工具的准确率会明显提升这比换模型管用。