ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

大模型Agent实战:MCP协议 + FastAPI 构建 Client/Server 架构,TaoToken 统一 Key 接入

大模型Agent实战:MCP协议 + FastAPI 构建 Client/Server 架构,TaoToken 统一 Key 接入 1. 为什么要把 Agent 工具调用收敛到一条 API 通道大模型 Agent 做工具调用最开始的写法通常很直接在 Agent 代码里写几个函数用 Function Calling 把工具描述塞进 prompt模型返回工具名和参数代码里 if/else 分发执行。工具少的时候没问题一旦工具数量上到十几个、还要跨服务复用问题就集中爆发了。我踩过的坑很典型新增一个检索工具要改 Agent 主逻辑、重新部署整个服务同一个向量检索能力问答 Agent 要用、报表 Agent 也要用只能复制两份代码工具参数 schema 散落在各个文件里改一个字段要全局搜。更麻烦的是工具执行和 Agent 编排耦合在同一个进程工具侧一慢整个 Agent 请求都被拖住。MCPModel Context Protocol模型上下文协议解决的正是这件事。它基于 JSON-RPC 规范把「Agent 调用外部工具、获取外部上下文」的通信格式统一了。工具不再硬编码在 Agent 里而是独立部署成 MCP ServerAgent 侧只保留一个 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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI Keys 管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite拿到 Key 之后记下两个核心信息。Base URL 统一用https://taotoken.net/api注意这个地址后面不加任何查询参数。模型 ID 按你实际要用的填比如做 Agent 决策可以用通用对话模型做代码相关工具可以用编码模型。具体可用模型列表在文档里查。接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite这里有个容易混淆的点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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite环境变量建议这样组织避免 Key 硬编码进代码export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_MODEL_ID你的模型IDPython 侧用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/callparams.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(lifespanlifespan) 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: ftool {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, jsonpayload) 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, jsonpayload) return resp.json()Client 只做两件事封装 JSON-RPC 请求、转发给 Server。它不关心工具内部怎么实现这样工具迭代就不影响 Agent 侧。4. 验证请求一次 Client 发起、Server 响应的完整动作骨架写完了得跑一次完整链路确认能通。先启动 Serveruvicorn 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 指向 TaoTokenfrom openai import OpenAI import os client OpenAI( base_urlos.environ[TAOTOKEN_BASE_URL], api_keyos.environ[TAOTOKEN_API_KEY], ) resp client.chat.completions.create( modelos.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格式必须是rolecontent的列表。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 ServerAgent 侧通过tools/list自动发现不用改代码、不用重启。同一套工具能力问答 Agent 能用报表 Agent 也能用跨服务复用变成默认选项。模型接入这条通道也一样。所有需要模型能力的环节不管是 Agent 决策、文档摘要还是结果润色都走同一个 Base URL 和同一把 Key。Key 轮换只改一个地方用量统计也集中。对于长期跑编码任务或多 Agent 协同的场景可以考虑用 Coding Plan 这类方案把调用额度管理起来。模型对话验证https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteCoding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite下一步可以做的给 MCP Server 加鉴权避免裸奔把工具调用日志打全方便排查用 SSE 做流式返回提升长任务的体验。这些都是在骨架之上叠加的工程细节核心的 Client/Server 解耦和统一 Key 通道已经立住了。
返回列表