A2A(Agent2Agent)系列专题 (六) A2A 的 JSON Schema:协议的核心规范

A2A 的 JSON Schema:协议的核心规范

摘要:JSON Schema 是 A2A(Agent2Agent)协议的基石,定义了代理间通信的数据结构和规则,确保互操作性和一致性。本文深入剖析 A2A 的 JSON Schema,聚焦 AgentCard、任务(Task)和认证(AgentAuthentication)等核心组件的设计与实现。结合 GitHub 仓库的代码、Mermaid 图表和协议细节,我们将揭示 A2A 如何通过标准化 Schema 支持动态协作,为开发者提供硬核的技术洞察。

1. 引言:JSON Schema 的力量

在分布式 AI 系统中,代理(Agent)需要以一致的方式交换信息,无论是描述自身能力、提交任务还是协商交互模式。Google 的 A2A(Agent2Agent) 协议通过 JSON Schema 提供了标准化的数据定义,类似 Web 的 OpenAPI 或 GraphQL Schema,确保不同代理能够无缝理解彼此。

A2A 的 JSON Schema(a2a.json)是协议的核心,涵盖了代理元数据(AgentCard)、任务结构(Task)、认证机制(AgentAuthentication)等关键部分。它的设计不仅保证了数据一致性,还支持动态性和扩展性。本文将深入解析 A2A 的 JSON Schema,结合 Google A2A GitHub 仓库 的实现,揭示其硬核内核。

2. JSON Schema 在 A2A 中的作用

JSON Schema 是一种用于定义 JSON 数据结构的规范(参考 json-schema.org),广泛应用于 API 设计和数据验证。A2A 利用 JSON Schema 实现以下目标:

  • 标准化:确保所有代理使用一致的数据格式,例如 AgentCard 和 Task 的字段定义。
  • 验证:客户端和服务器可以验证数据的合法性,减少错误。
  • 动态发现:通过 Schema 描述代理能力和任务要求,支持运行时协商。
  • 扩展性:允许社区添加新功能(例如新的交互模式)而无需破坏兼容性。

A2A 的 JSON Schema 文件(a2a.json)托管于 GitHub,定义了协议的完整规范。以下是 Schema 的核心组件示意图:

classDiagram
    class A2A_Schema {
        +AgentCard agentCard
        +Task task
        +AgentAuthentication authentication
    }
    class AgentCard {
        +String name
        +String description
        +String url
        +Object capabilities
        +Object schema
    }
    class Task {
        +String taskId
        +String type
        +Object data
        +String status
        +Object result
        +Object error
    }
    class AgentAuthentication {
        +Array schemes
        +String credentials
    }
    class AgentCapabilities {
        +Boolean streaming
        +Boolean pushNotifications
        +Array interactionModes
    }
    A2A_Schema --> AgentCard
    A2A_Schema --> Task
    A2A_Schema --> AgentAuthentication
    AgentCard --> AgentAuthentication
    AgentCard --> AgentCapabilities

3. AgentCard Schema:代理的数字蓝图

3.1 结构与字段

AgentCard 是 A2A 的核心组件,描述代理的元数据。其 JSON Schema 定义了以下字段:

  • name(字符串):代理的唯一标识符,例如 “ExpenseAgent”。
  • description(字符串):代理的功能说明,例如 “Processes expense reimbursements”。
  • url(字符串):通信端点,例如 https://example.com/a2a
  • authentication(对象):认证方案,包含 schemes(支持的认证类型,如 ["Bearer"])和 credentials(可选凭据)。
  • capabilities(对象):代理的功能,例如:
    • streaming(布尔值):是否支持流式传输。
    • pushNotifications(布尔值):是否支持推送通知。
    • interactionModes(数组):支持的交互模式,如 ["text", "form", "video"]
  • schema(对象):任务输入/输出的 JSON Schema,例如定义 amountcurrency

以下是 AgentCard 的简化 JSON 示例:

 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
{
  "name": "ExpenseAgent",
  "description": "Processes expense reimbursements",
  "url": "https://example.com/a2a",
  "authentication": {
    "schemes": ["Bearer"],
    "credentials": "token123"
  },
  "capabilities": {
    "streaming": false,
    "pushNotifications": true,
    "interactionModes": ["text", "form"]
  },
  "schema": {
    "input": {
      "type": "object",
      "properties": {
        "amount": {"type": "number"},
        "currency": {"type": "string"}
      },
      "required": ["amount", "currency"]
    },
    "output": {
      "type": "object",
      "properties": {
        "status": {"type": "string"},
        "result": {"type": "string"}
      }
    }
  }
}

3.2 设计亮点

  • 模块化capabilitiesschema 分离,允许独立扩展功能或数据结构。
  • 验证requiredproperties 确保输入数据的完整性。
  • 动态性interactionModes 支持运行时协商,例如从文本切换到表单。

3.3 Mermaid 类图

以下是 AgentCard 的详细结构:

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
    }
    class Schema {
        +Object input
        +Object output
    }
    AgentCard --> AgentAuthentication
    AgentCard --> AgentCapabilities
    AgentCard --> Schema

4. Task Schema:任务的标准化描述

4.1 结构与字段

任务(Task)是代理间的工作单元,其 Schema 定义了以下字段:

  • taskId(字符串):任务的唯一标识符,例如 “task-001”。
  • type(字符串):任务类型,例如 “expense”。
  • data(对象):任务输入数据,符合 AgentCard 的 schema.input
  • status(字符串):任务状态,枚举值包括 createdin_progresscompletedfailedcanceled
  • result(对象):任务输出,符合 AgentCard 的 schema.output
  • error(对象):错误信息,仅在 status: failed 时存在。

以下是 Task 的 JSON 示例:

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

完成后的响应:

1
2
3
4
5
6
7
8
9
{
  "taskId": "task-001",
  "type": "expense",
  "status": "completed",
  "result": {
    "status": "approved",
    "message": "Processed 100 USD"
  }
}

4.2 状态生命周期

Task Schema 内置了状态机,定义了任务的合法状态转换:

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]
  • 验证:Schema 的 enum 限制 status 的值,确保状态一致性。
  • 扩展性resulterror 的结构由 AgentCard 的 schema.output 定义,允许自定义。

4.3 动态性支持

Task Schema 支持动态调整。例如,代理可能在 in_progress 状态返回一个表单请求(通过 interactionModes: form),客户端根据 Schema 渲染 UI。这种机制增强了交互的灵活性。

5. AgentAuthentication Schema:安全的基石

5.1 结构与字段

AgentAuthentication 定义了代理的认证方式,包含以下字段:

  • schemes(数组):支持的认证类型,例如 ["Bearer", "Basic"]
  • credentials(字符串):认证凭据,例如令牌或密钥(可选)。

示例:

1
2
3
4
{
  "schemes": ["Bearer"],
  "credentials": "token123"
}

5.2 设计考量

  • 简单性:当前支持基础认证方案,易于实现。
  • 扩展性schemes 允许添加新认证类型(例如 OAuth 2.0)。
  • 安全性:凭据通过 HTTPS 或 WebSocket 加密传输。

GitHub Issues 提到,社区计划引入更复杂的授权机制,例如基于角色的访问控制(RBAC)。

6. 代码示例:基于 Schema 的代理实现

以下是一个基于 samples/python/agents/google_adk 的费用报销代理,展示如何使用 JSON Schema 定义和验证 AgentCard 与 Task。

 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
63
64
# 费用报销代理
from a2a import A2AServer, AgentCard, Task
from jsonschema import validate

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

    async def handle_task(self, task: Task) -> dict:
        # 验证任务输入
        try:
            validate(instance=task["data"], schema=self.card.schema["input"])
        except Exception as e:
            return {"status": "failed", "error": str(e)}

        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"}

        # 模拟处理
        result = {
            "status": "approved",
            "message": f"Processed {amount} {currency}"
        }
        # 验证输出
        validate(instance=result, schema=self.card.schema["output"])
        return {"status": "completed", "result": result}

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

代码解析

  1. AgentCard 定义:包含完整的 Schema,指定输入/输出的数据结构。
  2. 任务验证:使用 jsonschema 库验证任务的 dataresult,确保符合 Schema。
  3. 错误处理:捕获验证失败或逻辑错误,返回标准化的错误响应。

7. 硬核设计:Schema 的权衡

7.1 标准化与复杂性

  • 优点:JSON Schema 确保数据一致性,降低集成成本。
  • 缺点:复杂的 Schema(例如嵌套的 capabilities)可能增加学习曲线。社区正在讨论简化提案(参考 GitHub Issues)。

7.2 动态性与性能

  • 优点:Schema 支持动态协商(例如 interactionModes),适配多模态场景。
  • 缺点:Schema 验证可能引入性能开销,尤其在高并发场景。优化方案包括缓存验证结果。

7.3 扩展性与兼容性

  • 优点:Schema 的 additionalProperties 允许扩展,社区可添加新字段。
  • 缺点:版本兼容性需谨慎管理,避免破坏现有代理。

8. 结语:Schema 的未来

A2A 的 JSON Schema 是协议的核心,为代理间通信提供了标准化和动态性的基石。通过 AgentCard、Task 和 AgentAuthentication 的定义,A2A 实现了从简单任务到复杂工作流的灵活支持。未来,协议可能优化 Schema 的性能、增强认证机制,进一步提升企业级应用的能力。

在下一篇文章中,我们将探讨 A2A 的代理发现与协商机制,深入分析动态交互的实现细节。欢迎访问 A2A GitHub 仓库,加入社区,探索 AI 协作的未来!

updatedupdated2025-04-172025-04-17