1. 为什么要把 Agent 工具调用收敛到一条 API 通道
大模型 Agent 做工具调用，最开始的写法通常很直接：在 Agent 代码里写几个函数，用 Function Calling 把工具描述塞进 prompt，模型返回工具名和参数，代码里 if/else 分发执行。工具少的时候没问题，一旦工具数量上到十几个、还要跨服务复用，问题就集中爆发了。
我踩过的坑很典型：新增一个检索工具，要改 Agent 主逻辑、重新部署整个服务；同一个向量检索能力，问答 Agent 要用、报表 Agent 也要用，只能复制两份代码；工具参数 schema 散落在各个文件里，改一个字段要全局搜。更麻烦的是，工具执行和 Agent 编排耦合在同一个进程，工具侧一慢，整个 Agent 请求都被拖住。
MCP（Model Context Protocol，模型上下文协议）解决的正是这件事。它基于 JSON-RPC 规范，把「Agent 调用外部工具、获取外部上下文」的通信格式统一了。工具不再硬编码在 Agent 里，而是独立部署成 MCP Server，Agent 侧只保留一个 MCP Client 负责协议封装和转发。工具动态发现、热插拔、跨服务共享，新增工具不用动 Agent 代码。
这篇要做的，是用 FastAPI 搭出 MCP 协议下的 Client/Server 双端骨架，让大模型 Agent 通过统一 Key 调用工具链。核心目标有三个：一是给出可复制的 FastAPI 路由与 MCP 消息结构配置；二是演示一次 Client 发起、Server 响应的完整验证动作；三是把多工具调用收敛到一条 API 通道，也就是所有模型请求都走同一个 Base URL 和同一把 Key。
适合谁看：正在做 Agent 工具链、被 Function Calling 硬编码困扰的后端和算法同学；想把工具能力服务化、跨 Agent 复用的团队；以及需要一套统一模型接入通道、不想每个工具各自配 Key 的工程同学。下面从环境准备开始，一步步把骨架跑通。
2. TaoToken 前置准备：统一 Key 与 Base URL 怎么配
在写 MCP 代码之前，先把模型接入这条通道理顺。MCP Server 本身不负责推理，但 Agent 侧要调模型做决策，工具链里也可能有需要模型能力的环节（比如文档解析后的摘要）。如果每个环节各自配一套模型 Key，管理成本会很高。TaoToken 的作用就是把这些调用收敛到一条 API 通道：一个 Base URL、一把 Key、多个模型 ID。
先拿到 Key。打开控制台页面，登录后在 API Keys 区域创建一把新 Key。建议按用途命名，比如mcp-agent-dev，方便后面区分环境。创建后立刻复制保存，页面刷新后完整 Key 不再显示。
- 控制台入口：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
- API Keys 管理：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
拿到 Key 之后，记下两个核心信息。Base URL 统一用https://taotoken.net/api，注意这个地址后面不加任何查询参数。模型 ID 按你实际要用的填，比如做 Agent 决策可以用通用对话模型，做代码相关工具可以用编码模型。具体可用模型列表在文档里查。
- 接入文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
这里有个容易混淆的点：MCP 协议本身和模型接入是两件事。MCP 管的是 Agent 和工具之间的通信格式，模型接入管的是 Agent 怎么调 LLM。两者通过 FastAPI 这个网关层串起来。所以配置上要分两块：一块是 MCP Server 的工具注册，一块是模型调用的 Base URL + Key + Model ID。后者就是所谓「三件套」，任何接入场景都要写全。
如果你用的是 Claude Code 这类编码 Agent，它的配置文件和 MCP 配置是分开的。Claude Code 的接入可以参考专门的配置说明，把 Base URL、Key、Model ID 填到对应位置。MCP 的 Server 配置则写在它自己的配置文件里，通常是 JSON 格式，指定 command、args、env 这些字段。
- Claude Code 接入：https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
环境变量建议这样组织，避免 Key 硬编码进代码：
export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_API_KEY="sk-你的Key" export TAOTOKEN_MODEL_ID="你的模型ID"Python 侧用os.environ读取即可。这样本地开发、容器部署、CI 都能复用同一套变量名，切换环境只改值不改代码。Key 千万不要提交到 Git，.env加进.gitignore。
3. 可复制配置：FastAPI 路由与 MCP 消息结构
这一节是全文的技术核心，给出可以直接复制的配置片段和代码骨架。先看 MCP Server 的配置文件，这是工具注册的入口。以常见的 JSON 配置为例，路径按你实际项目放，字段含义我逐行标注：
{ "mcpServers": { "rag-tools": { "command": "python", "args": ["-m", "mcp_server.rag_server"], "env": { "TAOTOKEN_BASE_URL": "https://taotoken.net/api", "TAOTOKEN_API_KEY": "sk-你的Key", "TAOTOKEN_MODEL_ID": "你的模型ID" } } } }这段配置里，command和args决定 Server 怎么启动，env把三件套注入进去。注意 Base URL 是https://taotoken.net/api，不带 UTM 参数，这是 API 调用的规范地址。
接下来是 MCP 消息结构。MCP 基于 JSON-RPC 2.0，一次工具调用请求长这样：
{ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "hybrid_search", "arguments": { "query": "橡塑配方评审要点", "top_k": 5 } } }对应的响应结构：
{ "jsonrpc": "2.0", "id": 1, "result": { "content": [ { "type": "text", "text": "检索到的文档片段..." } ], "isError": false } }id用于请求响应配对，method是方法名，工具调用固定用tools/call，params.name是工具名，params.arguments是参数对象。响应里content是数组，支持多种内容类型，isError标记是否出错。这套结构统一之后，不管底层工具是检索、解析还是查数据库，Agent 侧看到的格式都一样。
现在写 FastAPI 的 Server 端骨架。用 lifespan 钩子在启动时初始化资源，这是避免每次请求重复创建连接的关键：
from contextlib import asynccontextmanager from fastapi import FastAPI, Request from pydantic import BaseModel import os class ToolCall(BaseModel): jsonrpc: str = "2.0" id: int method: str params: dict TOOLS = {} def register_tool(name, func, schema): TOOLS[name] = {"func": func, "schema": schema} @asynccontextmanager async def lifespan(app: FastAPI): # 启动时装配：注册工具、初始化连接池 register_tool( "hybrid_search", lambda query, top_k: f"检索[{query}] top{top_k} 的结果", {"type": "object", "properties": {"query": {"type": "string"}, "top_k": {"type": "integer"}}} ) app.state.base_url = os.environ["TAOTOKEN_BASE_URL"] app.state.api_key = os.environ["TAOTOKEN_API_KEY"] yield # 关闭时释放资源 TOOLS.clear() app = FastAPI(lifespan=lifespan) @app.post("/mcp") async def mcp_endpoint(call: ToolCall): if call.method == "tools/list": return { "jsonrpc": "2.0", "id": call.id, "result": {"tools": [{"name": k, "inputSchema": v["schema"]} for k, v in TOOLS.items()]} } if call.method == "tools/call": name = call.params.get("name") args = call.params.get("arguments", {}) if name not in TOOLS: return {"jsonrpc": "2.0", "id": call.id, "error": {"code": -32601, "message": f"tool {name} not found"}} result = TOOLS[name]["func"](**args) return {"jsonrpc": "2.0", "id": call.id, "result": {"content": [{"type": "text", "text": result}], "isError": False}} return {"jsonrpc": "2.0", "id": call.id, "error": {"code": -32601, "message": "method not found"}}这段代码里，/mcp是统一入口，tools/list返回工具清单，tools/call执行具体工具。工具注册在 lifespan 里完成，全局复用。Client 端只需要往这个入口发 JSON-RPC 请求即可。
Client 端骨架：
import httpx class MCPClient: def __init__(self, server_url: str): self.server_url = server_url self._id = 0 def _next_id(self): self._id += 1 return self._id async def list_tools(self): payload = {"jsonrpc": "2.0", "id": self._next_id(), "method": "tools/list", "params": {}} async with httpx.AsyncClient() as client: resp = await client.post(self.server_url, json=payload) return resp.json() async def call_tool(self, name: str, arguments: dict): payload = {"jsonrpc": "2.0", "id": self._next_id(), "method": "tools/call", "params": {"name": name, "arguments": arguments}} async with httpx.AsyncClient() as client: resp = await client.post(self.server_url, json=payload) return resp.json()Client 只做两件事：封装 JSON-RPC 请求、转发给 Server。它不关心工具内部怎么实现，这样工具迭代就不影响 Agent 侧。
4. 验证请求：一次 Client 发起、Server 响应的完整动作
骨架写完了，得跑一次完整链路确认能通。先启动 Server：
uvicorn mcp_server.main:app --host 0.0.0.0 --port 8000看到 Uvicorn running 就说明起来了。然后写一个验证脚本，模拟 Client 发起调用：
import asyncio from mcp_client import MCPClient async def main(): client = MCPClient("http://127.0.0.1:8000/mcp") tools = await client.list_tools() print("工具列表:", tools) result = await client.call_tool("hybrid_search", {"query": "配方评审", "top_k": 3}) print("调用结果:", result) asyncio.run(main())预期输出里，tools/list会返回注册过的工具清单，tools/call会返回content数组和isError: false。如果这两步都正常，说明 MCP 的 Client/Server 通道打通了。
接下来把模型接进来，验证统一 Key 这条通道。用 OpenAI 兼容的调用方式，Base URL 指向 TaoToken：
from openai import OpenAI import os client = OpenAI( base_url=os.environ["TAOTOKEN_BASE_URL"], api_key=os.environ["TAOTOKEN_API_KEY"], ) resp = client.chat.completions.create( model=os.environ["TAOTOKEN_MODEL_ID"], messages=[{"role": "user", "content": "用一句话说明 MCP 协议的作用"}], ) print(resp.choices[0].message.content)跑通后你会看到模型返回的内容。这一步验证的是模型接入通道，和 MCP 通道是两条独立的链路，但都收敛到同一把 Key 和同一个 Base URL 下。Agent 的完整流程就是：模型决策 → 返回工具调用意图 → MCP Client 封装请求 → MCP Server 执行 → 结果回填 → 模型生成最终答案。
把这两步串起来，一个最小的 Agent 工具调用闭环就跑通了。你可以在这个基础上加更多工具，只要在 lifespan 里注册，Client 侧通过tools/list就能动态发现，不用改 Agent 主逻辑。
5. 本篇常见错排查：401、local proxy failed、reading choices、OAuth
实际跑的时候，报错基本集中在几个地方。我按真实遇到的顺序列一下，对照排查。
401 Unauthorized。这个最常见，原因是 Key 没传对或者传了空值。检查三处：环境变量TAOTOKEN_API_KEY是否真的导出成功（echo $TAOTOKEN_API_KEY看一下）；代码里读取的变量名是否和导出的一致；Key 是否带了多余空格。还有一种情况是 Key 被撤销了，去控制台确认状态。注意 Base URL 要用https://taotoken.net/api，不要自己拼路径。
local proxy failed。这个报错通常出现在网络层，说明请求根本没发到目标地址。检查 Base URL 是否写错、端口是否被占用、容器网络是否能通外网。如果是本地开发，确认没有多余的代理环境变量干扰，unset http_proxy https_proxy再试。这个报错和 Key 无关，是链路问题。
reading choices 相关报错。典型的是KeyError: 'choices'或者reading 'choices'时对象为 None。原因是响应结构和你预期的不一致，可能是模型 ID 填错了，返回了错误对象而不是正常的 completion。打印完整resp看一下，确认model字段是你配置的那个。另外检查messages格式，必须是role+content的列表。
OAuth 相关报错。如果你在 Claude Code 或某些客户端里看到 OAuth 报错，通常是客户端的认证方式和 API Key 方式冲突了。API Key 接入不需要走 OAuth 流程，检查客户端配置里是不是误开了 OAuth 选项。把认证方式改成 API Key，填上 Base URL、Key、Model ID 三件套。
排查顺序建议：先确认环境变量 → 再确认 Base URL → 再确认模型 ID → 最后看响应体。大部分问题在前两步就能定位。如果工具调用报tool not found，检查工具名是否和注册时一致，大小写敏感。
6. 把工具链收敛到一条通道之后
骨架跑通之后，工程上的收益会慢慢显现。工具独立部署，新增检索能力只要起一个新的 MCP Server，Agent 侧通过tools/list自动发现，不用改代码、不用重启。同一套工具能力，问答 Agent 能用，报表 Agent 也能用，跨服务复用变成默认选项。
模型接入这条通道也一样。所有需要模型能力的环节，不管是 Agent 决策、文档摘要还是结果润色，都走同一个 Base URL 和同一把 Key。Key 轮换只改一个地方，用量统计也集中。对于长期跑编码任务或多 Agent 协同的场景，可以考虑用 Coding Plan 这类方案，把调用额度管理起来。
- 模型对话验证：https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
- Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
下一步可以做的：给 MCP Server 加鉴权，避免裸奔；把工具调用日志打全，方便排查；用 SSE 做流式返回，提升长任务的体验。这些都是在骨架之上叠加的工程细节，核心的 Client/Server 解耦和统一 Key 通道已经立住了。
