1. 从一次真实的报错说起为什么你的 MCP Client 跑不通如果你正在用 Python 写 MCP Client大概率会遇到这样的场景照着官方教程把client.py和weather.py都写好了uv run client.py weather.py一敲终端卡住不动或者直接抛出一句TypeError: NoneType object is not iterable。更让人抓狂的是你明明看到模型返回了tool_use但 Tool 就是没被执行循环走不下去。问题往往不在 MCP 协议本身而在于三套对象混在一起Python MCP SDK 的Client、模型 Provider 的 Messages API、以及 MCP Server 返回的 Content Block。这三层各自有独立的类型系统和 ID 命名空间一旦搞混Tool Calling Loop 就会在某个环节断掉。这篇内容聚焦用 Python 从零实现 MCP Client 的完整 Tool Calling Loop覆盖工具发现、调用、结果回填与多轮循环。我会给出可复制的config.toml/settings.json骨架以及用 TaoToken 统一 Key 接入的示例最后演示一次端到端验证启动 Client、触发工具调用、确认循环收敛。适合已经写过基础 Python 异步代码、想真正把 MCP Client 跑起来的开发者。2. TaoToken 前置统一 Key 与配置骨架在写 Client 之前先把模型接入这一层理顺。MCP Client 本身不关心你用哪家模型但 Tool Calling Loop 需要一个能返回结构化 Tool Call 的 Provider。TaoToken 的价值在于用一套 Key 和统一的 Base URL就能在 Anthropic、OpenAI 兼容格式之间切换省去为每个 Provider 单独维护 SDK 和鉴权逻辑的麻烦。2.1 获取 Key 与确认接入点先到控制台创建 API Key建议按项目分 Key方便后续排查和限额。拿到 Key 后模型对话入口可以用来快速验证 Key 是否可用接入文档则给出不同 SDK 的 Base URL 和参数写法。注意Key 只放在环境变量或本地配置文件里不要提交到 Git。.env、config.toml、settings.json都要进.gitignore。2.2 config.toml 骨架如果你用uv或poetry管理项目可以把 Provider 配置抽到config.toml让 Client 代码只读配置、不硬编码# config.toml [provider] name taotoken base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model claude-opus-5 max_tokens 1000 [mcp] server_script ./weather.py transport stdio connect_timeout 30 tool_call_timeout 60 [loop] max_rounds 8这里max_rounds是 Tool Calling Loop 的收敛上限防止模型反复请求 Tool 导致死循环。tool_call_timeout单独设置是因为 MCP Server 执行 Tool 可能比模型推理慢得多。2.3 settings.json 骨架如果你的项目更偏向 Node 生态或需要跨语言共享配置用settings.json等价表达{ provider: { name: taotoken, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, model: claude-opus-5, max_tokens: 1000 }, mcp: { server_script: ./weather.py, transport: stdio, connect_timeout: 30, tool_call_timeout: 60 }, loop: { max_rounds: 8 } }两份配置语义一致选一份即可。核心是把 Provider、MCP、Loop 三层参数分开后面排障时能快速定位是哪一层出的问题。2.4 环境变量与依赖# .env TAOTOKEN_API_KEY你的密钥uv add mcp anthropic python-dotenv tomlitomli用于读取config.tomlPython 3.11 以下。如果你用 3.11标准库tomllib就够了。3. 可复制配置把 Tool Calling Loop 拆成四层很多教程把整个 Loop 塞进一个process_query函数结果一出错就无从下手。我的做法是拆成四层配置加载层、MCP 连接层、Provider 适配层、循环控制层。这样每一层都能单独测试。3.1 配置加载层# config_loader.py import os import tomli from dataclasses import dataclass from dotenv import load_dotenv load_dotenv() dataclass class ProviderConfig: base_url: str api_key: str model: str max_tokens: int dataclass class MCPConfig: server_script: str transport: str connect_timeout: int tool_call_timeout: int dataclass class LoopConfig: max_rounds: int def load_config(path: str config.toml): with open(path, rb) as f: raw tomli.load(f) provider ProviderConfig( base_urlraw[provider][base_url], api_keyos.environ[raw[provider][api_key_env]], modelraw[provider][model], max_tokensraw[provider][max_tokens], ) mcp MCPConfig(**raw[mcp]) loop LoopConfig(**raw[loop]) return provider, mcp, loop这里api_key从环境变量读配置里只存变量名。这样即使config.toml被误提交Key 也不会泄露。3.2 MCP 连接层# mcp_layer.py import sys from contextlib import AsyncExitStack from mcp import Client from mcp.client.stdio import stdio_client from mcp.types import StdioServerParameters def build_server_params(script_path: str) - StdioServerParameters: if script_path.endswith(.py): command python elif script_path.endswith(.js): command node else: raise ValueError(Server script must be .py or .js) return StdioServerParameters(commandcommand, args[script_path]) class MCPConnection: def __init__(self): self.stack AsyncExitStack() self.client None async def connect(self, script_path: str): params build_server_params(script_path) transport stdio_client(params) self.client await self.stack.enter_async_context(Client(transport)) return self.client async def close(self): await self.stack.aclose()AsyncExitStack是关键它保证无论循环怎么退出子进程和 stdio 管道都会被清理。我试过不用它结果 Ctrl-C 之后weather.py变成僵尸进程端口和管道都没释放。3.3 Provider 适配层这一层负责把 MCP 的 Tool Definition 转成 Provider 格式再把 Provider 的 Tool Call 转成 MCP 的tools/call。# provider_adapter.py from anthropic import Anthropic from mcp.types import TextContent class ProviderAdapter: def __init__(self, cfg): self.client Anthropic( base_urlcfg.base_url, api_keycfg.api_key, ) self.model cfg.model self.max_tokens cfg.max_tokens def to_provider_tools(self, mcp_tools): return [ { name: t.name, description: t.description, input_schema: t.input_schema, } for t in mcp_tools ] def create(self, messages, tools): return self.client.messages.create( modelself.model, max_tokensself.max_tokens, messagesmessages, toolstools, ) staticmethod def extract_tool_result(mcp_result): text \n.join( block.text for block in mcp_result.content if isinstance(block, TextContent) ) return { type: tool_result, tool_use_id: None, # 由调用方填入 content: text, is_error: mcp_result.is_error, }注意tool_use_id留空因为适配层不知道当前对应哪个tool_use这个关联必须由循环控制层完成。3.4 循环控制层这是整个 Tool Calling Loop 的核心也是最多人写错的地方。# loop_layer.py import asyncio from provider_adapter import ProviderAdapter async def run_tool_loop(mcp_client, adapter, query, max_rounds8): messages [{role: user, content: query}] tool_list await mcp_client.list_tools() provider_tools adapter.to_provider_tools(tool_list.tools) for round_idx in range(max_rounds): response adapter.create(messages, provider_tools) text_parts [] tool_results [] for block in response.content: if block.type text: text_parts.append(block.text) elif block.type tool_use: result await mcp_client.call_tool(block.name, block.input) tr adapter.extract_tool_result(result) tr[tool_use_id] block.id tool_results.append(tr) if not tool_results: return \n.join(text_parts) messages.append({role: assistant, content: response.content}) messages.append({role: user, content: tool_results}) raise RuntimeError(fTool loop did not converge in {max_rounds} rounds)关键点有三个第一response.content必须原样回填到messages模型需要看到自己之前的tool_use第二tool_use_id必须和block.id对应否则 Provider 会报tool_use_id not found第三循环出口是not tool_results而不是固定轮数。4. 验证请求端到端跑一次 Tool Calling Loop配置和代码都齐了现在跑一次完整验证。4.1 启动 Client# main.py import asyncio from config_loader import load_config from mcp_layer import MCPConnection from provider_adapter import ProviderAdapter from loop_layer import run_tool_loop async def main(): provider_cfg, mcp_cfg, loop_cfg load_config() adapter ProviderAdapter(provider_cfg) conn MCPConnection() try: client await conn.connect(mcp_cfg.server_script) tool_list await client.list_tools() print(Connected. Tools:, [t.name for t in tool_list.tools]) query 北京今天适合穿什么衣服 answer await run_tool_loop( client, adapter, query, loop_cfg.max_rounds ) print(\nFinal answer:\n, answer) finally: await conn.close() if __name__ __main__: asyncio.run(main())uv run main.py4.2 预期输出与循环收敛正常运行时你会看到类似这样的过程Connected. Tools: [get_forecast, get_current_weather] [round 0] model requested tool: get_current_weather [round 0] mcp call_tool returned: {temp: 18, condition: cloudy} [round 1] model returned text, loop converged Final answer: 北京当前 18 度多云建议穿薄外套加长裤。循环收敛的标志是某一轮response.content里只有text没有tool_use。这时tool_results为空函数返回最终文本。如果连续 8 轮都在请求 Tool说明要么 Tool 返回的数据模型看不懂要么tool_use_id没对上需要回到第 5 节排查。4.3 多轮 Tool 调用的场景有些任务需要连续调用多个 Tool。比如「对比北京和上海今天天气」模型可能第一轮请求get_current_weather(北京)第二轮请求get_current_weather(上海)第三轮才生成对比文本。上面的循环天然支持这种场景因为每一轮都会把新的tool_result追加到messages模型能看到完整历史。5. 本篇常见错排查5.1tool_use_id not found这是最高频的报错。原因通常是tool_result里的tool_use_id和tool_use的id不一致。检查两点一是block.id是否被正确读取二是messages.append({role: assistant, content: response.content})是否在追加tool_result之前执行。顺序反了Provider 就找不到对应的tool_use。5.2TypeError: NoneType object is not iterable多半是mcp_client.call_tool返回了None或者result.content为空。先确认 MCP Server 的 Tool 函数有返回值再确认call_tool的参数名和 Tool 定义里的input_schema一致。参数名写错时MCP SDK 可能不报错但 Server 端返回空结果。5.3 循环不收敛一直请求 Tool如果模型每轮都请求同一个 Tool通常是 Tool 返回的内容模型无法理解。比如 Server 返回了二进制或非文本 Content Block而适配层只提取了TextContent模型看到的是空字符串就会反复重试。解决方法是让适配层处理ImageContent、EmbeddedResource等类型或者让 Server 统一返回文本。5.4 连接卡住终端无输出stdio_client启动子进程后如果 Server 脚本有语法错误或缺少依赖子进程会直接退出但 Client 可能一直等 stdin。排查方法先单独运行python weather.py确认 Server 能正常启动再检查server_params里的command是否和脚本扩展名匹配。5.5 Key 无效或 Base URL 写错如果报 401 或 404先确认TAOTOKEN_API_KEY环境变量已加载再确认base_url没有多余斜杠。用模型对话入口可以快速验证 Key 是否有效避免在 Client 代码里反复调试鉴权。6. 把 Loop 跑稳之后下一步做什么Tool Calling Loop 跑通只是起点。真实项目里Tool 数量会从几个涨到几十个每轮把所有 Schema 塞给模型既慢又贵。这时候需要引入 Tool Registry 和按需发现先用一个轻量的list_tools摘要让模型选择候选再加载完整 Schema。另外生产环境还要加 Tool 权限控制、用户确认、超时和审计日志这些都不在最小示例的范围内。如果你打算长期做编码类 AgentCoding Plan 提供了更完整的额度和管理能力适合把 MCP Client 接入到日常开发流里。接入过程中遇到鉴权或 Base URL 问题直接查接入文档比翻源码快得多。