智能体架构中的协议设计三部曲：MCP → A2A → AG-UI 与 TaoToken 统一 Key 通道实践

1. 三层协议到底解决什么问题从一次本地智能体联调说起如果你正在做智能体应用大概率遇到过这种局面工具调用写一套、智能体之间通信写一套、前端流式渲染再写一套三套代码互不相干改一个字段要动三个仓库。MCP、A2A、AG-UI 这三个协议本质上就是把这三段通信各自标准化让「工具接入」「智能体协作」「前端交互」不再各写各的。先把三个角色摆清楚。用户在前端界面输入一句话前端需要把这句话交给智能体智能体在执行过程中要调用外部工具查数据库、读文件、发请求如果任务复杂还要把子任务转交给另一个智能体最后把执行过程和结果实时推回前端。这条链路里MCP 负责智能体与外部工具之间的标准化交互A2A 负责智能体与智能体之间的标准通信AG-UI 负责前端应用与智能体之间的标准交互。三者不是竞争关系而是分层互补同一个智能体可以一边通过 A2A 和另一个智能体对话一边通过 AG-UI 把状态推给用户同时通过 MCP 调用工具。为什么需要标准化因为现在的智能体框架太多LangGraph、AutoGen、CrewAI、ADK 各有各的写法工具定义方式不同状态传递方式不同前端适配方式也不同。你用一个框架写完想换另一个框架几乎等于重写。协议层的价值就在于把这些差异收敛到标准接口上框架可以换协议不变。我试过把三层协议拆开单独跑再拼起来联调发现最容易出问题的不是协议本身而是鉴权通道。三个协议各自要调模型、各自要带 Key如果每个协议都维护一套密钥管理成本会迅速失控。所以这篇的实践路线是用 TaoToken 统一 Key 通道承接三层协议的模型调用MCP 服务端、A2A 消息路由、AG-UI 事件绑定都走同一个 Base URL 和同一把 Key减少配置分叉。适合谁看如果你已经写过至少一个能跑通的智能体 demo想把它拆成可维护的三层结构或者你正在做多智能体协作 前端实时交互的产品原型这篇的配置和验证步骤可以直接跟做。如果你还没接触过 MCP建议先把 MCP 服务端跑通再回来看 A2A 和 AG-UI 部分。下面按「前置准备 → MCP 配置 → A2A 路由 → AG-UI 事件绑定 → 联调验证 → 排错」的顺序展开每一步都给可复制的配置和验证命令。2. TaoToken 统一 Key 通道前置准备一个 Base URL 打通三层协议三层协议联调时模型调用会出现在三个位置MCP 服务端里工具执行后需要模型总结、A2A 路由中远程智能体需要模型推理、AG-UI 后端在流式推送时需要模型生成 token。如果这三处分别配置不同的模型服务地址和密钥联调时你会花大量时间在「哪个 Key 过期了」「哪个地址写错了」上。TaoToken 在这里的角色是统一模型调用通道。你只需要一个 API Key 和一个 Base URL三层协议里的模型请求都指向同一个入口。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 注意 API 地址不带 UTM 参数配置时直接用这个。先拿 Key。进入控制台创建 API Key路径是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 创建后复制保存后面三层配置都要用。如果你还没决定用哪个模型可以先在模型对话页测试一下连通性地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 选一个模型发一条消息确认返回正常再往下走。Key 拿到后建议先写一个最小验证脚本确认 Base URL 和 Key 能通。用 curl 测试curl https://taotoken.net/api/v1/chat/completions \ -H Content-Type: application/json \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -d { model: claude-sonnet-4-20250514, messages: [{role: user, content: ping}], max_tokens: 16 }如果返回里有choices字段和内容说明通道正常。这一步很重要因为后面三层协议联调时任何一层报错你都要先排除「是不是 Key 或 Base URL 的问题」。把 Key 写进环境变量不要硬编码在代码里export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api模型 ID 的选择上MCP 工具执行后的总结建议用响应快的模型A2A 路由中的任务规划建议用推理能力强的模型AG-UI 流式输出建议用支持流式的模型。你可以在模型对话页逐个试确认哪些模型支持流式返回。实测下来同一把 Key 可以调用多个模型切换模型只需要改model字段不需要换 Key 或换地址。这里有个容易踩的坑有些框架默认读OPENAI_API_KEY和OPENAI_BASE_URL你需要把 TaoToken 的 Key 和 Base URL 映射到这些变量上或者显式在配置里指定。后面每一层的配置我都会写清楚用哪个变量名。前置准备做完你应该有一个可用的 API Key、确认可通的 Base URL、一个测试通过的模型 ID。接下来进入 MCP 服务端配置。3. MCP 服务端可复制配置stdio 与 SSE 两种接入方式MCP 服务端是三层协议里最底层的一环它把外部工具标准化暴露给智能体。MCP 的核心角色有三个Host 是使用 LLM 的程序Client 是与 Server 保持 1:1 连接的客户端Server 是暴露具体功能的轻量程序。Server 对外暴露三类元素Prompts 由用户控制Resources 由应用控制Tools 由模型控制。联调时最常用的是 Tools因为工具调用是智能体执行任务的主要手段。先写一个最小 MCP 服务端用 Python 实现暴露一个查询天气的工具。安装依赖pip install mcp httpx创建weather_server.pyfrom mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import httpx app Server(weather-server) app.list_tools() async def list_tools(): return [ Tool( nameget_weather, description查询指定城市的天气, inputSchema{ type: object, properties: { city: {type: string, description: 城市名} }, required: [city] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name get_weather: city arguments[city] # 这里替换为真实天气 API示例返回固定结构 return [TextContent(typetext, textf{city} 今天晴25 摄氏度)] raise ValueError(f未知工具: {name}) async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这是 stdio 方式适合本地开发Host 通过标准输入输出与 Server 通信。如果你要让远程智能体也能调用这个工具需要改成 SSE 方式。SSE 服务端配置from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Route, Mount sse SseServerTransport(/messages/) async def handle_sse(request): async with sse.connect_sse( request.scope, request.receive, request._send ) as streams: await app.run( streams[0], streams[1], app.create_initialization_options() ) app_star Starlette(routes[ Route(/sse, endpointhandle_sse), Mount(/messages/, appsse.handle_post_message), ])启动 SSE 服务uvicorn weather_server:app_star --host 0.0.0.0 --port 8080MCP 服务端本身不直接调模型但工具执行后的结果往往需要模型总结。这时候在 Host 侧调用 TaoToken 通道。Host 配置示例以 Claude Desktop 的claude_desktop_config.json为例{ mcpServers: { weather: { command: python, args: [/path/to/weather_server.py], env: { TAOTOKEN_API_KEY: sk-你的key, TAOTOKEN_BASE_URL: https://taotoken.net/api } } } }如果你用的是 Cline 或 CC Switch 这类支持 MCP 的客户端配置结构类似关键是三件套要写全Base URL 填https://taotoken.net/apiKey 填你的 API KeyModel ID 填你测试通过的模型。Cline 的 MCP 配置在设置面板里添加 Server 时选择 stdio 或 SSE把上面的 JSON 片段贴进去即可。验证 MCP 服务端是否正常可以用 MCP Inspectornpx modelcontextprotocol/inspector python weather_server.py打开浏览器界面点击 List Tools应该能看到get_weather。点击 Call Tool传入{city: 北京}应该返回天气文本。这一步通了说明 MCP 层没问题。注意MCP 服务端不要直连生产数据库工具实现里要做参数校验和权限控制。示例里的天气查询是只读操作实际业务中写操作要加确认机制。4. A2A 消息路由与 AG-UI 事件绑定从 Agent Card 到 SSE 流A2A 解决的是智能体之间的通信。核心概念是 Agent Card每个实现 A2A 的智能体通过 Agent Card 公开自己的能力目录其他智能体通过这个目录发现可用功能。Agent Card 通常放在https://domain/.well-known/agent.json本地开发时可以用http://localhost:8001/.well-known/agent.json。先写一个最小 A2A 服务端暴露一个翻译能力。安装依赖pip install fastapi uvicorn httpx创建a2a_server.pyfrom fastapi import FastAPI, Request from fastapi.responses import JSONResponse import httpx import os app FastAPI() AGENT_CARD { name: translator-agent, description: 中英互译智能体, url: http://localhost:8001, capabilities: [translate], endpoints: { tasks: /tasks } } app.get(/.well-known/agent.json) async def agent_card(): return AGENT_CARD app.post(/tasks) async def create_task(request: Request): body await request.json() text body.get(text, ) target_lang body.get(target_lang, en) # 调用 TaoToken 通道做翻译 async with httpx.AsyncClient() as client: resp await client.post( f{os.environ[TAOTOKEN_BASE_URL]}/v1/chat/completions, headers{ Authorization: fBearer {os.environ[TAOTOKEN_API_KEY]}, Content-Type: application/json }, json{ model: claude-sonnet-4-20250514, messages: [ {role: user, content: f把下面内容翻译成{target_lang}{text}} ] }, timeout30 ) result resp.json()[choices][0][message][content] return JSONResponse({task_id: t1, status: completed, result: result})启动export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api uvicorn a2a_server:app --port 8001验证 Agent Cardcurl http://localhost:8001/.well-known/agent.json验证任务调用curl -X POST http://localhost:8001/tasks \ -H Content-Type: application/json \ -d {text: 你好世界, target_lang: en}返回翻译结果说明 A2A 层通了。A2A 建立在 HTTP、SSE、JSON-RPC 之上默认支持企业级鉴权实际部署时在/tasks上加 Bearer Token 校验即可。AG-UI 负责前端与智能体的实时交互。核心机制是客户端 POST 启动会话建立 SSE 流智能体持续推送事件前端根据事件类型更新界面。事件类型包括TEXT_MESSAGE_CONTENTtoken 流式、TOOL_CALL_START工具执行、STATE_DELTA状态更新、AGENT_HANDOFF智能体交接。后端 Python 端接入pip install ag-ui-protocol创建agui_server.pyfrom fastapi import FastAPI from fastapi.responses import StreamingResponse from ag_ui.core import TextMessageContentEvent, EventType from ag_ui.encoder import EventEncoder import httpx import os import json app FastAPI() encoder EventEncoder() async def event_stream(prompt: str): async with httpx.AsyncClient() as client: async with client.stream( POST, f{os.environ[TAOTOKEN_BASE_URL]}/v1/chat/completions, headers{ Authorization: fBearer {os.environ[TAOTOKEN_API_KEY]}, Content-Type: application/json }, json{ model: claude-sonnet-4-20250514, messages: [{role: user, content: prompt}], stream: True }, timeout60 ) as resp: async for line in resp.aiter_lines(): if line.startswith(data: ): data line[6:] if data [DONE]: break chunk json.loads(data) delta chunk[choices][0][delta].get(content, ) if delta: event TextMessageContentEvent( typeEventType.TEXT_MESSAGE_CONTENT, message_idmsg_1, deltadelta ) yield encoder.encode(event) app.post(/agui) async def agui_endpoint(payload: dict): prompt payload.get(prompt, ) return StreamingResponse( event_stream(prompt), media_typetext/event-stream )启动uvicorn agui_server:app --port 8002前端 TypeScript 端接入npm install ag-ui/core ag-ui/client前端订阅const response await fetch(http://localhost:8002/agui, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt: 写一段自我介绍 }) }); const reader response.body.getReader(); const decoder new TextDecoder(); while (true) { const { done, value } await reader.read(); if (done) break; const text decoder.decode(value); // 解析 SSE 事件根据 type 更新 UI console.log(text); }验证 AG-UI 流curl -N -X POST http://localhost:8002/agui \ -H Content-Type: application/json \ -d {prompt: 你好}应该看到逐条data: {...}事件输出。到这里三层协议各自跑通了。接下来做联调。5. 三层协议联调常见报错排查401、local proxy failed、reading choices联调阶段最容易出问题的是鉴权链路和流式解析。下面按真实报错逐个排查。报错一401 Unauthorized。出现在任何一层调用 TaoToken 通道时。先检查环境变量是否导出echo $TAOTOKEN_API_KEY echo $TAOTOKEN_BASE_URL如果为空说明当前 shell 没加载。注意 Base URL 不要带 UTM 参数正确值是https://taotoken.net/api。如果 Key 正确但仍 401检查请求头格式必须是Authorization: Bearer sk-xxxBearer 后面有一个空格。另外确认 Key 没有过期可以在控制台重新生成一个。报错二local proxy failed。这个报错通常出现在 MCP 客户端连接 SSE 服务端时。原因是客户端配置的 URL 和服务端实际监听地址不一致。检查服务端启动命令里的 host 和 port如果是0.0.0.0:8080客户端要填http://localhost:8080/sse不要填0.0.0.0。如果客户端在容器里localhost 指向容器本身要改成宿主机的实际 IP。另外确认 SSE 路由路径是/ssePOST 消息路径是/messages/两者不能混。报错三reading choices 相关错误。典型信息是KeyError: choices或list index out of range。这说明模型返回结构和你预期的不一致。先打印完整响应print(resp.status_code) print(resp.text)常见原因有三个一是模型 ID 写错返回了错误信息而不是正常结构二是请求体里stream参数和解析逻辑不匹配流式返回的 chunk 结构和非流式不同三是超时导致返回空。修复方式是先确认模型 ID 在模型对话页可用再确认流式解析里取的是chunk[choices][0][delta]而不是[message]。报错四OAuth 相关错误。如果你用的客户端要求 OAuth 登录但你想用 API Key需要在客户端设置里切换到 API Key 模式。以 Codex 为例auth.json配置{ api_key: sk-你的key, base_url: https://taotoken.net/api }如果客户端同时支持 OAuth 和 API Key确认没有同时启用否则会优先走 OAuth 导致鉴权失败。报错五A2A 任务超时。远程智能体执行时间长客户端等待超时。A2A 协议本身支持长期任务需要在 Agent Card 里声明任务状态查询端点客户端轮询而不是同步等待。最小改法是在/tasks返回task_id和status: pending再加一个/tasks/{task_id}查询端点。报错六AG-UI 事件不更新 UI。检查 SSE 响应头是否包含Content-Type: text/event-stream以及每条事件是否以\n\n结尾。EventEncoder 已经处理了结尾如果你手动拼接记得加双换行。另外前端解析时注意data:后面有一个空格。排查顺序建议先单独验证每一层再两两联调最后三层一起跑。每层验证时都用 curl 直接打接口排除前端和框架的干扰。6. 把三层协议串成一条可维护链路统一 Key 通道的长期价值三层协议各自跑通后真正的挑战是维护。MCP 服务端可能从 1 个变成 10 个A2A 智能体可能跨团队部署AG-UI 前端可能同时对接多个后端。如果每个新增节点都配一套密钥管理成本会指数上升。统一 Key 通道的价值在这里体现所有模型调用都走同一个 Base URL 和同一把 Key新增节点只需要复制环境变量不需要重新申请和配置。TaoToken 的 API 入口https://taotoken.net/api在三层协议里保持一致MCP 服务端的 env、A2A 服务端的 httpx 调用、AG-UI 后端的流式请求用的都是同一套凭证。如果你要把这套链路用于长期编码或 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 里面有各协议的详细参数说明。API Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 可以创建多个 Key 做环境隔离。最后给一个实用技巧把三层协议的启动脚本写成一个start_all.sh统一导出环境变量按顺序启动 MCP、A2A、AG-UI 服务联调时一条命令拉起全部#!/bin/bash export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api python weather_server.py uvicorn a2a_server:app --port 8001 uvicorn agui_server:app --port 8002 wait这样每次联调不用重复配置改 Key 也只改一处。三层协议的分层设计加上统一 Key 通道能让你的智能体架构在扩展时保持可控。