Google Agent2Agent 协议 (A2A): 大模型时代 AI 智能体互操作的基石

Agent2Agent 协议 (A2A) 的出现，预示着大模型技术驱动的 AI 智能体生态正在迎来一个崭新的互联互通时代。本文将深入剖析 A2A 协议的技术架构、工作流程以及关键概念，帮助开发者全面了解这一协议，并掌握如何利用它构建更强大、更协作的 AI 智能体系统。A2A 协议旨在解决不同 AI 智能体之间的互操作性问题，让它们能够像人类一样协同完成复杂的任务，从而释放大模型技术的全部潜力。

核心概念：AgentCard – AI 智能体的身份标识

在 A2A 协议中，每个 AI 智能体都拥有一个独一无二的身份标识，称为 AgentCard。你可以把 AgentCard 想象成 AI 智能体的名片，它包含了该智能体的关键信息，例如：

Capabilities and skills (能力和技能)：智能体能够执行的任务类型，例如 “预订机票”、”创建披萨订单” 等。
Endpoint URL (终端 URL)：接收 A2A 请求的地址，其他智能体可以通过此 URL 与之通信。
Authentication requirements (认证要求)：访问该智能体所需的认证方式，例如 API 密钥、OAuth 2.0 等。
Protocol version compatibility information (协议版本兼容性信息)：支持的 A2A 协议版本，确保不同智能体之间的兼容性。

AgentCard 使用 JSON 格式进行描述，这使得它易于解析和使用。例如，一个披萨销售智能体的 AgentCard 可能包含以下信息：

{
  "name": "pizza_seller_agent",
  "description": "Helps with creating pizza orders",
  "url": "http://pizza-seller.example.com/",
  "version": "1.0.0",
  "authentication": {
    "schemes": ["Basic"]
  },
  "defaultInputModes": ["text", "text/plain"],
  "defaultOutputModes": ["text", "text/plain"],
  "capabilities": [],
  "skills": [
    {
      "id": "create_pizza_order",
      "name": "Pizza Order Creation Tool",
      "description": "Helps with creating pizza orders",
      "tags": ["pizza order creation"],
      "examples": ["I want to order 2 pepperoni pizzas"]
    }
  ]
}

拥有了 AgentCard，其他智能体就能够了解该智能体的能力，并根据需要与之建立连接，发起 Task (任务)。

核心流程：Discovery – 智能体的发现与连接

A2A 协议的第一步是 Discovery (发现) 过程，即客户端智能体如何找到能够满足其需求的远程智能体。这个过程类似于在互联网上搜索服务，客户端智能体根据自身的需求，在已知提供商处检索 AgentCard，从而找到合适的远程智能体。

在企业级应用中，通常需要一个 centralized agent catalog/registry (集中式智能体目录/注册中心) 来管理和维护所有可用的 AgentCard。这样可以确保客户端智能体能够找到可信赖的智能体，避免恶意智能体的攻击。例如，一个旅游预订智能体需要预订机票的功能，它可以从一个可信赖的智能体目录中查找提供机票预订服务的智能体。

在简单的演示环境中，可以直接初始化远程连接，例如文章中提供的代码示例：

from a2a_client.card_resolver import A2ACardResolver

for address in remote_agent_addresses:
    card_resolver = A2ACardResolver(address)
    try:
        card = card_resolver.get_agent_card()
        # The URL accessed here should be the same as the one provided in the agent card
        # However, in this demo we are using the URL provided in the key arguments
        remote_connection = RemoteAgentConnections(
            agent_card=card, agent_url=address
        )
        self.remote_agent_connections[card.name] = remote_connection
        self.cards[card.name] = card
    except httpx.ConnectError:
        print(f"ERROR: Failed to get agent card from : {address}")

这段代码展示了如何通过 A2ACardResolver 获取远程智能体的 AgentCard，并建立 RemoteAgentConnections 对象，用于后续的通信。

核心机制：Task – 智能体之间的协作单元

Task (任务) 是 A2A 协议中的核心工作单元，它代表了客户端智能体和远程智能体之间的协作过程。与简单的 API 调用不同，A2A 任务具有复杂的状态生命周期，能够模拟现实世界中复杂的协作场景。

Task 的状态可以包括：

submitted (已提交)：初始请求已发送。
working (工作中)：正在积极处理。
input-required (需要输入)：需要额外的输入信息，允许智能体之间的交互对话。
completed (已完成)：成功完成任务并返回结果。
failed (失败)：任务无法完成。
canceled (已取消)：任务被显式终止。

这种状态管理机制使得 A2A 协议能够处理复杂的、需要多轮交互的任务。例如，在预订机票的任务中，客户端智能体可能需要先询问用户的偏好（例如，出发时间、航空公司等），然后才能向远程智能体发送最终的预订请求。

通信模式：Synchronous vs. Asynchronous – 满足不同场景需求

A2A 协议支持两种主要的通信模式：Synchronous (同步) 和 Asynchronous (异步)。

Synchronous (同步)：客户端智能体发送请求后，会等待远程智能体处理完成并返回响应。这种模式适用于快速完成的任务，例如简单的信息查询。A2A 协议使用 HTTP(S) 协议和 JSON-RPC 2.0 消息格式来实现同步通信。
Asynchronous (异步)：客户端智能体发送请求后，无需等待远程智能体立即返回响应。远程智能体可以在任务处理过程中，通过 Server-Sent Events (SSE) 或 push notifications (推送通知) 向客户端智能体发送实时更新。这种模式适用于长时间运行的任务，例如视频处理或文件上传。

对于推送通知，A2A 协议允许客户端智能体提供一个 HTTPS webhook URL，远程智能体可以在任务状态发生变化时，通过该 URL 向客户端智能体发送通知。为了保证安全性，客户端智能体可以提供一个 token (令牌)，远程智能体需要在通知请求中包含该令牌，以便客户端智能体进行验证。

安全性：Authentication schemes and security model – 企业级保障

A2A 协议内置了多种 authentication schemes (认证方案)，以满足企业级安全需求。这些认证方案包括：

Basic Authentication (基本认证)：使用用户名和密码进行认证。
API Key (API 密钥)：使用预先生成的 API 密钥进行认证。
OAuth2/Bearer/JWT Token (OAuth2/Bearer/JWT 令牌)：使用 OAuth2 协议获取的令牌进行认证。

每个智能体的 AgentCard 中都会列出其支持的认证方法，客户端智能体需要根据 AgentCard 中的信息，选择合适的认证方式进行连接。例如，如果一个智能体只支持 OAuth2 认证，那么客户端智能体需要先通过 OAuth2 流程获取访问令牌，才能访问该智能体。

代码示例：Send Task to Remote Agent – 发送任务到远程智能体

以下代码示例展示了如何使用 A2A 客户端库向远程智能体发送任务：

from a2a_client.client import A2AClient
from a2a_types import TaskSendParams, Message, TextPart

async def send_task_to_remote_agent(client: A2AClient, task: str, session_id: str, task_id: str):
    request: TaskSendParams = TaskSendParams(
        id=task_id,
        sessionId=session_id,
        message=Message(
            role="user",
            parts=[TextPart(text=task)],
            metadata={},
        ),
        acceptedOutputModes=["text", "text/plain"],
        metadata={"conversation_id": session_id},
    )
    task = await client.send_task(request, self.task_callback)
    return task

这段代码首先创建了一个 TaskSendParams 对象，其中包含了任务的 ID、会话 ID、消息内容以及其他元数据。然后，它调用 A2AClient.send_task 方法将任务发送到远程智能体。task_callback 函数用于处理远程智能体返回的响应。

Server Side: 模拟 A2A 服务器 – 暴露智能体能力

A2A 服务器需要暴露其 AgentCard，并处理来自客户端智能体的请求。以下代码示例展示了如何使用 Starlette 框架创建一个简单的 A2A 服务器：

from starlette.applications import Starlette
from starlette.routing import Route
from starlette.responses import JSONResponse

async def process_request(request):
    # Verify authentication scheme
    # Based on the request (GetTaskRequest, SendTaskRequest, CancelTaskRequest, ...)
    # Pass request to the task manager
    return JSONResponse({"status": "OK"})

app = Starlette(debug=True, routes=[
    Route("/a2a", process_request, methods=["POST"])
])

这段代码创建了一个 Starlette 应用，并将 /a2a 路由映射到 process_request 函数。process_request 函数负责验证认证信息，并根据请求类型将其传递给任务管理器进行处理。

推送通知机制详解：Push Notification – 保证长时间任务的实时性

对于需要长时间运行的任务，A2A 协议提供了 push notification (推送通知) 机制，允许远程智能体在任务状态发生变化时，主动通知客户端智能体。

客户端智能体需要提供一个 PushNotificationConfig 对象，其中包含：

HTTPS webhook URL：接收推送通知的 URL。
token (令牌)：客户端智能体生成的令牌，用于验证推送通知的来源。
AuthenticationInfo (认证信息)：远程智能体用于认证客户端智能体 webhook URL 的信息。

远程智能体需要在发送推送通知时，将令牌包含在请求头中（例如，X-A2A-Notification-Token），以便客户端智能体进行验证。

Task Manager 的作用：任务管理的核心

Task Manager (任务管理器) 负责管理和维护任务的生命周期。它需要实现以下功能：

任务队列管理：接收和管理来自客户端智能体的任务请求。
任务状态更新：跟踪任务的状态变化，例如从 “working” 到 “completed”。
推送通知触发：在任务状态发生变化时，触发推送通知。
任务结果存储：存储任务的执行结果，以便客户端智能体后续查询。

一个简单的任务管理器可以使用内存存储来保存任务信息，例如 InMemoryTaskManager。对于生产环境，建议使用持久化存储（例如，数据库）来保证任务的可靠性。

总结与展望：A2A 的未来

Agent2Agent 协议 (A2A) 为大模型技术驱动的 AI 智能体互操作提供了一个标准化的解决方案。通过 AgentCard、Task、同步/异步通信模式以及安全认证机制，A2A 协议能够让不同的 AI 智能体像人类一样协同工作，共同完成复杂的任务。

随着大模型技术的不断发展，AI 智能体的数量和种类将会越来越多。A2A 协议的普及将有助于构建一个更加开放、互联互通的 AI 智能体生态，从而释放大模型技术的全部潜力。未来，A2A 协议可能会进一步扩展，例如支持更多的通信协议、认证方案以及任务类型。同时，也需要关注 A2A 协议的安全性和隐私性问题，例如如何防止恶意智能体的攻击，以及如何保护用户数据的安全。

Google Agent2Agent 协议 (A2A): 大模型时代 AI 智能体互操作的基石