作者：PySuper | 来源：zhengxingtao.com
2024 年大家比谁家的模型更强，2025 年大家比谁家的 Agent 更多，到了 2026 年——大家开始比谁家的协议更标准。MCP、A2A、AG-UI 三大协议井喷式发展，很多人第一反应是：这仨是竞争关系吗？我该用哪个？答案是：它们不是竞争，而是互补——就像 TCP、HTTP 和 HTML 的关系，各管一层。这篇文章带你彻底搞懂。
一、2026：从"模型竞赛"到"协议竞赛"
先回顾一下时间线：
plaintext
2023 模型竞赛 ──── GPT-4 vs Claude vs Gemini，比谁更聪明
2024 Agent竞赛 ── AutoGPT / CrewAI / LangGraph，比谁能干活
2025 工具竞赛 ── MCP 爆发，比谁能接工具
2026 协议竞赛 ── MCP + A2A + AG-UI，比谁能互联互通
为什么协议突然变得这么重要？因为 Agent 从"玩具"走向"生产"的过程中，遇到了三个硬伤：
工具碎片化：每个框架自己搞一套工具集成，换框架就要重写
Agent 孤岛：不同框架/不同团队构建的 Agent 无法协作
前端黑盒：Agent 跑了五分钟，用户只看到一个 loading 转圈
三大协议正好对应解决这三个问题。
二、MCP 深度回顾：工具集成层
2.1 MCP 是什么
MCP（Model Context Protocol）由 Anthropic 于 2024 年底发布开源，定位是 "大模型与外部世界的桥梁" ，被社区形象地称为 "USB-C for AI" 。
一句话概括：MCP 解决的是 Agent 怎么调用工具的问题。
2.2 核心架构
plaintext
┌─────────────────────────────────────────────────┐
│ Host │
│ (Claude Desktop / Cursor / IDE) │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ MCP │ │ MCP │ │ MCP │ │
│ │ Client A │ │ Client B │ │ Client C │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
└───────┼───────────────┼───────────────┼─────────┘
│ │ │
▼ ▼ ▼
┌───────────┐ ┌───────────┐ ┌───────────┐
│ MCP │ │ MCP │ │ MCP │
│ Server │ │ Server │ │ Server │
│ (Git) │ │ (Notion) │ │ (Database)│
│ │ │ │ │ │
│ ├─Tools │ │ ├─Tools │ │ ├─Tools │
│ ├─Resources│ │ ├─Resources│ │ ├─Resources│
│ └─Prompts │ │ └─Prompts │ │ └─Prompts │
└───────────┘ └───────────┘ └───────────┘
三个核心概念：
表格
2.3 通信协议
MCP 基于 JSON-RPC 2.0 进行消息传输，支持三种传输方式：
python
# 传输方式对比
transports = {
"stdio": {
"desc": "本地子进程通信",
"latency": "最低",
"use_case": "本地开发、CLI 工具",
"example": "Claude Desktop 调用本地 Git Server"
},
"Streamable HTTP": {
"desc": "HTTP + SSE 流式传输",
"latency": "中等",
"use_case": "生产环境、远程部署",
"example": "AWS Bedrock AgentCore Runtime"
},
"gRPC": {
"desc": "高性能远程调用",
"latency": "比 HTTP 低 65%",
"use_case": "低延迟生产场景",
"example": "高频交易、实时数据分析"
}
}
2.4 MCP Server 暴露的三种能力
python
# 1. Tools —— Agent 可调用的函数
{
"name": "query_database",
"description": "Execute SQL query on the analytics database",
"inputSchema": {
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SQL query"},
"limit": {"type": "integer", "default": 100}
},
"required": ["sql"]
}
}
# 2. Resources —— Agent 可读取的数据
{
"uri": "db://analytics/schema",
"name": "Analytics Database Schema",
"description": "Read-only schema information",
"mimeType": "application/json"
}
# 3. Prompts —— 可复用的提示模板
{
"name": "code_review",
"description": "Review code for best practices",
"arguments": [
{"name": "code", "required": True},
{"name": "language", "required": False}
]
}
2.5 生态现状
截至 2026 年中，MCP 已经成为 Agent 工具集成的事实标准：
表格
一个真实案例：GitHub Copilot 通过 MCP 调用本地 Git 工具，自动执行 git add/commit/push，开发者手动操作减少 70%。
2.6 MCP 的局限性
踩坑提醒：
紧耦合问题：工具升级时可能出现兼容性问题。某 AI 绘画工具升级 API 后，15% 的调用方出现兼容性故障
不是 Agent 间通信：如果你的研究 Agent 需要委派子任务给编码 Agent，MCP 不是正确选择——那是 A2A 的领域
没有 UI 事件流：MCP 不支持前端实时更新事件——那是 AG-UI 的领域
三、A2A 深度回顾：Agent 间通信层
3.1 A2A 是什么
A2A（Agent-to-Agent Protocol）由 Google 于 2025 年 4 月推出，并捐赠给 Linux 基金会，定位是 "智能体间的通用语言" ，被社区称为 "HTTP for Agents" 。
一句话概括：A2A 解决的是 Agent 怎么跟 Agent 通信的问题。
3.2 核心架构
plaintext
┌─────────────────────────────────────────────────────────────┐
│ A2A 生态 │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Agent A │ │ Agent B │ │
│ │ (Client) │ │ (Server) │ │
│ │ │ HTTP/HTTPS │ │ │
│ │ 发现 ────────┼─── JSON-RPC 2.0 ──┼──→ Agent Card│ │
│ │ 委派 ────────┼─── Task ──────────┼──→ 执行 │ │
│ │ 监控 ────────┼─── SSE Stream ────┼──→ 状态更新 │ │
│ │ 获取 ────────┼─── Artifact ──────┼──→ 结果返回 │ │
│ │ │ │ │ │
│ └──────────────┘ └──────────────┘ │
│ │
│ 核心原则：灰盒解耦 │
│ ├─ 你能看到任务状态（透明性） │
│ └─ 但看不到内部实现（封装性） │
└─────────────────────────────────────────────────────────────┘
3.3 核心概念
Agent Card —— 智能体的"名片"
json
// .well-known/agent.json
{
"name": "Google Maps Agent",
"description": "Provides location and routing services",
"url": "https://maps-agent.example.com",
"skills": [
{
"name": "find_route",
"description": "Find optimal route between locations",
"inputModes": ["text", "json"],
"outputModes": ["json", "image"]
}
],
"authentication": {
"type": "oauth2",
"authorizationUrl": "https://auth.example.com/authorize"
},
"performance": {
"p95_latency_ms": 2000
}
}
Task —— 任务的生命周期
plaintext
submitted ──→ working ──→ completed
│ │
│ ├──→ failed
│ └──→ canceled
│
└──→ canceled
Artifact —— 任务的产出物
json
{
"taskId": "task-123",
"artifacts": [
{
"name": "route_map",
"type": "image/png",
"content": "base64_encoded_image_data"
},
{
"name": "route_description",
"type": "text/plain",
"content": "Take I-280 North for 12 miles..."
}
]
}
3.4 三种交互模式
python
# 模式1：同步 — 简单任务，立即返回
response = a2a_client.send_task(
agent_url="https://translate-agent.example.com",
message={"text": "Translate 'hello' to Chinese"}
)
# 直接获取结果
# 模式2：SSE 流式 — 长时间任务，流式获取进度
async for event in a2a_client.send_task_streaming(
agent_url="https://research-agent.example.com",
message={"text": "Research AI chip trends in 2026"}
):
if event.type == "task_update":
print(f"Progress: {event.status}")
elif event.type == "artifact":
print(f"Got artifact: {event.artifact.name}")
# 模式3：Webhook 异步 — 跨组织任务，回调通知
a2a_client.send_task_async(
agent_url="https://vendor-billing-agent.example.com",
message={"text": "Check invoice #12345"},
callback_url="https://my-agent.example.com/callbacks/billing"
)
3.5 A2A 的"灰盒"哲学
这是 A2A 最核心的设计思想：
表格
真实案例：某银行客服 Agent 通过 A2A 串联风控系统、工单系统，实现"咨询→风险评估→工单生成"闭环流程，任务流转效率提升 60%。
3.6 A2A 的局限
集中式管理瓶颈：依赖集中式管理节点，扩展性受限，万级 Agent 并发协作仍有挑战
状态同步延迟：灰盒模式下仍存在状态同步延迟，复杂流程中可能出现任务阻塞
生态仍早期：截至 2026 年中，生产环境的 A2A 实现不到 50 个，多数仍在 PoC 阶段
四、AG-UI 深度回顾：Agent-前端交互层
4.1 AG-UI 是什么
AG-UI（Agent Gateway User Interface Protocol）由 CopilotKit 团队推出，专注解决Agent 与前端应用之间的交互标准化问题。Google、Microsoft、AWS 已采用，月下载量约 300 万。
一句话概括：AG-UI 解决的是 Agent 怎么跟用户交互的问题。
4.2 为什么需要 AG-UI
传统模式下，Agent 运行时用户只能盯着 loading 转圈：
plaintext
用户体验（没有 AG-UI）：
用户: "帮我分析一下这个数据集"
⏳⏳⏳⏳⏳⏳⏳⏳⏳⏳ (5分钟后...)
Agent: "分析完成，结果是..." (用户已经去倒咖啡了)
有了 AG-UI：
plaintext
用户体验（有 AG-UI）：
用户: "帮我分析一下这个数据集"
Agent: 🔍 正在读取数据集... ← run_started
Agent: 📊 发现 3 个异常值... ← tool_call_start
Agent: 📈 生成可视化图表... ← state_delta
Agent: ✅ 分析完成！ ← run_finished
(用户全程可见，可以中途干预)
4.3 核心架构
plaintext
┌──────────────────────────────────────────────────┐
│ Frontend App │
│ ┌─────────────────────────────────────────┐ │
│ │ AG-UI Client (React/Angular) │ │
│ │ │ │
│ │ 监听事件流: │ │
│ │ ├── run_started │ │
│ │ ├── text_message_start │ │
│ │ ├── text_message_content ──→ 渲染UI │ │
│ │ ├── tool_call_start ──→ 显示进度 │ │
│ │ ├── state_delta ──→ 更新状态 │ │
│ │ └── run_finished │ │
│ └──────────────────┬──────────────────────┘ │
└─────────────────────┼────────────────────────────┘
│ SSE / WebSocket / Webhook
┌─────────────────────┼────────────────────────────┐
│ Agent Backend │
│ ┌──────────────────▼──────────────────────┐ │
│ │ AG-UI Server │ │
│ │ │ │
│ │ 发送事件流: │ │
│ │ ├── 生命周期事件 │ │
│ │ ├── 文本消息事件 │ │
│ │ ├── 工具调用事件 │ │
│ │ ├── 状态更新事件 │ │
│ │ └── 错误/取消事件 │ │
│ └─────────────────────────────────────────┘ │
└──────────────────────────────────────────────────┘
4.4 16 种标准化事件类型
AG-UI 定义了 16 种事件，覆盖五大场景：
plaintext
┌──────────────────────────────────────────────────────┐
│ AG-UI 事件类型全景 │
├──────────────────┬───────────────────────────────────┤
│ 生命周期事件 │ run_started │
│ │ run_finished │
├──────────────────┼───────────────────────────────────┤
│ 文本消息事件 │ text_message_start │
│ │ text_message_content │
│ │ text_message_end │
├──────────────────┼───────────────────────────────────┤
│ 工具调用事件 │ tool_call_start │
│ │ tool_call_args │
│ │ tool_call_end │
├──────────────────┼───────────────────────────────────┤
│ 状态更新事件 │ state_delta │
│ │ steps │
├──────────────────┼───────────────────────────────────┤
│ 错误处理事件 │ error │
│ │ cancellation │
├──────────────────┼───────────────────────────────────┤
│ 活动事件 │ activity │
│ (不落状态但 │ (用户可见的进度提示) │
│ 需要展示) │ │
└──────────────────┴───────────────────────────────────┘
4.5 三种传输方式
python
transport_options = {
"SSE": {
"desc": "Server-Sent Events，单向流式",
"best_for": "Agent 向前端推送进度",
"browser_support": "原生支持",
"complexity": "低"
},
"WebSocket": {
"desc": "双向实时通信",
"best_for": "需要用户中途干预的场景",
"browser_support": "原生支持",
"complexity": "中"
},
"Webhook": {
"desc": "HTTP 回调",
"best_for": "异步通知、跨系统集成",
"browser_support": "N/A（服务端）",
"complexity": "低"
}
}
4.6 关键特性：Human-in-the-Loop
AG-UI 的 "Interrupt" 机制是最重要的安全特性：
python
# 高风险操作时暂停，等用户确认
# Agent 端逻辑
async def delete_resource(resource_id: str):
# AG-UI 发送 interrupt 事件
# 前端弹窗: "确认删除资源 XXX？"
# 用户点击确认后，AG-UI 发回确认信号
# Agent 继续执行
pass
这确保 Agent 始终是"助手"而不是"失控的操作者"。
4.7 AG-UI vs A2UI
一个常见的困惑：AG-UI 和 Google 推出的 A2UI 是什么关系？
plaintext
AG-UI = 传输协议（管道）—— 负责 Agent 和前端之间的事件流传输
A2UI = 声明式 UI 规范（内容）—— 定义 Agent 生成的 UI 应该长什么样
关系：AG-UI 是管道，A2UI 是流过管道的内容之一
类比：AG-UI = HTTP，A2UI = HTML
五、三层协议栈架构
5.1 全景图
这是 2026 年 Agent 系统最核心的一张图：
plaintext
┌─────────────────────────────────────────────────────────────┐
│ 用户 (User) │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ AG-UI (交互层 / Interaction Layer) │ │
│ │ │ │
│ │ "Agent 怎么跟用户交互？" │ │
│ │ ├── 事件流: 16种标准化事件 │ │
│ │ ├── 传输: SSE / WebSocket / Webhook │ │
│ │ ├── 特性: Human-in-the-Loop, 双向同步 │ │
│ │ └── 产出: A2UI 组件 / 文本 / 状态更新 │ │
│ └──────────────────────────┬──────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ A2A (协作层 / Collaboration Layer) │ │
│ │ │ │
│ │ "Agent 怎么跟 Agent 通信？" │ │
│ │ ├── 发现: Agent Card (.well-known/agent.json) │ │
│ │ ├── 通信: HTTP + JSON-RPC 2.0 │ │
│ │ ├── 模式: 同步 / SSE流式 / Webhook异步 │ │
│ │ ├── 原则: 灰盒解耦 │ │
│ │ └── 产出: Task / Artifact / Message │ │
│ └──────────────────────────┬──────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ MCP (工具层 / Tool Layer) │ │
│ │ │ │
│ │ "Agent 怎么使用工具？" │ │
│ │ ├── 架构: Client-Server (JSON-RPC 2.0) │ │
│ │ ├── 传输: stdio / Streamable HTTP / gRPC │ │
│ │ ├── 暴露: Tools / Resources / Prompts │ │
│ │ └── 发现: 自动工具列表 │ │
│ └──────────────────────────┬──────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 外部世界 (External World) │ │
│ │ │ │
│ │ ├── 数据库 (PostgreSQL / MySQL / MongoDB) │ │
│ │ ├── API 服务 (GitHub / Notion / Slack) │ │
│ │ ├── 文件系统 (本地 / S3 / GCS) │ │
│ │ └── 自定义服务 (企业内部 API) │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
5.2 类比理解
如果把这三层比作互联网协议栈：
plaintext
互联网 Agent协议栈
───────── ────────────
HTML A2UI / AG-UI事件 ← 用户看到的
HTTP A2A ← 服务间通信
TCP/IP MCP ← 底层连接
5.3 递进式采用
你不需要一次性采用所有三层：
plaintext
阶段1: MCP 独立使用
├── 单 Agent + 工具集成
├── 80% 的使用场景
└── 如: Claude Desktop + Git MCP Server
阶段2: MCP + A2A
├── 多 Agent 协作
├── 跨团队/跨框架通信
└── 如: 研究 Agent + 编码 Agent + 部署 Agent
阶段3: MCP + A2A + AG-UI
├── 完整的 Agent 应用
├── 用户可见、可干预
└── 如: 企业级 AI 工作台
六、横向对比表
表格
七、选型决策树
plaintext
你的需求是什么？
│
├── Agent 需要调用外部工具/API/数据库？
│ └── 是 → MCP ✅
│ ├── 本地工具 → stdio 传输
│ ├── 远程服务 → Streamable HTTP
│ └── 低延迟场景 → gRPC
│
├── 多个 Agent 需要协作？
│ └── 是 → A2A ✅
│ ├── 同框架内部 → 先试框架原生机制
│ ├── 跨框架/跨团队 → A2A
│ └── 跨组织 → A2A + Webhook 异步
│
├── 需要用户看到 Agent 执行过程？
│ └── 是 → AG-UI ✅
│ ├── 简单聊天 → 不需要（文本流即可）
│ ├── 需要实时进度 → AG-UI SSE
│ ├── 需要用户中途干预 → AG-UI + Interrupt
│ └── 需要动态生成 UI → AG-UI + A2UI
│
└── 组合场景
├── 单 Agent + 工具 + 无 UI → MCP
├── 多 Agent + 工具 + 无 UI → MCP + A2A
├── 单 Agent + 工具 + 有 UI → MCP + AG-UI
└── 多 Agent + 工具 + 有 UI → MCP + A2A + AG-UI 🏆
八、组合实战：MCP + A2A + AG-UI 联合架构
8.1 场景：企业级 AI 工作台
假设我们要构建一个企业级 AI 工作台，功能是：用户输入一个商业问题，系统自动安排研究 Agent 调研、分析 Agent 分析、报告 Agent 生成报告，用户全程可见且可干预。
8.2 架构设计
plaintext
┌─────────────────────────────────────────────────────────────┐
│ 用户浏览器 │
│ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ AG-UI Client (React) │ │
│ │ │ │
│ │ ┌─────────┐ ┌──────────┐ ┌──────────────────┐ │ │
│ │ │ 进度面板 │ │ 中断确认 │ │ A2UI 动态组件 │ │ │
│ │ └─────────┘ └──────────┘ └──────────────────┘ │ │
│ └──────────────────────┬────────────────────────────┘ │
└─────────────────────────┼───────────────────────────────────┘
│ SSE / WebSocket
┌─────────────────────────┼───────────────────────────────────┐
│ 编排 Agent │
│ (Orchestrator) │
│ │ │
│ ┌────────────────┼────────────────┐ │
│ │ A2A │ A2A │ A2A │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ 研究 Agent │ │ 分析 Agent │ │ 报告 Agent │ │
│ │ │ │ │ │ │ │
│ │ MCP │ │ MCP │ │ MCP │ │
│ │ ├─搜索工具 │ │ ├─统计工具 │ │ ├─模板工具 │ │
│ │ ├─论文数据库│ │ ├─数据库 │ │ ├─图表工具 │ │
│ │ └─新闻API │ │ └─可视化 │ │ └─PDF生成 │ │
│ └────────────┘ └────────────┘ └────────────┘ │
│ │ │ │ │
│ └────────────────┼────────────────┘ │
│ │ │
│ ┌────────────────┼────────────────┐ │
│ │ MCP │ MCP │ MCP │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Web Search │ │ PostgreSQL │ │ S3 Storage │ │
│ │ Server │ │ Server │ │ Server │ │
│ └────────────┘ └────────────┘ └────────────┘ │
└─────────────────────────────────────────────────────────────┘
8.3 代码实现（关键部分）
python
"""
企业级 AI 工作台 - 三协议联合实现
MCP: 工具层
A2A: Agent 协作层
AG-UI: 前端交互层
"""
import asyncio
from typing import AsyncGenerator
from dataclasses import dataclass
from enum import Enum
# ============================================================
# MCP 层：工具集成
# ============================================================
class MCPServer:
"""简化的 MCP Server 实现"""
def __init__(self, name: str, tools: list[dict]):
self.name = name
self.tools = tools
async def list_tools(self) -> list[dict]:
"""返回可用工具列表"""
return self.tools
async def call_tool(self, tool_name: str, arguments: dict) -> dict:
"""调用指定工具"""
# 实际实现中，这里会执行真正的工具逻辑
return {"status": "ok", "tool": tool_name, "result": "..."}
# 创建 MCP Server 实例
web_search_server = MCPServer(
name="web-search",
tools=[
{
"name": "search",
"description": "Search the web for information",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"num_results": {"type": "integer", "default": 10}
},
"required": ["query"]
}
},
{
"name": "fetch_page",
"description": "Fetch and extract text from a web page",
"inputSchema": {
"type": "object",
"properties": {
"url": {"type": "string"}
},
"required": ["url"]
}
}
]
)
database_server = MCPServer(
name="analytics-database",
tools=[
{
"name": "query",
"description": "Execute SQL query on analytics database",
"inputSchema": {
"type": "object",
"properties": {
"sql": {"type": "string"},
"limit": {"type": "integer", "default": 100}
},
"required": ["sql"]
}
}
]
)
# ============================================================
# A2A 层：Agent 间协作
# ============================================================
@dataclass
class AgentCard:
"""A2A Agent Card"""
name: str
url: str
skills: list[dict]
description: str = ""
@dataclass
class A2ATask:
"""A2A Task"""
task_id: str
status: str # submitted, working, completed, failed
message: str
artifacts: list[dict] = None
def __post_init__(self):
if self.artifacts is None:
self.artifacts = []
class A2AClient:
"""简化的 A2A Client"""
async def discover_agent(self, agent_url: str) -> AgentCard:
"""发现远程 Agent"""
# 实际实现中，会请求 agent_url/.well-known/agent.json
return AgentCard(
name="research-agent",
url=agent_url,
skills=[{"name": "research", "description": "Research a topic"}]
)
async def send_task(
self,
agent_url: str,
message: str,
mode: str = "sync"
) -> A2ATask:
"""发送任务给远程 Agent"""
task = A2ATask(
task_id="task-001",
status="submitted",
message=message
)
# 实际实现中，这里会发送 HTTP 请求
task.status = "completed"
task.artifacts = [{"type": "text", "content": "Research results..."}]
return task
async def send_task_streaming(
self,
agent_url: str,
message: str
) -> AsyncGenerator[dict, None]:
"""流式发送任务，逐步获取状态更新"""
yield {"type": "task_update", "status": "working", "progress": 0.3}
yield {"type": "task_update", "status": "working", "progress": 0.6}
yield {"type": "artifact", "name": "research_report", "content": "..."}
yield {"type": "task_update", "status": "completed", "progress": 1.0}
# ============================================================
# AG-UI 层：前端交互
# ============================================================
class AGUIEventType(Enum):
RUN_STARTED = "run_started"
RUN_FINISHED = "run_finished"
TEXT_MESSAGE_START = "text_message_start"
TEXT_MESSAGE_CONTENT = "text_message_content"
TEXT_MESSAGE_END = "text_message_end"
TOOL_CALL_START = "tool_call_start"
TOOL_CALL_END = "tool_call_end"
STATE_DELTA = "state_delta"
ERROR = "error"
CANCELLATION = "cancellation"
@dataclass
class AGUIEvent:
"""AG-UI 事件"""
type: AGUIEventType
data: dict
class AGUIServer:
"""简化的 AG-UI Server"""
async def stream_events(
self,
run_id: str,
user_input: str
) -> AsyncGenerator[AGUIEvent, None]:
"""生成 AG-UI 事件流"""
# 1. 开始运行
yield AGUIEvent(
type=AGUIEventType.RUN_STARTED,
data={"run_id": run_id, "thread_id": "thread-001"}
)
# 2. 发送文本开始
yield AGUIEvent(
type=AGUIEventType.TEXT_MESSAGE_START,
data={"role": "assistant", "message_id": "msg-001"}
)
# 3. 工具调用
yield AGUIEvent(
type=AGUIEventType.TOOL_CALL_START,
data={"tool_name": "search", "call_id": "call-001"}
)
# 4. 状态更新
yield AGUIEvent(
type=AGUIEventType.STATE_DELTA,
data={"state": "researching", "progress": 0.3}
)
# 5. 工具调用完成
yield AGUIEvent(
type=AGUIEventType.TOOL_CALL_END,
data={"call_id": "call-001", "result": "Search completed"}
)
# 6. 文本内容
yield AGUIEvent(
type=AGUIEventType.TEXT_MESSAGE_CONTENT,
data={"content": "Based on the research, here is the analysis..."}
)
# 7. 文本结束
yield AGUIEvent(
type=AGUIEventType.TEXT_MESSAGE_END,
data={"message_id": "msg-001"}
)
# 8. 运行完成
yield AGUIEvent(
type=AGUIEventType.RUN_FINISHED,
data={"run_id": run_id, "status": "success"}
)
# ============================================================
# 三层联合：完整编排
# ============================================================
class EnterpriseAIWorkbench:
"""企业级 AI 工作台 - 三协议联合"""
def __init__(self):
# MCP 层
self.mcp_servers = {
"web_search": web_search_server,
"database": database_server
}
# A2A 层
self.a2a_client = A2AClient()
self.agent_cards = {}
async def setup(self):
"""初始化：发现所有协作 Agent"""
agents = [
"https://research-agent.internal:9000",
"https://analysis-agent.internal:9000",
"https://report-agent.internal:9000"
]
for url in agents:
card = await self.a2a_client.discover_agent(url)
self.agent_cards[card.name] = card
async def execute(
self,
user_query: str
) -> AsyncGenerator[AGUIEvent, None]:
"""执行用户查询，三层协议联动"""
agui = AGUIServer()
run_id = "run-" + str(hash(user_query))[:8]
# AG-UI: 发送开始事件
yield AGUIEvent(
type=AGUIEventType.RUN_STARTED,
data={"run_id": run_id}
)
# A2A Step 1: 委派研究任务
yield AGUIEvent(
type=AGUIEventType.TOOL_CALL_START,
data={"tool_name": "delegate_research"}
)
research_task = await self.a2a_client.send_task(
agent_url="https://research-agent.internal:9000",
message=user_query
)
yield AGUIEvent(
type=AGUIEventType.TOOL_CALL_END,
data={"tool_name": "delegate_research", "result": "completed"}
)
# A2A Step 2: 委派分析任务
yield AGUIEvent(
type=AGUIEventType.STATE_DELTA,
data={"state": "analyzing", "progress": 0.5}
)
analysis_task = await self.a2a_client.send_task(
agent_url="https://analysis-agent.internal:9000",
message=f"Analyze the following research: {research_task.artifacts}"
)
# A2A Step 3: 委派报告任务
yield AGUIEvent(
type=AGUIEventType.STATE_DELTA,
data={"state": "generating_report", "progress": 0.8}
)
report_task = await self.a2a_client.send_task(
agent_url="https://report-agent.internal:9000",
message=f"Generate report from analysis: {analysis_task.artifacts}"
)
# AG-UI: 发送最终结果
yield AGUIEvent(
type=AGUIEventType.TEXT_MESSAGE_START,
data={"role": "assistant"}
)
yield AGUIEvent(
type=AGUIEventType.TEXT_MESSAGE_CONTENT,
data={"content": f"Report generated successfully!"}
)
yield AGUIEvent(
type=AGUIEventType.TEXT_MESSAGE_END,
data={}
)
yield AGUIEvent(
type=AGUIEventType.RUN_FINISHED,
data={"run_id": run_id, "status": "success"}
)
# ============================================================
# 运行示例
# ============================================================
async def main():
workbench = EnterpriseAIWorkbench()
await workbench.setup()
async for event in workbench.execute(
"Analyze the competitive landscape of AI agent frameworks in 2026"
):
print(f"[{event.type.value}] {event.data}")
if __name__ == "__main__":
asyncio.run(main())
九、未来趋势：协议收敛还是持续分裂？
9.1 当前格局
plaintext
协议生态 2026:
├── MCP ──── 已成事实标准，Anthropic/OpenAI/Google 共同支持
├── A2A ──── Google 主导，捐赠 Linux 基金会，50+ 合作伙伴
├── AG-UI ── CopilotKit 主导，3M 月下载，AWS/MS/Google 采用
├── ANP ──── 去中心化协议，P2P 发现，小众但有潜力
└── AGNTCY ── Cisco/SAP 主导，企业级，量子上加密预留
9.2 两种预测
乐观派：协议收敛
plaintext
2026 现状 2027 预测 2028 预测
───────── ────────── ──────────
MCP ───┐ MCP ───┐ 统一
A2A ───┼──→ A2A ───┤──→ Agent
AG-UI ─┘ AG-UI ─┘ Protocol
(边界模糊,
互操作增强)
理由：
MCP 已被三大 AI 巨头共同采用
A2A 捐赠 Linux 基金会，走向中立
AG-UI 的事件模型可以作为其他协议的传输层
悲观派：持续分裂
plaintext
2026 现状 2027 预测 2028 预测
───────── ────────── ──────────
MCP ───┐ MCP v2 ──┐ MCP v3
A2A ───┤──→ A2A v1 ───┤──→ A2A v2
AG-UI ─┘ AG-UI v2 ─┘ AG-UI v3
+ 新协议? + 更多新协议?
理由：
每个大厂都有推出自己协议的冲动
ANP、AGNTCY 等新协议仍在涌现
企业级需求催生私有协议变体
9.3 我的判断
短期（2026-2027）：三层互补格局不会变，但边界会模糊：
MCP 可能吸收部分 A2A 的发现机制
A2A 可能增加更丰富的状态同步
AG-UI 可能内置更多 A2UI 组件
长期（2028+）：会收敛到一个统一栈，就像 TCP/IP/HTTP 最终统一了网络通信一样。但这个过程不会一蹴而就——先有事实标准，再有正式标准。
十、踩坑记录
坑1：MCP Server 版本升级导致兼容性故障
plaintext
场景: AI 绘画工具 MCP Server 升级 API v2，参数名从 prompt 改为 description
影响: 15% 的调用方（使用旧 schema 的 Agent）请求失败
修复: 在 MCP Server 端做参数兼容层，旧参数名映射到新参数名
教训: MCP Server 的 schema 变更是 breaking change，必须做版本管理
坑2：A2A 灰盒模式下的状态同步延迟
plaintext
场景: 银行风控 Agent 需要实时获取客服 Agent 的任务状态
问题: A2A 的 SSE 流式更新有 200-500ms 延迟，风控决策滞后
修复: 对于实时性要求高的场景，改用 WebSocket 替代 SSE
教训: 灰盒不等于"慢盒"，选对传输方式很重要
坑3：AG-UI 事件风暴
plaintext
场景: 复杂 Agent 工作流同时触发多个工具调用
问题: 前端收到密集事件流（每秒数百个），UI 卡顿
修复: 在 AG-UI Client 端做事件节流（throttle），合并同类事件
教训: 不是所有事件都需要实时渲染，区分"必须渲染"和"可以合并"
坑4：三层协议的认证割裂
plaintext
场景: MCP 用 OAuth2，A2A 用 OpenAPI 认证，AG-UI 用委托链
问题: 一个请求穿过三层，需要三种认证，凭据管理混乱
修复: 统一使用 JWT，MCP/A2A/AG-UI 各层验证同一个 token
教训: 在架构设计阶段就要统一认证方案，不要各搞各的
坑5：MCP 和 A2A 的职责混淆
plaintext
场景: 团队试图用 MCP 实现 Agent 间通信
问题: MCP 是 Client-Server 紧耦合模型，不支持任务委派和灰盒
修复: Agent 间协作改用 A2A，MCP 只负责工具调用
教训: 三个协议各管一层，不要混用。MCP 管"用工具"，A2A 管"找人"
坑6：AG-UI 的"打字机效果"性能问题
plaintext
场景: text_message_content 事件逐 token 发送
问题: 每秒 60+ 个 DOM 更新，React 渲染跟不上
修复: 使用 requestAnimationFrame 批量更新，或虚拟化长文本
教训: AG-UI 事件流 ≠ 直接 DOM 操作，中间需要一层渲染优化
总结
表格
我的建议：不要被协议焦虑绑架。从 MCP 开始，等你有多个 Agent 需要协作时再加 A2A，等你有前端需要实时交互时再加 AG-UI。三个协议是递进关系，不是并列选择。
本文由 PySuper 撰写，首发于 zhengxingtao.com
参考来源：
据《The Agent Protocol Stack: MCP vs. A2A vs. AG-UI》(DZone, 2026-05-15)
据《MCP, A2A, and AG-UI: Which Agent Protocol Do You Need?》(SoftwareCurated, 2026-05-21)
据《MCP vs A2A: Agent Protocols Compared》(TokenMix, 2026-04-25)
据《Generative UI Spectrum: How Agents Now Ship Their Own Interfaces》(CopilotKit, 2026-04-23)
评论区