A2A 协议架构：客户端-服务器模型解析

摘要：A2A（Agent2Agent）协议通过客户端-服务器模型实现了 AI 代理间的动态协作，为企业场景提供了标准化通信框架。本文深入剖析 A2A 的架构设计，聚焦客户端与服务器的交互、AgentCard 的作用、通信协议（HTTP 和 WebSocket）以及任务管理机制。结合 GitHub 仓库的实现、Mermaid 图表和代码示例，我们将揭示 A2A 如何在技术细节上支持多代理系统的互操作性，为开发者提供硬核的技术洞察。

1. 引言：架构驱动的代理协作

在企业 AI 系统中，代理（Agent）需要像微服务一样高效协作，处理从费用报销到供应链优化的复杂任务。Google 的 A2A（Agent2Agent） 协议通过客户端-服务器模型，为代理间通信提供了标准化框架。这种架构不仅支持动态发现和多模态交互，还确保了系统的可扩展性和可靠性。

A2A 的客户端-服务器模型借鉴了分布式系统的设计理念，但针对 AI 代理的动态性进行了优化。核心组件包括客户端（Host Agent）、服务器（Remote Agent）、AgentCard（元数据描述）和任务（Task）管理。本文将深入解析这一架构，结合 Google A2A GitHub 仓库的实现，揭示其硬核内核。

2. A2A 架构概览

A2A 的架构基于 客户端-服务器模型，但与传统 REST API 不同，它强调代理间的对等协作和动态协商。以下是核心组件的示意图：

graph TD
    A[Client: Host Agent] -->|HTTP/WebSocket| B[Server: Remote Agent]
    A -->|请求 AgentCard| B
    B -->|返回 AgentCard| A
    A -->|提交 Task| B
    B -->|返回 Task Result| A
    B --> C[AgentCard]
    B --> D[Task Lifecycle]
    C --> E[Capabilities]
    D --> F[Status Updates]

2.1 核心组件

客户端（Host Agent）：任务的发起者和协调者，负责发现 Remote Agent、协商交互模式并分派任务。
服务器（Remote Agent）：任务的执行者，暴露 AgentCard 和任务处理接口。
AgentCard：代理的元数据，定义名称、URL、能力（如 streaming、interactionModes）和任务 schema。
任务（Task）：代理间的工作单元，包含输入数据、状态（Created/In Progress/Completed）和输出结果。
通信层：支持 HTTP（同步请求）和 WebSocket（实时流和推送通知）。

2.2 设计原则

A2A 的架构遵循以下原则：

模块化：客户端和服务器松耦合，代理可独立开发和部署。
动态性：通过 AgentCard 实现运行时发现和协商。
可扩展性：支持多代理协作，适应从单机到分布式系统的场景。
可靠性：任务生命周期和状态更新确保通信一致性。

3. 客户端-服务器交互：动态协作的核心

A2A 的客户端-服务器交互分为三个阶段：发现、协商和任务执行。以下是详细解析。

3.1 代理发现

客户端（Host Agent）通过请求服务器的 AgentCard 了解其能力。AgentCard 的 JSON Schema（a2a.json）定义了以下关键字段：

name：代理名称（例如 “ExpenseAgent”）。
url：通信端点（例如 https://example.com/a2a）。
capabilities：功能描述（例如 {"streaming": false, "interactionModes": ["text", "form"]}）。
schema：任务输入/输出的数据结构。

发现过程的时序图如下：

sequenceDiagram
    participant C as Client (Host Agent)
    participant S as Server (Remote Agent)
    C->>S: GET /agentcard
    S-->>C: Return AgentCard JSON
    C->>S: Validate capabilities
    C->>S: Proceed to negotiation

3.2 交互协商

客户端根据 AgentCard 的 capabilities 协商交互模式。例如：

如果服务器支持 interactionModes: ["text", "form"]，客户端可能选择文本交互。
如果支持 streaming: true，客户端可以启用流式传输。

协商过程允许动态调整，例如服务器提议表单输入以补充数据。以下是协商的流程图：

flowchart TD
    A[Client Receives AgentCard] --> B[Parse Capabilities]
    B --> C{Supported Modes?}
    C -->|Text| D[Propose Text]
    C -->|Form| E[Propose Form]
    D --> F[Server Confirms]
    E --> F
    F --> G[Start Task]

3.3 任务执行

客户端提交任务（Task）后，服务器执行逻辑并返回结果。任务遵循状态生命周期（Created → In Progress → Completed/Failed）。任务的 JSON 结构包括：

taskId：唯一标识符。
type：任务类型（例如 “expense”）。
data：输入数据（符合 AgentCard 的 schema）。
status：当前状态。

任务执行的时序图如下：

sequenceDiagram
    participant C as Client
    participant S as Server
    C->>S: POST /task (Task JSON)
    S-->>C: Task Accepted (taskId)
    S->>S: Process Task
    S-->>C: Status Update (In Progress)
    S-->>C: Final Result (Completed/Failed)

4. 通信机制：HTTP vs. WebSocket

A2A 的通信层支持两种协议，分别满足不同场景：

4.1 HTTP

特点：基于请求-响应模型，适合简单任务（如查询 AgentCard 或提交任务）。
优势：实现简单，兼容性强，适合低频交互。
局限：无法支持实时流或推送通知。

示例 HTTP 请求：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
POST /a2a/task HTTP/1.1
Host: example.com
Content-Type: application/json

{
  "taskId": "task-001",
  "type": "expense",
  "data": {
    "amount": 100,
    "currency": "USD"
  }
}

响应：

1
2
3
4
5
6
7
HTTP/1.1 200 OK
Content-Type: application/json

{
  "taskId": "task-001",
  "status": "accepted"
}

4.2 WebSocket

特点：基于持久连接，支持实时双向通信，适合流式传输和推送通知。
优势：低延迟，适配复杂场景（如音视频交互或任务状态更新）。
局限：连接管理复杂，可能增加服务器负载。

示例 WebSocket 消息：

1
2
3
4
5
6
{
  "event": "task_update",
  "taskId": "task-001",
  "status": "in_progress",
  "progress": 50
}

通信机制的对比图：

graph TD
    A[Client] -->|HTTP| B[Server]
    A -->|WebSocket| C[Server]
    B --> D[Task Response]
    C --> E[Streaming Updates]
    C --> F[Push Notifications]
    style B fill:#bbf,stroke:#333
    style C fill:#bfb,stroke:#333

4.3 选择策略

HTTP：用于 AgentCard 获取、简单任务提交和状态查询。
WebSocket：用于多模态交互（例如音视频流）或实时任务监控。
混合模式：客户端可能先用 HTTP 获取 AgentCard，再用 WebSocket 执行任务。

GitHub Issues 提到，社区正在优化 WebSocket 的重连机制，以提升可靠性。

5. 任务管理：状态机与可靠性

A2A 的任务管理基于状态机，确保通信一致性和可靠性。任务生命周期包括：

Created：任务被客户端提交。
In Progress：服务器开始处理。
Completed：任务成功，返回结果。
Failed：任务失败，返回错误。
Canceled：任务被主动取消（可选）。

状态机的流程图如下：

flowchart TD
    A[Created] --> B[In Progress]
    B --> C{Outcome}
    C --> D[Completed]
    C --> E[Failed]
    C --> F[Canceled]
    D --> G[Result Returned]
    E --> H[Error Reported]
    F --> I[Task Aborted]

5.1 可靠性机制

幂等性：任务通过 taskId 确保重复提交不会导致副作用。
状态同步：WebSocket 推送实时更新，HTTP 提供轮询备用。
错误处理：服务器返回详细错误信息（例如 { "error": "Invalid amount" }）。

5.2 性能考量

任务管理的性能受以下因素影响：

网络延迟：WebSocket 的低延迟适合实时场景，但 HTTP 的轮询可能增加开销。
并发性：多任务并发需要服务器优化调度，GitHub 仓库的样本代码使用了异步处理（asyncio）。
负载均衡：分布式部署需考虑 AgentCard 的缓存和任务分发。

6. 代码示例：实现 A2A 客户端与服务器

以下是一个基于 samples/python/agents/google_adk 的费用报销代理实现，展示客户端-服务器交互。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
# 服务器：费用报销代理
from a2a import A2AServer, AgentCard, Task

class ExpenseAgent(A2AServer):
    def __init__(self):
        card = AgentCard(
            name="ExpenseAgent",
            description="Processes expense reimbursements",
            url="http://localhost:8080/a2a",
            capabilities={
                "streaming": False,
                "interactionModes": ["text", "form"],
                "pushNotifications": True
            },
            schema={
                "input": {
                    "type": "object",
                    "properties": {
                        "amount": {"type": "number"},
                        "currency": {"type": "string"}
                    }
                }
            }
        )
        super().__init__(card=card)

    async def handle_task(self, task: Task) -> dict:
        if task["type"] != "expense":
            return {"status": "failed", "error": "Invalid task type"}
        amount = task["data"]["amount"]
        currency = task["data"]["currency"]
        if amount <= 0:
            return {"status": "failed", "error": "Invalid amount"}
        return {
            "status": "completed",
            "result": f"Approved {amount} {currency}"
        }

# 客户端：调用费用报销代理
from a2a import A2AClient

async def main():
    client = A2AClient("http://localhost:8080/a2a")
    # 获取 AgentCard
    agent_card = await client.get_agent_card()
    print(f"Agent: {agent_card['name']}, Capabilities: {agent_card['capabilities']}")

    # 提交任务
    task = {
        "taskId": "task-001",
        "type": "expense",
        "data": {"amount": 100, "currency": "USD"}
    }
    result = await client.submit_task(task)
    print(f"Result: {result}")

if __name__ == "__main__":
    import asyncio
    server = ExpenseAgent()
    # 异步运行服务器和客户端
    asyncio.run(main())
    server.run(port=8080)