A2A 的核心概念：代理、AgentCard 和任务

摘要：A2A（Agent2Agent）协议的核心在于代理（Agent）、AgentCard 和任务（Task）。这些概念定义了 AI 代理如何描述自身、发现彼此并协作完成工作。本文深入探讨这三大组件的设计理念、技术细节和实现方式，结合 GitHub 仓库的代码和 Mermaid 图表，揭示 A2A 如何通过标准化实现多代理系统的互操作性。无论你是开发者还是 AI 研究者，这篇文章将带你走进 A2A 的技术内核。

1. 引言：从孤立到协作的基石

在企业 AI 场景中，代理（Agent）是执行任务的基本单元，例如处理费用报销、生成报表或协调物流。然而，代理的多样性和复杂性带来了挑战：如何让不同来源、不同功能的代理高效协作？Google 的 A2A 协议通过三个核心概念解决了这一问题：

代理（Agent）：任务的执行者，分为 Host Agent 和 Remote Agent。
AgentCard：代理的“身份卡”，描述其能力和通信方式。
任务（Task）：代理间交换的工作单元，遵循明确的状态生命周期。

这三大概念构成了 A2A 的基础，类似互联网的 IP 地址、DNS 和数据包，定义了多代理系统的通信规则。本文将从技术视角深入剖析每个概念，结合 Google A2A GitHub 仓库的实现，揭示其硬核设计。

2. 代理（Agent）：A2A 的执行单元

2.1 代理的定义

在 A2A 中，代理是能够接收、处理和响应任务的独立实体。代理可以是简单的脚本（例如验证费用报销的 Python 程序），也可以是复杂的 AI 系统（例如结合大语言模型和数据库的客服代理）。A2A 将代理分为两类：

Host Agent：任务的发起者和协调者，负责发现 Remote Agent 并分派任务。
Remote Agent：任务的执行者，通过 A2A 协议向 Host Agent 返回结果。

代理之间的关系是动态的，一个代理可以同时扮演 Host 和 Remote 角色，类似微服务架构中的服务调用。

2.2 代理的职责

每个代理的核心职责包括：

自我描述：通过 AgentCard 声明自己的功能和通信方式。
任务处理：接收任务，执行逻辑，返回结果或错误。
动态交互：与对端代理协商交互模式（例如文本、表单或音视频）。

以下是一个代理交互的简化架构图：

graph TD
    A[User] -->|提交请求| B[Host Agent]
    B -->|任务分派| C[Remote Agent 1]
    B -->|任务分派| D[Remote Agent 2]
    C -->|结果返回| B
    D -->|结果返回| B
    B -->|汇总结果| A

2.3 实现细节

根据 GitHub 仓库的 samples/python/agents/google_adk，一个典型的代理实现包括：

初始化：定义 AgentCard 和通信端点。
任务处理：实现任务逻辑，例如解析输入并生成输出。
通信：通过 HTTP 或 WebSocket 与其他代理交互。

我们将在后续小节通过代码示例进一步展示。

3. AgentCard：代理的数字名片

3.1 AgentCard 的作用

AgentCard 是 A2A 协议的核心创新，相当于代理的“身份证”。它以 JSON 格式描述代理的元数据，让其他代理能够快速了解其身份、能力和通信方式。AgentCard 的设计灵感类似于 Web 的 OpenAPI 规范，但更专注于 AI 代理的动态交互。

3.2 AgentCard 的结构

根据 a2a.json 的 JSON Schema，AgentCard 包含以下关键字段：

name：代理的唯一标识符（字符串，例如 “ExpenseAgent”）。
description：代理的功能描述（字符串，例如 “Handles expense reimbursements”）。
url：代理的通信端点（字符串，例如 https://example.com/a2a）。
authentication：认证方案（对象，包含 schemes 和 credentials）。
capabilities：代理的功能描述（对象，例如支持 streaming 或 pushNotifications）。
schema：任务输入/输出的 JSON Schema（对象，定义数据结构）。

以下是 AgentCard 的 Mermaid 类图：

classDiagram
    class AgentCard {
        +String name
        +String description
        +String url
        +Object authentication
        +Object capabilities
        +Object schema
    }
    class AgentAuthentication {
        +Array schemes
        +String credentials
    }
    class AgentCapabilities {
        +Boolean streaming
        +Boolean pushNotifications
        +Boolean stateTransitionHistory
        +Array interactionModes
    }
    AgentCard --> AgentAuthentication
    AgentCard --> AgentCapabilities

3.3 动态发现与协商

AgentCard 的最大价值在于支持 动态发现。当 Host Agent 需要协作时，它会请求 Remote Agent 的 AgentCard，解析其 capabilities 和 schema，从而决定如何交互。例如：

如果 Remote Agent 支持 interactionModes: ["text", "form"]，Host Agent 可以选择文本或表单交互。
如果支持 streaming: true，Host Agent 可以启用流式传输。

这种机制减少了硬编码的配置，提高了系统的灵活性。以下是一个发现过程的时序图：

sequenceDiagram
    participant H as Host Agent
    participant R as Remote Agent
    H->>R: GET /agentcard
    R-->>H: Return AgentCard JSON
    H->>R: Propose interaction (e.g., text)
    R-->>H: Confirm or suggest alternative
    H->>R: Submit Task
    R-->>H: Task Result

3.4 示例：AgentCard 的 JSON

以下是一个费用报销代理的 AgentCard 示例（简化版）：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
{
  "name": "ExpenseAgent",
  "description": "Processes expense reimbursements",
  "url": "https://example.com/a2a/expense",
  "authentication": {
    "schemes": ["Bearer"],
    "credentials": "token123"
  },
  "capabilities": {
    "streaming": false,
    "pushNotifications": true,
    "interactionModes": ["text", "form"]
  },
  "schema": {
    "input": {
      "type": "object",
      "properties": {
        "amount": {"type": "number"},
        "currency": {"type": "string"}
      }
    }
  }
}

4. 任务（Task）：A2A 的工作单元

4.1 任务的定义

任务是代理间交换的工作单元，包含输入数据、执行指令和输出结果。A2A 的任务设计借鉴了工作流系统（Workflow System），但更轻量，专注于代理间的动态协作。

4.2 任务的生命周期

任务遵循明确的状态机，生命周期包括：

Created：任务被 Host Agent 创建，等待分派。
In Progress：Remote Agent 开始处理任务。
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]

4.3 任务的结构

任务以 JSON 格式传递，典型字段包括：

taskId：任务的唯一标识符（字符串）。
type：任务类型（字符串，例如 “expense”）。
data：任务输入数据（对象，符合 AgentCard 的 schema）。
status：任务状态（字符串，例如 “in_progress”）。
result：任务输出（对象，仅在 Completed 状态）。
error：错误信息（对象，仅在 Failed 状态）。

以下是一个任务示例：

1
2
3
4
5
6
7
8
9
{
  "taskId": "task-001",
  "type": "expense",
  "data": {
    "amount": 100,
    "currency": "USD"
  },
  "status": "created"
}

4.4 任务的动态性

A2A 的任务支持动态调整。例如，Remote Agent 可能在处理过程中请求额外输入（通过表单交互），Host Agent 会根据 AgentCard 的 interactionModes 渲染界面。这种动态性是 A2A 区别于传统 API 的关键。

5. 代码示例：实现一个 A2A 代理

为了将概念落地，我们基于 GitHub 仓库的 samples/python/agents/google_adk 提供一个费用报销代理的实现，展示代理、AgentCard 和任务的交互。

 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
# 费用报销代理实现
from a2a import A2AServer, AgentCard, Task

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

    def handle_task(self, task: Task) -> dict:
        # 处理任务逻辑
        if task["type"] == "expense":
            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}"
            }
        return {
            "status": "failed",
            "error": "Unsupported task type"
        }

if __name__ == "__main__":
    server = ExpenseAgent()
    server.run(port=8080)