1. 从零开发大模型 Agent,为什么工具调用链路总是卡住
大模型 Agent(智能体)说白了就是让模型自己决定“下一步该干什么”,而工具调用是它真正能干活的关键。你给它一个天气查询接口、一个开关窗函数,它就能根据用户一句话自动规划:先查天气,再判断要不要关窗。听起来很顺,但真正动手写的时候,问题往往不在模型本身,而在“模型怎么知道有哪些工具可用、怎么把工具结果喂回去、怎么管理多个模型的访问凭证”。
我见过太多人卡在三个地方:第一,每个模型厂商的 API Key 格式、Base URL、鉴权方式都不一样,写一个 Agent 要维护三四套配置;第二,MCP(Model Context Protocol)工具注册后,模型返回的 tool_calls 结构解析经常报错,尤其是reading choices这类字段缺失;第三,本地调试时环境变量没配对,直接local proxy failed或者 401。这些问题单独看都不难,但凑在一起就会让一个最小可用 Agent 拖好几天。
这篇就按“能跑通”的标准来:用 TaoToken 统一管理模型访问的 Key 和通道,用 MCP 协议把工具注册成标准接口,最后写一个可复制的 Agent 骨架,端到端验证一次“查天气→判断→开关窗”的完整链路。适合已经会 Python、想快速搭一个能用的 Agent 的开发者,也适合被多模型配置折腾过的人。
核心检索词先明确:大模型 Agent 开发、MCP 工具调用、统一 Key 管理、智能体骨架配置。下面从环境准备开始,每一步都给可复制的代码和配置。
2. TaoToken 统一 Key 与 MCP 工具链路的前置准备
在写 Agent 之前,先把“模型访问”这一层收拢。TaoToken 的作用是提供一个统一的 API 通道,你不需要为每个模型单独记 Base URL 和 Key,而是用一套凭证访问多个模型。官网入口是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 地址是 https://taotoken.net/api (注意这个不加 UTM 参数,直接用于代码里的 base_url)。
你需要准备的东西不多:一个 TaoToken 账号,创建一个 API Key;Python 3.10 以上环境;安装mcp、openai、requests、cachetools这几个库。安装命令如下:
pip install mcp openai requests cachetools这里有个容易踩的坑:mcp库的版本更新比较快,建议用pip install mcp --upgrade确保拿到支持FastMCP和stdio_client的最新版。如果你之前装过旧版,可能会出现ImportError: cannot import name 'FastMCP',直接升级即可。
关于 Key 的获取,进入 TaoToken 控制台创建 API Key,地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。创建后复制那串sk-开头的字符串,后面配置里会用到。如果你打算长期跑编码类 Agent,可以顺便看一下 Coding Plan 的入口 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,它更适合高频调用场景。
MCP 这一层的前置知识只需要记住三个原语:Tools(可执行函数)、Resources(上下文数据)、Prompts(预定义模板)。我们这篇重点用 Tools,因为 Agent 的工具调用主要靠它。MCP 的通信基于 JSON-RPC 2.0,客户端和服务端之间是有状态会话,所以你会看到代码里大量用async和AsyncExitStack来管理连接生命周期。
配置统一 Key 的时候,建议用环境变量而不是硬编码。在项目根目录建一个.env文件:
TAOTOKEN_API_KEY=sk-你的实际Key TAOTOKEN_BASE_URL=https://taotoken.net/api然后在 Python 里用os.getenv读取。这样做的好处是,后面无论你换模型还是换工具,凭证只改一处。如果你用的是 Claude Code 这类工具做辅助开发,它的配置里也需要填 Base URL、Key、Model ID 三件套,Base URL 同样是https://taotoken.net/api,Model ID 按你实际调用的模型名填。
前置准备做完后,目录结构建议这样组织,后面所有代码都基于这个结构:
agent-demo/ ├── .env ├── server.py # MCP 服务端,注册工具 ├── mcp_client.py # MCP 客户端封装 ├── agent.py # Agent 主逻辑 └── main.py # 端到端测试入口这样拆的好处是,工具注册、连接管理、Agent 决策三层解耦,排错时能快速定位是哪一层的问题。
3. 可复制的 Agent 骨架配置与 MCP 工具注册示例
先写 MCP 服务端server.py,把天气查询和窗户控制注册成标准工具。这里用FastMCP,它会把函数的签名、参数类型、docstring 自动解析成模型能理解的 JSON Schema。
# server.py import requests from mcp.server.fastmcp import FastMCP mcp = FastMCP("IoTServer") @mcp.tool() def weather_query(location: str): """根据提供的城市名,查询天气情况""" api_key = "你的高德天气Key" base_url = "https://restapi.amap.com/v3/weather/weatherInfo" params = {"key": api_key, "city": location} response = requests.get(base_url, params=params, timeout=10) if response.status_code == 200: data = response.json() return data["lives"][0] return {"error": "无法获取天气信息,请检查城市名称"} @mcp.tool() def window_control(status: str): """控制窗户的开和关,status 有开和关两种""" if status == "开": return "窗户已打开" elif status == "关": return "窗户已关闭" return "不支持该操作:" + status if __name__ == "__main__": mcp.run()注意@mcp.tool()装饰器下面的 docstring 很重要,模型就是靠它来判断这个工具是干什么的。写得太模糊,模型选错工具的概率会明显上升。
接下来是 MCP 客户端封装mcp_client.py,负责连接服务端、缓存工具列表、执行工具调用:
# mcp_client.py import asyncio from contextlib import AsyncExitStack import cachetools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class MCPClient: def __init__(self): self.session = None self.exit_stack = AsyncExitStack() self.tools_ttl_cache = cachetools.TTLCache(maxsize=5, ttl=5) self.CACHE_TOOLS_KEY = "tools" async def connect_to_server(self, server_script_path: str): server_params = StdioServerParameters( command="python", args=[server_script_path], env=None ) stdio_transport = await self.exit_stack.enter_async_context( stdio_client(server_params) ) self.stdio, self.write = stdio_transport self.session = await self.exit_stack.enter_async_context( ClientSession(self.stdio, self.write) ) await self.session.initialize() await self.list_tools() async def list_tools(self): if self.CACHE_TOOLS_KEY in self.tools_ttl_cache: return self.tools_ttl_cache[self.CACHE_TOOLS_KEY] response = await self.session.list_tools() available_tools = [{ "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": tool.inputSchema } } for tool in response.tools] self.tools_ttl_cache[self.CACHE_TOOLS_KEY] = available_tools return available_tools async def call_tool(self, tool_name, tool_args): result = await self.session.call_tool(tool_name, tool_args) return result async def cleanup(self): await self.exit_stack.aclose()这里用TTLCache缓存工具列表,ttl 设 5 秒,避免 Agent 每轮都去问服务端“有哪些工具”。实测下来,这个缓存能明显减少 stdio 通信次数,尤其在多轮工具调用时。
然后是 Agent 主逻辑agent.py,用 TaoToken 的统一通道调用模型,自动选择工具:
# agent.py import os import json from openai import OpenAI class MCPAgent: def __init__(self, mcp_client): self.mcp_client = mcp_client self.client = OpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api") ) self.llm_model_name = "glm-4-plus" async def process_message(self, query: str) -> str: messages = [{"role": "user", "content": query}] message = await self.select_tools(messages) final_text = [message.content or ""] while message.tool_calls: for tool_call in message.tool_calls: tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) result = await self.mcp_client.call_tool(tool_name, tool_args) final_text.append( f"[Calling tool {tool_name} with args {tool_args}]" ) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result.content) }) message = await self.select_tools(messages) final_text.append(message.content or "") return "\n".join(final_text) async def select_tools(self, messages): available_tools = await self.mcp_client.list_tools() response = self.client.chat.completions.create( model=self.llm_model_name, messages=messages, tools=available_tools ) return response.choices[0].message这段代码里,base_url指向 TaoToken 的 API 地址,api_key从环境变量读。模型名按你实际可用的填,这里用glm-4-plus只是示例。如果你要换成别的模型,只改llm_model_name这一行,Key 和 Base URL 都不用动,这就是统一通道的好处。
如果你用的是 Claude Code 做辅助编码,它的 settings 配置里同样需要三件套。一个可复制的 settings 片段如下:
{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_API_KEY": "sk-你的实际Key", "ANTHROPIC_MODEL": "claude-sonnet-4-20250514" } }注意路径和字段名要和你实际使用的工具版本一致,不同版本可能字段名有差异。Cline MCP 的配置也是类似逻辑,Base URL、Key、Model ID 三件套缺一不可。
4. 端到端验证请求与成功结果解析
现在写main.py做端到端测试:
# main.py import asyncio from mcp_client import MCPClient from agent import MCPAgent async def main(): mcp_client = MCPClient() agent = MCPAgent(mcp_client) try: await mcp_client.connect_to_server("./server.py") message = "查询今天深圳的天气情况。如果天气是下雨,就关上窗户;否则打开窗户。" final_text = await agent.process_message(message) print(final_text) finally: await mcp_client.cleanup() if __name__ == "__main__": asyncio.run(main())运行前确保.env已加载,可以用python-dotenv或者在终端里export。执行python main.py,预期输出类似:
[Calling tool weather_query with args {'location': '深圳'}] [Calling tool window_control with args {'status': '开'}] 今天深圳的天气情况是阴天,气温27℃,东南风3级,湿度56%。因为没有下雨,所以窗户已经打开了。这个结果说明整条链路通了:模型先选了weather_query,拿到天气后判断没下雨,又选了window_control并传入开,最后汇总成自然语言回复。整个过程 Agent 自主完成了两轮工具调用,没有人工干预。
如果你想验证模型对话本身是否正常,可以单独用 TaoToken 的模型对话入口 https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 发一条测试消息,确认 Key 和通道没问题,再跑 Agent。这样能把“模型访问问题”和“工具调用问题”分开排查。
验证时重点看三个信号:第一,list_tools返回的工具列表里有没有你注册的两个工具;第二,模型返回的tool_calls字段是否非空且结构完整;第三,工具执行结果是否正确拼回messages。这三个都正常,链路就没问题。
5. 本篇常见错误排查:401、local proxy failed 与 reading choices
实际跑的时候,报错基本集中在这几类,逐个说清楚怎么定位。
401 Unauthorized:最常见的原因是 Key 没读到或者读错了。先检查.env是否被正确加载,可以在agent.py里临时打印os.getenv("TAOTOKEN_API_KEY")[:8]确认前几位。如果 Key 是对的还报 401,检查base_url是否写成了https://taotoken.net/api,注意不要多加斜杠或者写成别的路径。另外,如果你在代码里同时设置了OPENAI_API_KEY环境变量,可能会被覆盖,建议显式传参。
local proxy failed:这个报错通常出现在 stdio 连接阶段,说明 MCP 客户端启动服务端脚本失败了。检查server.py路径是否正确,StdioServerParameters里的command是不是当前环境可用的python。如果你用的是虚拟环境,command要指向虚拟环境里的 python 绝对路径,否则可能用了系统 python 导致mcp库找不到。实测下来,用sys.executable替代硬编码的"python"最稳。
reading choices 相关报错:典型的是KeyError: 'choices'或者AttributeError: 'NoneType' object has no attribute 'choices'。这说明模型返回的响应结构不符合预期。先确认response本身不是 None,再检查response.choices是否存在。常见原因是模型名写错了,或者 tools 参数格式不对导致请求被拒。可以在select_tools里加一行print(response)看原始返回。如果返回里带error字段,按错误信息调整。
OAuth 相关报错:如果你在配置 Claude Code 或类似工具时看到 OAuth 报错,通常是鉴权方式选错了。用 API Key 方式接入时,不需要走 OAuth 流程,直接填 Base URL 和 Key 即可。检查配置文件里是否误开了 OAuth 开关,或者 Model ID 填成了需要 OAuth 的模型。
工具调用结果拼不回上下文:表现为 Agent 一直循环调用同一个工具。检查messages.append里tool_call_id是否和模型返回的tool_call.id一致,role是否为"tool"。这两个字段错一个,模型就无法把结果和调用关联起来。
排错时建议按“模型访问→工具列表→工具调用→结果回填”的顺序逐层验证,不要一上来就改 Agent 逻辑。大部分问题其实在前两层。
6. 语义一致的接入入口与后续扩展方向
跑通最小 Agent 之后,下一步通常是扩展工具数量和模型切换。工具多了以后,list_tools的缓存策略要调整,ttl 可以适当延长,但要注意工具变更时的失效。模型切换只需要改llm_model_name,Key 和 Base URL 保持不变,这是统一通道最直接的价值。
如果你要接入更多 MCP 工具,注册方式和server.py里一样,加@mcp.tool()装饰器即可。工具描述要写清楚输入输出,模型选工具的准确率会高很多。对于需要多步规划的复杂任务,可以在 Agent 循环里加最大轮次限制,避免无限调用。
需要创建新的 API Key 或查看用量,走控制台入口 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,里面有各语言的调用示例。API Keys 管理页是 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,建议给不同项目建不同的 Key,方便排查和限额。
最后说一个实用技巧:调试 Agent 时,把每轮messages完整打印出来,尤其是tool_calls和tool消息的对应关系。很多“模型不听话”的问题,其实是上下文里工具结果没拼对。这个习惯能省掉大量猜测时间。