从 MCP 到 A2A 再到 AG-UI：AI Agent 协议栈三层架构全解析——工具层、协作层与交互层的技术真相与工程实践

引言：当 Agent 从单兵走向军团

2026 年的 AI Agent 战场，剧本已经悄然改写。

去年这个时候，所有团队都在做同一件事：让单个 Agent 变得更强——更长的上下文、更好的推理能力、更丰富的工具调用。市场上大部分 Agent 产品依然是"单兵作战"的状态：一个 Agent，配一堆工具，靠 prompt 工程驱动。

但从 2026 年 Q2 开始，拐点出现了。

单 Agent 的能力天花板清晰可见：代码 Agent 能写代码，但不懂最新 API；搜索 Agent 擅长找资料，但写出来的报告上下文断裂；多模态 Agent 能看图，但推理链太浅。没有人能做到全知全能。

于是，Multi-Agent System（多智能体系统）成了新的主战场。多个专业 Agent 配合干活，才是正解。

但问题随之而来：Agent 之间怎么通信？

在没有标准之前，每个团队都在造自己的轮子：用 HTTP 硬写的、用消息队列的、用 WebSocket 的，接口五花八门，参数格式各写各的。Agent A 给 Agent B 传数据，得先写个适配层。跨框架、跨供应商的协作更是奢望。

这就是协议层基础设施的缺失带来的代价。

从 2024 年底到 2026 年，三家巨头分别交出了自己的答卷：

• Anthropic 推出了 MCP（Model Context Protocol）——解决 Agent 与工具/数据的连接问题
• Google 推出了 A2A（Agent-to-Agent）——解决 Agent 与 Agent 之间的协作问题
• CopilotKit 团队 发起了 AG-UI（Agent–User Interaction Protocol）——解决 Agent 与前端应用的交互问题

这三个协议并非竞争关系，而是互补共存，共同构成 AI Agent 时代的三层协议栈：

┌─────────────────────────────────────────────────────────────┐
│                 AI Agent Protocol Stack                      │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌─────────────────────────────────────────────────────┐  │
│   │  Layer 3: AG-UI    Agent ↔ User (Frontend)          │  │
│   │  CopilotKit  ·  Streaming · Generative UI · Events   │  │
│   └─────────────────────────────────────────────────────┘  │
│                            ↕                                 │
│   ┌─────────────────────────────────────────────────────┐  │
│   │  Layer 2: A2A     Agent ↔ Agent (Collaboration)      │  │
│   │  Google  ·  Agent Card · Task Push · JSON-RPC 2.0     │  │
│   └─────────────────────────────────────────────────────┘  │
│                            ↕                                 │
│   ┌─────────────────────────────────────────────────────┐  │
│   │  Layer 1: MCP     Agent ↔ Tools & Data              │  │
│   │  Anthropic ·  Tools/Resources/Prompts · JSON-RPC 2.0 │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

本文将从第一性原理出发，深度拆解这三层协议的技术内核、架构设计、核心概念，并通过完整代码示例展示如何从零构建一套完整的 Agent 协议栈系统。无论你是正在构建 Multi-Agent 系统的后端工程师，还是想让 Agent 与前端应用深度集成的全栈开发者，这篇文章都能给你真正可落地的工程指导。

第一部分：MCP（Model Context Protocol）——Agent 的万能插头

1.1 背景与问题

AI 模型再强大，如果不能访问外部世界，也只是空中楼阁。

传统的工具调用模式是这样的：开发者为每个工具写一个硬编码的 API 适配，prompt 里塞一段描述，然后靠模型的 function calling 能力触发。这种方式的问题在于：

1. 每换一个工具就要重写适配代码，无法复用
2. 工具描述靠手写，没有标准化格式，模型理解质量参差不齐
3. 安全边界模糊：工具的权限、认证、数据隔离，全靠开发者的"约定"
4. 多工具协作缺乏标准：工具 A 的输出怎么传给工具 B？没有统一路径

MCP 正是为了解决这些问题而生的。

1.2 架构解析

MCP 采用经典的客户端-服务器架构，基于 JSON-RPC 2.0 协议通信。核心角色有三个：

┌─────────────────────────────────────────────────────────────┐
│                        MCP Host                              │
│  (Claude Desktop / Cursor / 你的 AI 应用)                    │
│                                                             │
│    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐   │
│    │ MCP Client  │    │ MCP Client  │    │ MCP Client  │   │
│    │ (FS Server) │    │ (DB Server) │    │ (API Server)│   │
│    └──────┬──────┘    └──────┬──────┘    └──────┬──────┘   │
└───────────┼──────────────────┼──────────────────┼───────────┘
            │                  │                  │
     ┌──────▼──────┐    ┌──────▼──────┐    ┌──────▼──────┐
     │ File System  │    │  Database   │    │  REST API   │
     │   Server     │    │   Server    │    │   Server    │
     └─────────────┘    └─────────────┘    └─────────────┘

Host（主机）：你的 AI 应用本体，负责管理多个 MCP Client 与外部 Server 的连接。Host 是信任边界的边界——它决定允许哪些 Server 接入。

Client（客户端）：Host 内部的 MCP 通信代理，每个 Server 对应一个 Client。Client 负责与对应的 Server 维持长连接，处理请求/响应。

Server（服务器）：向外暴露能力（文件系统、数据库、API 等）。Server 是无状态的，通过标准化的接口描述自己的能力。

1.3 三大核心能力

MCP 定义了四个核心原语，用于向模型暴露外部能力：

1.3.1 Resources（资源）——给 AI 读数据

Resources 是 AI 可以读取的数据源。Server 通过 URI 模板声明自己提供哪些资源，Client 负责缓存和管理。

// Server 暴露的 resources 声明
{
  "resources": [
    {
      "uriTemplate": "file:///{path}",
      "name": "Local File System",
      "description": "Read files from the local filesystem",
      "mimeType": "text/plain"
    },
    {
      "uriTemplate": "db://{table}",
      "name": "PostgreSQL Database",
      "description": "Query data from PostgreSQL tables",
      "mimeType": "application/json"
    }
  ]
}

模型通过 resources/read 工具读取资源内容——就像 AI 获得了一个标准化的文件系统和数据库读取接口。

1.3.2 Tools（工具）——给 AI 动手操作

Tools 是 AI 可以执行的操作。每个 Tool 通过标准化的 schema 描述输入参数，模型通过 function calling 触发。

// 一个典型的 MCP Tool 声明
{
  "name": "send_email",
  "description": "Send an email via the configured SMTP server",
  "inputSchema": {
    "type": "object",
    "properties": {
      "to": {
        "type": "string",
        "description": "Recipient email address"
      },
      "subject": {
        "type": "string",
        "description": "Email subject line"
      },
      "body": {
        "type": "string",
        "description": "Email body content (plain text or markdown)"
      }
    },
    "required": ["to", "subject", "body"]
  }
}

这种标准化 schema 的意义极为重大：有了它，Claude/Cursor 这类 IDE 可以在用户编写代码时，自动弹出正确补全的 Tool 调用建议——就像 IDE 为编程语言的标准库提供 IntelliSense 一样，MCP 为 AI 的工具调用提供了 AI -native 的 IntelliSense。

1.3.3 Prompts（提示模板）——给 AI 预置工作流

Prompts 允许 Server 暴露预定义的提示模板，减少每次调用的 token 消耗：

{
  "prompts": [
    {
      "name": "code_review",
      "description": "Standardized code review prompt with checklist",
      "arguments": [
        {
          "name": "language",
          "description": "Programming language of the code"
        },
        {
          "name": "code",
          "description": "The code to review"
        }
      ]
    }
  ]
}

1.3.4 Sampling（采样）——让 Server 反向调用模型

这是 MCP 1.0 到 1.x 版本中相对较少被讨论但极具潜力的能力：Server 可以请求 Host 代表的模型进行推理，创建一个双向的通信通道。

# MCP Sampling 示例（Server 端）
{
  "method": "sampling/createMessage",
  "params": {
    "systemPrompt": "You are a security analyst. Analyze the following code for vulnerabilities.",
    "messages": [{"role": "user", "content": "Check: " + userCode}],
    "temperature": 0.3,
    "maxTokens": 1024
  }
}

1.4 完整 MCP Server 实现

下面我们用 Python 实现一个完整的、生产级的 MCP Server，暴露文件系统搜索和 GitHub API 两个工具：

# mcp_file_search_server.py
import json
import subprocess
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("file-search-server")

# ─── 工具定义 ───
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_code",
            description="Search for code patterns in files using grep",
            inputSchema={
                "type": "object",
                "properties": {
                    "pattern": {
                        "type": "string",
                        "description": "Regex pattern to search for"
                    },
                    "path": {
                        "type": "string",
                        "description": "Directory path to search in (default: current directory)"
                    },
                    "file_type": {
                        "type": "string",
                        "description": "File extension filter, e.g. 'py' or 'go'"
                    }
                },
                "required": ["pattern"]
            }
        ),
        Tool(
            name="git_log",
            description="Get recent git commits with statistics",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Repository path"
                    },
                    "count": {
                        "type": "integer",
                        "description": "Number of commits to retrieve (default: 10)"
                    }
                }
            }
        ),
        Tool(
            name="github_pr_list",
            description="List open pull requests from a GitHub repository",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "state": {
                        "type": "string",
                        "enum": ["open", "closed", "all"],
                        "default": "open"
                    }
                },
                "required": ["owner", "repo"]
            }
        )
    ]

# ─── 工具执行 ───
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    if name == "search_code":
        return await _search_code(arguments)
    elif name == "git_log":
        return await _git_log(arguments)
    elif name == "github_pr_list":
        return await _github_pr_list(arguments)
    else:
        raise ValueError(f"Unknown tool: {name}")

async def _search_code(args: dict) -> list[TextContent]:
    path = args.get("path", ".")
    pattern = args["pattern"]
    file_type = args.get("file_type")

    cmd = ["grep", "-rn", "--color=never", pattern, path]
    if file_type:
        cmd.insert(1, f"--include=*.{file_type}")

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    output = result.stdout if result.stdout else "(no matches found)"
    return [TextContent(type="text", text=output)]

async def _git_log(args: dict) -> list[TextContent]:
    path = args.get("path", ".")
    count = args.get("count", 10)

    cmd = ["git", "-C", path, "log", f"-{count}", "--stat", "--pretty=format:%h %s (%an)"]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
    return [TextContent(type="text", text=result.stdout or "(not a git repo)")]

async def _github_pr_list(args: dict) -> list[TextContent]:
    import os
    token = os.environ.get("GITHUB_TOKEN")
    owner, repo, state = args["owner"], args["repo"], args.get("state", "open")

    cmd = ["gh", "pr", "list",
           "--repo", f"{owner}/{repo}",
           "--state", state,
           "--json", "number,title,author,url,createdAt",
           "-q", "."]

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    return [TextContent(type="text", text=result.stdout or "[]")]

# ─── 启动 Server ───
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

这个 Server 通过 stdio 通信，可以被任何 MCP Host（如 Claude Desktop、Cursor）连接使用。核心在于 stdio_server() ——它将 MCP 协议封装为标准输入/输出流，非常适合本地工具集成。

1.5 MCP 在协议栈中的定位

MCP 位于协议栈的最底层，解决的是"Agent 如何看见和操作外部世界"的问题。它的价值在于：

• 标准化：所有工具和资源用同一套描述格式，Host 只需实现一次 Client
• 安全隔离：Host 控制信任边界，Server 无法主动访问 Host 的内部状态
• 热插拔：可以动态注册/注销 Server，Agent 在运行时发现自己有哪些可用工具

第二部分：A2A（Agent-to-Agent）——Agent 军团的通信标准

2.1 背景与问题

MCP 让单个 Agent 拥有了工具，但当多个 Agent 需要协同工作时，新的问题出现了：

• Agent A 如何知道 Agent B 的存在？ 需要手动配置地址吗？
• Agent A 交给 Agent B 的任务如何追踪？ 是同步阻塞还是异步？
• Agent B 在处理过程中产生的中间状态如何实时推送给 Agent A？ 长任务跑了 5 分钟，Agent A 怎么知道进展？
• Agent B 处理失败怎么办？ 如何优雅降级？

在没有 A2A 之前，每个团队都用私有方案解决这些问题：HTTP 回调、WebSocket 推送、消息队列……接口不统一，跨框架协作等于天方夜谭。

Google 在 2026 年 4 月正式发布 A2A 协议，第一次为 Agent 间通信建立了工业级标准。

2.2 核心概念

A2A 的设计哲学是让 Agent 之间的协作变得像 HTTP API 一样简单，同时支持复杂的异步长任务。核心概念包括：

2.2.1 Agent Card（Agent 卡片）

每个兼容 A2A 的 Agent 必须暴露一个标准的机器可读配置文件，通过 /.well-known/agent-card.json 提供：

{
  "name": "code-review-agent",
  "version": "1.0.0",
  "description": "Specialized in code quality analysis and security scanning",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitions": true
  },
  "skills": [
    {
      "id": "security-scan",
      "name": "Security Vulnerability Scanner",
      "description": "Scans code for OWASP Top 10 vulnerabilities",
      "tags": ["security", "vulnerability", "owasp"],
      "inputModes": ["text", "code"],
      "outputModes": ["text", "json"]
    },
    {
      "id": "code-review",
      "name": "Code Quality Reviewer",
      "description": "Reviews code quality, style, and maintainability",
      "tags": ["quality", "style", "best-practices"],
      "inputModes": ["text", "code"],
      "outputModes": ["text"]
    }
  ],
  "authentication": {
    "schemes": ["bearer"],
    "credentials": "required"
  },
  "defaultApiVersion": "0.9.0"
}

Agent Card 就是 Agent 的"简历"——它让其他 Agent 可以发现自己、了解自己的能力、选择合适的 Agent 来协作，而不需要在代码里硬编码对方的地址。

2.2.2 Task（任务）与状态机

A2A 将跨 Agent 的协作建模为一个有生命周期状态机：

                    ┌──────────┐
                    │ submitted │
                    └────┬─────┘
                         │
            ┌────────────▼────────────┐
            │       working          │
            │  (推送中间状态/进度)     │
            └────────────┬────────────┘
                         │
         ┌───────────────┼───────────────┐
         │               │               │
    ┌────▼────┐   ┌─────▼─────┐  ┌──────▼──────┐
    │completed│   │  failed   │  │ input-required│
    └─────────┘   └───────────┘  └──────────────┘
                                              ↑ 
                                    (等待外部输入后继续)

• submitted：任务已提交，等待处理
• working：正在处理，可能通过 Server-Sent Events（SSE）推送中间状态
• completed：成功完成，结果已返回
• failed：处理失败，包含错误信息
• input-required：需要外部输入（如人工审批），任务暂停等待

这种状态机设计让长任务的进度追踪变得简单而可靠。每个状态变化都可以通过 WebSocket/SSE 推送给发起方，实现实时协作。

2.2.3 JSON-RPC 2.0 通信

A2A 底层采用 JSON-RPC 2.0，有两类核心消息：

tasks/send（发起任务）：

// 请求
{
  "jsonrpc": "2.0",
  "method": "tasks/send",
  "params": {
    "id": "task-uuid-123",
    "sessionId": "session-456",
    "message": {
      "role": "user",
      "parts": [
        {
          "type": "text",
          "text": "Please review the authentication module in PR #42"
        },
        {
          "type": "data",
          "data": {
            "prNumber": 42,
            "repo": "acme/backend",
            "language": "go"
          }
        }
      ]
    },
    "skillId": "code-review"
  },
  "id": 1
}

// 响应
{
  "jsonrpc": "2.0",
  "result": {
    "taskId": "task-uuid-123",
    "status": {"state": "working"},
    "artifacts": []
  },
  "id": 1
}

tasks/sendSubscribe（订阅任务，启用 SSE）：

# 发起一个需要长时间处理的任务，并订阅进度更新
async def review_pr_with_streaming():
    async with a2a_client.session("https://review-agent.company.com") as session:
        task = await session.send_subscribe({
            "skillId": "security-scan",
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": "Scan repo acme/backend for vulnerabilities"}]
            }
        })

        # 实时消费中间状态推送
        async for event in task.stream():
            if event.kind == "status_update":
                print(f"Progress: {event.status}% — {event.message}")
            elif event.kind == "artifact":
                print(f"New artifact: {event.artifact.name}")
            elif event.kind == "error":
                print(f"Error: {event.error}")

2.3 A2A vs MCP：边界在哪里？

这是最关键的理解点：MCP 和 A2A 解决的是完全不同层面的问题。

一个具体例子：

你有一个"代码审查 Agent"（Reviewer）和一个"安全扫描 Agent"（Scanner）。

• Reviewer 通过 MCP 调用 GitHub API 获取 PR 详情（Agent → 工具）
• Reviewer 通过 A2A 将代码片段发送给 Scanner 进行安全扫描（Agent → Agent）
• Scanner 通过 A2A 的 SSE 实时推送扫描进度给 Reviewer
• Scanner 通过 MCP 调用漏洞数据库 API 获取 CVE 详情（Agent → 工具）

两者各司其职，互不越界。

2.4 Supervisor 模式：A2A 的生产级编排

在真实的 Multi-Agent 系统中，通常需要一个 Supervisor（监督者） 来编排多个 Worker Agent：

# supervisor_orchestrator.py
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import a2a

class AgentRole(Enum):
    RESEARCHER = "researcher-agent.company.internal"
    CODER = "coder-agent.company.internal"
    REVIEWER = "reviewer-agent.company.internal"

@dataclass
class TaskContext:
    user_request: str
    pr_number: int
    repo: str
    research_data: Optional[dict] = None
    code_changes: Optional[str] = None
    review_result: Optional[dict] = None

class Supervisor:
    def __init__(self):
        self.a2a_client = a2a.Client()

    async def handle_pr_task(self, context: TaskContext) -> dict:
        """
        典型的三阶段编排流程：
        1. Researcher 收集信息
        2. Coder 生成修改
        3. Reviewer 审核结果
        """
        print(f"📋 Starting PR #{context.pr_number} workflow")

        # ── Stage 1: Research ──
        print("🔍 [Stage 1/3] Dispatching to Researcher Agent...")
        research_task = await self.a2a_client.send(
            url=AgentRole.RESEARCHER.value,
            skill_id="pr-research",
            message=f"Gather context for PR #{context.pr_number} in {context.repo}",
            params={"pr_number": context.pr_number, "repo": context.repo}
        )
        research_result = await research_task.wait_for_completion(timeout=60)
        context.research_data = research_result.data
        print(f"✅ Research complete: {len(context.research_data.get('commits', []))} commits analyzed")

        # ── Stage 2: Code generation ──
        print("💻 [Stage 2/3] Dispatching to Coder Agent...")
        code_task = await self.a2a_client.send(
            url=AgentRole.CODER.value,
            skill_id="pr-fix-generator",
            message=f"Based on research findings, apply fixes for PR #{context.pr_number}",
            params={
                "pr_number": context.pr_number,
                "research": context.research_data,
                "repo": context.repo
            }
        )

        # 支持流式接收中间产物
        async for update in code_task.stream():
            if update.event == "artifact_generated":
                print(f"  📄 Generated: {update.filename}")

        code_result = await code_task.wait_for_completion(timeout=300)
        context.code_changes = code_result.artifacts[0].content
        print(f"✅ Code changes ready: {len(context.code_changes)} chars")

        # ── Stage 3: Review ──
        print("🔎 [Stage 3/3] Dispatching to Reviewer Agent...")
        review_task = await self.a2a_client.send(
            url=AgentRole.REVIEWER.value,
            skill_id="code-review",
            message=f"Review the following changes for PR #{context.pr_number}",
            params={"code": context.code_changes, "context": context.research_data}
        )

        review_result = await review_task.wait_for_completion(timeout=120)
        context.review_result = review_result.data
        print(f"✅ Review complete: {context.review_result.get('score', 'N/A')}/10 quality score")

        # ── Final: Aggregate & respond ──
        return {
            "summary": f"PR #{context.pr_number} processed successfully",
            "research": context.research_data,
            "changes": context.code_changes,
            "review": context.review_result,
            "recommendation": context.review_result.get("decision", "needs_human_review")
        }

async def main():
    supervisor = Supervisor()
    result = await supervisor.handle_pr_task(
        TaskContext(
            user_request="Fix authentication bug in PR #42",
            pr_number=42,
            repo="acme/backend"
        )
    )
    print(f"\n🎉 Final result: {result['recommendation']}")

if __name__ == "__main__":
    asyncio.run(main())

这就是典型的 Supervisor Pattern：一个编排者负责任务分解、子 Agent 调度、结果聚合，对外呈现为单一 Agent 接口。这种模式在 LangGraph 的 Supervisor 节点、Google ADK 的 Agent Group 等框架中广泛使用。

第三部分：AG-UI——让 Agent 真正走进用户界面

3.1 背景与问题

MCP 连接 Agent 与工具，A2A 连接 Agent 与 Agent。但还有一环始终没有标准：Agent 如何与前端应用交互？

传统模式下，前端通过 REST API 调用后端，后端再调用 Agent API。这种方式有几个根本性问题：

1. 无法流式：LLM 输出是逐 token 生成的，传统 HTTP 请求/响应模型只能等完全生成后才能返回，用户体验极差
2. 状态同步困难：Agent 在思考过程中的中间状态、工具调用结果，无法实时推送给前端
3. 工具调用黑盒：前端不知道 Agent 正在调用哪个工具、参数是什么、结果如何
4. UI 适配混乱：每个 AI 应用都用自己的方式处理 Agent 输出，组件无法复用

AG-UI（Agent–User Interaction Protocol）正是为解决这些问题而生的。

3.2 核心定位

AG-UI 由 CopilotKit 团队发起，是一个开放的、基于事件的、双向的协议，用于标准化 AI Agent 与用户界面之间的实时交互。

它与 A2A 的区别非常清晰：

• A2A：Agent ↔ Agent（协作层）
• AG-UI：Agent ↔ User（交互层）

两者可以叠加使用：Agent 通过 A2A 协作，通过 AG-UI 向用户展示进展和结果。

3.3 四大构建模块

根据 AG-UI 官方文档（docs.ag-ui.com），该协议定义了四个核心构建块：

3.3.1 Streaming Chat（流式聊天）

AG-UI 支持逐 token 的流式输出，前端实时渲染 Agent 的思考过程：

// 前端订阅 Agent 的流式响应
import { useAgentStream } from '@copilotkit/react-core';

function ChatInterface() {
  const { messages, submit, isLoading } = useAgentStream({
    agentUrl: "https://api.company.com/agent/assistant",
    agentToken: userAuthToken,
  });

  return (
    <div className="chat-container">
      {messages.map((msg, i) => (
        <MessageBubble
          key={i}
          role={msg.role}
          content={msg.content}
          // Agent 正在思考时显示加载指示器
          isGenerating={isLoading && i === messages.length - 1 && msg.role === "assistant"}
        />
      ))}

      <input
        onKeyDown={(e) => e.key === "Enter" && submit(e.currentTarget.value)}
        disabled={isLoading}
      />
    </div>
  );
}

3.3.2 Multimodality（多模态）

支持文件、图片、音频、视频等附件的实时传输：

// 上传图片并获取 AI 分析
const result = await agent.send({
  type: "multimodal_input",
  attachments: [
    {
      type: "image",
      url: "file:///tmp/screenshot.png",
      mimeType: "image/png",
    }
  ],
  message: "Analyze this screenshot and identify any UI issues"
});

// AI 可以实时返回带标注的图像
result.on("annotation", (annotatedImage) => {
  showAnnotationOverlay(annotatedImage);
});

3.3.3 Generative UI（生成式 UI）

这是 AG-UI 最具想象力的能力：Agent 可以动态生成 UI 组件，前端负责渲染：

// Agent 返回了一个动态生成的图表组件
result.on("generative_ui", (component) => {
  if (component.type === "chart") {
    renderChart(component.spec);  // Agent 定义了图表的 spec，前端负责渲染
  } else if (component.type === "form") {
    renderDynamicForm(component.fields);  // Agent 生成了表单字段
  }
});

前端控制渲染，意味着 Agent 的"想法"不会破坏应用的设计系统——Agent 提供数据 spec，前端决定如何展示。这解决了"LLM 生成 HTML/CSS 难以融入真实产品"的问题。

3.3.4 State Sync（状态同步）

Agent 和前端共享会话状态，支持取消和恢复：

// 用户取消正在进行的 Agent 任务
agent.cancel();

// 恢复之前的会话状态
agent.restore({
  sessionId: "session-123",
  lastMessageIndex: 15,
  toolCallStack: ["search_api", "db_query"]
});

3.4 AG-UI Server 实现

下面实现一个兼容 AG-UI 的 Python 后端，使用 FastAPI + SSE：

# ag_ui_server.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse
import asyncio
import json
import uuid
from typing import AsyncGenerator

app = FastAPI(title="AG-UI Agent Server")

# ─── Agent 实现（简化版） ───
class StreamingAgent:
    def __init__(self):
        self.sessions: dict[str, dict] = {}

    async def process_stream(
        self,
        session_id: str,
        message: str,
        on_event: callable
    ) -> AsyncGenerator[dict, None]:
        """
        模拟流式 Agent 处理过程，产生各种类型的 AG-UI 事件
        """
        # 1. 发送 thinking 事件
        await on_event({
            "type": "thinking",
            "data": {" status": "analyzing", "step": "理解用户意图" }
        })
        await asyncio.sleep(0.3)

        # 2. 发送 tool_call 事件（Agent 正在调用工具）
        await on_event({
            "type": "tool_call",
            "data": {
                "tool": "code_search",
                "input": {"query": message, "limit": 5},
                "status": "invoking"
            }
        })

        # 3. 模拟工具执行延迟
        await asyncio.sleep(0.8)

        await on_event({
            "type": "tool_result",
            "data": {
                "tool": "code_search",
                "output": {"found": 3, "results": [
                    {"file": "auth.py", "line": 42, "match": "jwt.verify"},
                    {"file": "middleware.py", "line": 15, "match": "auth_check"},
                ]}
            }
        })

        # 4. 发送流式文本 token
        response_text = (
            "根据搜索结果，我找到了 3 处相关的认证代码。"
            "最关键的是 `auth.py` 第 42 行的 `jwt.verify()` 调用。"
            "建议在此处增加异常处理，防止 token 过期导致未认证请求通过。"
        )

        for token in response_text:
            await on_event({
                "type": "text_delta",
                "data": {" token": token }
            })
            await asyncio.sleep(0.05)  # 模拟 token 生成速度

        # 5. 发送生成式 UI（建议的代码修复片段）
        await on_event({
            "type": "generative_ui",
            "data": {
                "component": {
                    "type": "code_editor",
                    "language": "python",
                    "filename": "auth.py",
                    "highlight_lines": [40, 41, 42, 43, 44],
                    "content": '''async def verify_token(token: str) -> User:
    try:
        payload = jwt.verify(token, SECRET_KEY, algorithms=["HS256"])
        return User(id=payload["sub"], role=payload["role"])
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")  # ← 新增
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")  # ← 新增
    return None  # ← 可删除，此行不会执行''',
                }
            }
        })

        # 6. 最终状态
        await on_event({
            "type": "complete",
            "data": {" summary": "分析完成，包含代码修复建议" }
        })

agent = StreamingAgent()

# ─── AG-UI 端点 ───
@app.post("/v1/chat")
async def chat(request: dict):
    """
    创建新的聊天会话，返回 session_id
    """
    session_id = str(uuid.uuid4())
    agent.sessions[session_id] = {
        "history": [],
        "status": "active"
    }
    return {" session_id": session_id }

@app.post("/v1/chat/{session_id}/messages")
async def send_message(session_id: str, request: dict):
    """
    发送消息，开始流式响应
    """
    if session_id not in agent.sessions:
        raise HTTPException(status_code=404, detail="Session not found")

    message = request.get("content", "")
    agent.sessions[session_id]["history"].append({"role": "user", "content": message})

    async def event_generator() -> AsyncGenerator[str, None]:
        collected_tokens = []

        async def on_event(event: dict):
            nonlocal collected_tokens
            event_type = event["type"]

            if event_type == "text_delta":
                collected_tokens.append(event["data"]["token"])
                yield json.dumps({"event": "text", "data": event["data"]}) + "\n"

            elif event_type in ("thinking", "tool_call", "tool_result",
                                "generative_ui", "complete"):
                yield json.dumps({"event": event_type, "data": event["data"]}) + "\n"

        try:
            await agent.process_stream(session_id, message, on_event)
        except Exception as e:
            yield json.dumps({"event": "error", "data": {"message": str(e)}}) + "\n"

    return EventSourceResponse(
        event_generator(),
        media_type="text/event-stream"
    )

@app.delete("/v1/chat/{session_id}")
async def cancel_session(session_id: str):
    """
    取消会话
    """
    if session_id in agent.sessions:
        agent.sessions[session_id]["status"] = "cancelled"
    return {"status": "cancelled"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3.5 AG-UI 在协议栈中的定位

AG-UI 位于协议栈的最顶层，解决的是"Agent 如何实时、可控地与用户交互"的问题。它与 A2A 的关系可以这样理解：

• A2A：Agent 军团的内部协作语言
• AG-UI：Agent 向外部世界（用户）展示自己的窗口

一个完整的现代 AI 应用，可能同时使用全部三层协议：

用户界面
  ↑ AG-UI（SSE 流式推送、生成式 UI 组件、状态同步）
  ↓
你的后端 Agent 网关
  ↑ A2A（委托给专业子 Agent：代码 Agent、搜索 Agent、分析 Agent）
  ↓
子 Agent
  ↑ MCP（调用 GitHub API、搜索工具、数据库等）
  ↓
外部工具和数据

第四部分：三层协议的工程整合

4.1 典型架构：Agent 网关 + 协议桥接

在实际生产中，建议采用 Agent Gateway（网关） 架构，让网关统一处理协议转换，对外暴露统一接口：

# agent_gateway.py
import asyncio
from dataclasses import dataclass
from typing import Union
import mcp
import a2a
import ag_ui

@dataclass
class UnifiedRequest:
    source: str  # "frontend" | "agent" | "tool"
    intent: str
    payload: dict
    context: dict

class AgentGateway:
    """
    统一网关：接收来自前端、Agent 或工具的请求，
    根据 intent 路由到对应的协议处理器。
    """

    def __init__(self):
        # 初始化各协议客户端
        self.mcp_registry = MCPServiceRegistry()
        self.a2a_router = A2ARouter()
        self.ag_ui_publisher = AGUIPublisher()

    async def handle(self, req: UnifiedRequest):
        if req.source == "frontend":
            # AG-UI 路径：前端发起，路由到特定 Agent
            return await self._handle_frontend_request(req)

        elif req.source == "agent":
            # A2A 路径：Agent 之间协作
            return await self._handle_agent_collaboration(req)

        elif req.source == "tool":
            # MCP 路径：Agent 调用外部工具
            return await self._handle_tool_invocation(req)

    async def _handle_frontend_request(self, req: UnifiedRequest):
        # 1. 通过 AG-UI 建立 SSE 连接
        session = await self.ag_ui_publisher.create_session(req.context.get("user_id"))

        # 2. 将请求路由到合适的 Agent（A2A）
        task = await self.a2a_router.send_task({
            "target": req.payload["target_agent"],
            "skill": req.payload["skill"],
            "params": req.payload["params"]
        })

        # 3. 将 Agent 的输出通过 AG-UI 推送给前端
        async for event in task.stream():
            await session.push(event)

        return {" session_id": session.id }

    async def _handle_agent_collaboration(self, req: UnifiedRequest):
        # A2A 多跳路由
        return await self.a2a_router.relay(req.payload)

    async def _handle_tool_invocation(self, req: UnifiedRequest):
        # MCP 工具调用
        tool_name = req.payload["tool"]
        params = req.payload["params"]

        server = self.mcp_registry.get_server_for_tool(tool_name)
        result = await server.call_tool(tool_name, params)
        return result

4.2 协议选择决策树

在实际开发中，根据场景选择正确的协议至关重要：

遇到以下问题？
  │
  ├─ "我的 Agent 需要调用外部工具/API/数据库" ──→ MCP
  │   └─ 典型问题：如何标准化工具描述？
  │
  ├─ "我的多个 Agent 需要协作完成复杂任务" ──→ A2A
  │   └─ 典型问题：任务状态如何追踪？
  │
  ├─ "我的用户需要实时看到 Agent 的思考过程" ──→ AG-UI
  │   └─ 典型问题：流式输出如何渲染？
  │
  └─ "以上三种场景都有" ──→ 三层协议栈组合使用

4.3 常见坑与工程建议

4.3.1 MCP 的 Token 边界问题

MCP 的资源（Resources）默认会全量加载到上下文中。如果工具返回的数据量很大（如大型日志文件），会导致上下文溢出。

解决方案：在 MCP Server 端实现分页和过滤，不要让模型处理超过上下文窗口 20% 的单次数据。

@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    if name == "search_logs":
        # 限制返回行数，分页加载
        limit = min(arguments.get("limit", 100), 500)  # 最多返回 500 行
        offset = arguments.get("offset", 0)

        results = await query_logs(
            pattern=arguments["pattern"],
            limit=limit,
            offset=offset
        )

        # 添加元数据，帮助模型理解分页
        return [TextContent(
            type="text",
            text=f"[Showing {offset}-{offset+len(results)} results] " + results
        )]

4.3.2 A2A 的超时与幂等性

A2A 的 Task 可能需要长时间处理（分钟级甚至更长），但 HTTP 请求通常有 30s 超时限制。

解决方案：Task ID + 轮询模式。发起方获取 Task ID 后，通过 tasks/get 轮询状态，或使用 SSE 长连接订阅更新：

# 发起方：Task 提交后不阻塞等待
task_info = await a2a_client.tasks_send({
    "message": {...},
    "skillId": "heavy-analysis"
})
task_id = task_info["task"]["id"]
print(f"Task submitted: {task_id} — will poll for results")

# 轮询（适用于无 SSE 支持的场景）
import asyncio, aiohttp

async def poll_result(task_id: str, poll_interval: int = 5, max_wait: int = 300):
    start = asyncio.get_event_loop().time()
    while (asyncio.get_event_loop().time() - start) < max_wait:
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{A2A_ENDPOINT}/tasks/{task_id}") as resp:
                result = await resp.json()
                state = result["status"]["state"]

                if state == "completed":
                    return result["artifacts"]
                elif state == "failed":
                    raise RuntimeError(f"Task failed: {result['status'].get('error')}")

        await asyncio.sleep(poll_interval)

    raise TimeoutError(f"Task {task_id} did not complete within {max_wait}s")

4.3.3 AG-UI 的安全边界

AG-UI 允许 Agent 动态生成 UI 组件，这带来了 XSS 风险。永远不要让 Agent 直接生成 HTML/CSS 并注入 DOM。正确做法是 Agent 返回结构化的 component spec，前端根据白名单的组件类型渲染：

// 白名单允许的组件类型
const ALLOWED_COMPONENTS = new Set([
  "chart", "code_editor", "table", "form", "markdown", "image_grid"
]);

function renderAgentComponent(raw: any) {
  const type = raw.component?.type;

  if (!ALLOWED_COMPONENTS.has(type)) {
    console.warn(`Blocked disallowed component type: ${type}`);
    return null;  // 丢弃不可信组件
  }

  // 仅渲染白名单内的组件
  switch (type) {
    case "chart": return renderChart(raw.component.spec);
    case "code_editor": return renderCodeEditor(raw.component);
    // ...
  }
}

第五部分：框架视角——四大主流 Agent 开发框架的协议支持

了解了三层协议的技术细节，我们再从框架视角，看主流 Agent 开发框架对这三层协议的支持情况：

5.1 LangGraph（偏工作流编排）

LangGraph 是目前最成熟的 Multi-Agent 编排框架之一，由 LangChain 团队维护。它的状态机式 DAG 模型非常适合 A2A 的 Task 生命周期管理。

协议支持情况：

• MCP：通过 LangChain MCP 集成包支持，可以将 MCP Server 作为 Tool 接入
• A2A：社区已有 experimental adapter，官方路线图中有原生支持计划
• AG-UI：通过 CopilotKit 的 LangGraph 集成包支持

from langgraph.prebuilt import ToolNode
from langchain_mcp import MCPToolSet

# 接入 MCP 工具
mcp_tools = MCPToolSet("./file_search_server.py")
tool_node = ToolNode(mcp_tools)

# 构建多 Agent 图
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_agent)
graph.add_node("coder", coder_agent)
# ... A2A 协作通过 State 传递实现

5.2 OpenAI Agents SDK（偏单 Agent 工具调用）

OpenAI Agents SDK 是 OpenAI 官方推出的 Agent 开发工具，设计哲学偏向简单直接。

协议支持情况：

• MCP：原生支持，通过 MCPTool 直接接入
• A2A：目前仅支持单 Agent 模式，多 Agent 协作需要自行实现
• AG-UI：通过第三方集成支持

from agents import Agent, MCPTool

# 接入 MCP 工具（极简方式）
agent = Agent(
    name="Data Analyst",
    instructions="You analyze data and generate insights.",
    tools=[
        MCPTool(
            name="sql_query",
            server_path="./mcp_servers/data_tools.py",
            description="Execute SQL queries against the analytics database"
        ),
        MCPTool(
            name="visualize",
            server_path="./mcp_servers/chart_tools.py",
            description="Generate chart specifications"
        )
    ]
)

5.3 Google ADK（偏多 Agent 协作）

Google Agent Development Kit（ADK）是最早将 A2A 纳入设计哲学的框架之一，天然支持 Agent 之间的协作编排。

协议支持情况：

• MCP：通过 MCP Tool Server 集成
• A2A：官方支持，是 ADK 的核心协作机制
• AG-UI：通过 Gemini API 的 native streaming 支持

5.4 CrewAI（偏多 Agent 角色扮演）

CrewAI 采用了独特的"船员（Crew）"隐喻，每个 Agent 是一个有特定角色的"船员"，多个 Agent 组成"crew"协作完成任务。

协议支持情况：

• MCP：通过 Tool 机制接入
• A2A：Agent 间通过预定义的 task 传递通信，本质上是一种简化的 A2A 实现
• AG-UI：通过 CrewAI 的 callback 机制支持流式输出

5.5 框架选型建议

总结与展望

协议栈全景回顾

┌──────────────────────────────────────────────────────────┐
│                 AI Agent Protocol Stack                   │
├──────────────────────────────────────────────────────────┤
│                                                          │
│   Layer 3: AG-UI  ── Agent 实时交互协议                   │
│   发起者：CopilotKit · 核心：SSE 流式 + 生成式 UI          │
│   解决：Agent 如何向用户实时展示思考过程和结果              │
│                                                          │
│   Layer 2: A2A   ── Agent 间协作协议                      │
│   发起者：Google · 核心：Agent Card 发现 + Task 状态机    │
│   解决：多个 Agent 如何发现彼此、协调任务、追踪进度         │
│                                                          │
│   Layer 1: MCP   ── Agent 工具连接协议                    │
│   发起者：Anthropic · 核心：Tools/Resources/Prompts      │
│   解决：Agent 如何安全、标准地调用外部工具和数据            │
│                                                          │
└──────────────────────────────────────────────────────────┘

三个关键洞察

1. 三层协议不是竞争关系，而是分工关系。

MCP 让 Agent 看见世界，A2A 让 Agent 互相协作，AG-UI 让 Agent 走进用户界面。缺少任何一层，Agent 系统都是残缺的。

2. 2026 年的 Agent 开发已经进入"协议优先"时代。

过去靠 prompt 工程和私有 API 粘合的 Agent 系统，正在被标准化的协议栈取代。这意味着：工具生态会真正打通，不同框架的 Agent 可以互相调用，开发者不再需要为每个工具写定制适配。

3. 工程化的挑战从"如何让 Agent 做某件事"变成了"如何让 Agent 系统可靠地做某件事"。

当 Agent 数量从 1 变成 10、100，协议层的重要性就凸显出来：发现机制、状态追踪、错误恢复、安全边界——这些才是生产级 Multi-Agent 系统的真正壁垒。

未来展望

随着协议层的成熟，我们可以预期几个趋势：

1. Agent 市场会出现：类似 API 市场，Agent Provider 会注册标准化的 Agent Card，需求方通过 A2A 发现并调用
2. 协议层会出现中间件：类似于 API Gateway，有专门处理认证、限流、监控的 A2A/MCP 网关产品
3. 调试工具会成熟：Multi-Agent 系统的可观测性（tracing、logging、replay）会成为刚需
4. 第四层协议可能出现：当 Agent 数量足够多，Agent 的"服务治理"（限流、熔断、A/B 测试、灰度发布）可能催生新的协议层

无论趋势如何演进，理解这三层协议栈的底层逻辑，都能让你在 Agent 时代保持清晰的技术判断力。

关于作者：程序员茄子，关注 AI Agent 工程化、云原生基础设施与高性能系统架构。