殷龙飞的博客

A2A（Agent2Agent）系列专题 (十三) 快速入门：搭建你的第一个 A2A 代理

 协议/大模型/A2A  2712  6 分钟

快速入门：搭建你的第一个 A2A 代理

摘要：A2A（Agent2Agent）协议通过标准化的通信机制实现了 AI 代理间的协作。本文通过分步教程，指导开发者基于 Google 的 A2A 协议搭建第一个代理，聚焦 Python 实现、AgentCard 配置和任务处理逻辑。结合 GitHub 仓库的 google_adk 示例、Mermaid 图表和调试技巧，我们将揭示 A2A 代理开发的硬核细节，帮助开发者快速上手并为企业 AI 系统构建高效的协作组件。

1. 引言：为什么搭建 A2A 代理？

在企业 AI 系统中，代理（Agent）是处理特定任务的独立模块，例如费用报销、客服支持或数据分析。Google 的 A2A（Agent2Agent） 协议通过 AgentCard、任务生命周期和 HTTP/WebSocket 通信，标准化了代理间的协作。搭建一个 A2A 代理不仅能帮助开发者理解协议的核心机制，还能为复杂系统（如多代理协作）奠定基础。

本文基于 GitHub 仓库 https://github.com/google/A2A 的 samples/python/agents/google_adk 示例，展示如何从零搭建一个费用报销代理，覆盖环境配置、代码实现、测试和调试。无论你是初学者还是资深开发者，这篇硬核教程都将为你提供实操指导。

2. 前置条件与环境准备

2.1 开发环境

  • 操作系统：Windows、macOS 或 Linux。
  • Python：3.8 或以上，推荐 3.10。
  • 依赖：aiohttp（异步 HTTP）、websockets（WebSocket 支持）、a2a（A2A 库，假设已发布）。
  • 工具：Git、VS Code 或 PyCharm、Postman（可选，测试 API）。

2.2 安装依赖

克隆 A2A 仓库并安装依赖：

1
2
3

git clone https://github.com/google/A2A.git
cd A2A/samples/python/agents/google_adk
pip install aiohttp websockets

如果 a2a 库尚未发布，可直接使用仓库中的 a2a.py 模块（假设包含 A2AServer 和 A2AClient 类）。

2.3 项目结构

创建以下目录结构：

my-a2a-agent/
├── agent.py        # 代理服务器实现
├── client.py       # 客户端测试脚本
├── requirements.txt # 依赖列表
└── config.json     # AgentCard 配置

3. 搭建流程：从零到运行

以下是搭建 A2A 代理的分步流程，参考规划中的 Mermaid 图表：

flowchart TD
    A[Clone Repository] --> B[Install Dependencies]
    B --> C[Configure AgentCard]
    C --> D[Run A2A Server]
    D --> E[Test with Client]
    E --> F[View Results]

3.1 步骤 1：配置 AgentCard

AgentCard 是 A2A 代理的元数据，定义名称、能力和服务地址。创建 config.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
32

{
  "name": "ExpenseAgent",
  "description": "Processes expense reimbursements",
  "url": "http://localhost:8080/a2a",
  "authentication": {
    "schemes": ["Bearer"],
    "credentials": "token123"
  },
  "capabilities": {
    "streaming": false,
    "pushNotifications": true,
    "interactionModes": ["text"],
    "stateTransitionHistory": true
  },
  "schema": {
    "input": {
      "type": "object",
      "properties": {
        "amount": {"type": "number"},
        "currency": {"type": "string"}
      },
      "required": ["amount", "currency"]
    },
    "output": {
      "type": "object",
      "properties": {
        "status": {"type": "string"},
        "message": {"type": "string"}
      }
    }
  }
}

解析：

  • authentication：使用 Bearer 令牌认证。
  • capabilities：支持推送通知和文本交互，记录状态转换历史。
  • schema：定义输入（金额和货币）和输出（状态和消息）格式。

3.2 步骤 2：实现代理服务器

创建 agent.py，实现费用报销代理：

 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

import asyncio
import json
from aiohttp import web
from a2a import A2AServer, AgentCard

class ExpenseAgent(A2AServer):
    def __init__(self):
        with open("config.json") as f:
            card_data = json.load(f)
        card = AgentCard(**card_data)
        super().__init__(card=card)

    async def verify_auth(self, request):
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            raise web.HTTPUnauthorized(text="Missing token")
        token = auth_header.replace("Bearer ", "")
        if token != self.card.authentication["credentials"]:
            raise web.HTTPForbidden(text="Invalid token")
        return True

    async def handle_task(self, request, task: dict) -> dict:
        await self.verify_auth(request)
        task_id = task["taskId"]
        await self.notify_status(task_id, "in_progress")

        if task["type"] != "expense":
            await self.notify_status(task_id, "failed")
            return {"status": "failed", "error": "Invalid task type"}

        amount = task["data"]["amount"]
        currency = task["data"]["currency"]
        if amount <= 0:
            await self.notify_status(task_id, "failed")
            return {"status": "failed", "error": "Amount must be positive"}

        # 模拟处理
        await asyncio.sleep(1)
        result = {"status": "approved", "message": f"Processed {amount} {currency}"}
        await self.notify_status(task_id, "completed")
        return {"status": "completed", "result": result}

    async def websocket_handler(self, websocket, path):
        async for message in websocket:
            data = json.loads(message)
            if data["event"] == "subscribe":
                task_id = data["taskId"]
                # 模拟状态推送
                await websocket.send(json.dumps({
                    "event": "task_update",
                    "taskId": task_id,
                    "status": "in_progress"
                }))

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

解析：

  • A2AServer：继承基类，加载 AgentCard 配置。
  • verify_auth：验证 Bearer 令牌，确保安全（参考第十一篇）。
  • handle_task：处理费用报销任务，验证输入并返回结果。
  • websocket_handler：支持实时状态更新。

3.3 步骤 3：创建客户端测试脚本

创建 client.py，测试代理功能：

 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

import asyncio
import aiohttp
from a2a import A2AClient

async def test_agent(remote_url: str):
    async with aiohttp.ClientSession(headers={"Authorization": "Bearer token123"}) as session:
        client = A2AClient(remote_url, session=session)

        # 获取 AgentCard
        agent_card = await client.get_agent_card()
        print(f"Agent: {agent_card['name']}")

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

        # 订阅状态更新
        async for update in client.subscribe_task_updates(task["taskId"]):
            print(f"Status update: {update}")
            if update["status"] in ["completed", "failed"]:
                break

if __name__ == "__main__":
    asyncio.run(test_agent("http://localhost:8080/a2a"))

解析：

  • A2AClient：封装 HTTP 和 WebSocket 通信。
  • get_agent_card：获取代理元数据。
  • submit_task：提交任务并验证响应。
  • subscribe_task_updates：通过 WebSocket 监控状态。

3.4 步骤 4：运行与测试

  1. 启动服务器：
    1

    python agent.py
  2. 运行客户端：
    1

    python client.py
  3. 预期输出：
    Agent: ExpenseAgent
Task submitted: {'taskId': 'task-001', 'status': 'accepted'}
Status update: {'event': 'task_update', 'taskId': 'task-001', 'status': 'in_progress'}
Status update: {'event': 'task_update', 'taskId': 'task-001', 'status': 'completed'}

3.5 步骤 5：调试与验证

  • 日志：在 handle_task 中添加日志，记录任务处理细节。
  • Postman 测试：发送 POST 请求到 http://localhost:8080/a2a/task，验证 API。
  • 错误检查：测试无效输入（如负金额），确保返回正确的错误响应。

4. 优化与扩展

4.1 性能优化

  • 异步处理：已使用 asyncio 和 aiohttp，支持高并发。
  • 缓存：将 AgentCard 缓存到 Redis，减少重复请求（参考第十篇）。
  • 连接池：限制 WebSocket 连接数，优化资源使用。

4.2 安全性

  • 认证：已实现 Bearer 令牌，未来可集成 JWT 或 OAuth 2.0（参考第十篇）。
  • 加密：部署时启用 HTTPS（参考第十一篇的安全性设计）。

4.3 扩展功能

  • 多模态交互：添加 form 或 video 模式（参考第十一篇）。
  • 状态持久化：使用 Redis 保存任务状态，增强可靠性（参考第十二篇）。
  • 多代理协作：扩展为多代理系统（下一篇文章主题）。

5. 调试技巧

5.1 常见问题

  • AgentCard 错误：检查 config.json 的 JSON 格式和 schema 字段。
  • 认证失败：确保客户端的 Authorization 头与服务器的 credentials 匹配。
  • WebSocket 断连：添加心跳机制（参考第十二篇）。

5.2 调试工具

  • 日志：使用 Python 的 logging 模块记录请求和状态。
  • Postman：测试 HTTP 端点，验证任务提交。
  • Wireshark：分析 WebSocket 通信（高级调试）。

5.3 日志示例

在 agent.py 中添加日志：

1
2
3
4
5
6
7
8
9

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def handle_task(self, request, task: dict) -> dict:
    logger.info(f"Processing task: {task['taskId']}")
    await self.verify_auth(request)
    # ...

6. 硬核设计：A2A 代理的权衡

6.1 简单性 vs 功能

  • 优势：简单的 AgentCard 和任务处理逻辑降低了开发门槛。
  • 挑战：复杂场景需要多模态支持和分布式部署。
  • 优化：通过模块化设计（如分离认证和任务逻辑），便于扩展。

6.2 性能 vs 可靠性

  • 优势：异步框架和 WebSocket 提供高性能和实时性。
  • 挑战：高并发下需优化连接管理和状态同步。
  • 优化：参考第十二篇的 Redis 和消息确认机制。

6.3 本地 vs 云端

  • 优势：本地开发便于调试和快速迭代。
  • 挑战：云端部署需考虑负载均衡和安全性。
  • 优化：下一篇文章将介绍云端部署（第十八篇）。

7. 应用场景与展望

你的第一个 A2A 代理可用于：

  • 费用报销：处理财务任务，验证金额和货币。
  • 客服支持：扩展为多模态客服代理（参考第十一篇）。
  • 企业自动化：作为多代理系统的基础组件（参考第十四篇）。

Future enhancements:

  • 插件化：支持动态加载新交互模式。
  • 容器化：使用 Docker 简化部署（第十八篇）。
  • AI 集成：结合 Google Cloud AI 增强任务处理。

8. 结语：迈向多代理协作

通过本教程，你已成功搭建并测试了一个 A2A 代理，掌握了 AgentCard 配置、任务处理和实时通信的核心技能。这只是开始！A2A 的真正潜力在于多代理协作和企业级部署。下一篇文章将探讨如何构建多代理系统，展示 A2A 的协同能力。

欢迎访问 A2A GitHub 仓库，加入社区，分享你的代理开发经验！

参考资料

updatedupdated2025-04-172025-04-17

相关文章：