智能体到智能体（A2A）协议

ADK-Rust 实现了用于跨网络智能体通信的 A2A 协议 v1.0.0。该实现位于 adk-server 中，受 a2a-v1 功能标志控制，并涵盖全部 11 个 JSON-RPC 操作、REST 绑定、智能体卡发现以及版本协商。关于语义比规范允许范围更窄的那一个操作，请参见 操作覆盖范围。线协议类型由 a2a-protocol-types 提供——这是由基金会验证的 Rust A2A SDK，由 @tomtom215 编写（a2a-rust）。

概览

A2A 在以下场景中很有用：

• 与第三方智能体服务集成
• 构建带有专用智能体的微服务架构
• 启用跨语言智能体通信（任何带有 A2A 客户端的语言都可以）
• 在智能体系统之间强制执行正式契约

对于简单的内部组织，建议使用本地子智能体而不是 A2A，以获得更好的性能。

v1.0.0 兼容性

该实现完全符合 A2A 协议 v1.0.0 规范：

Agent 卡片

每个 A2A agent 都会在 /.well-known/agent-card.json 公开一个 agent 卡片，用于描述其能力、技能和支持的接口。

use adk_server::a2a::v1::card::build_v1_agent_card;
use a2a_protocol_types::{AgentCapabilities, AgentSkill};

let card = build_v1_agent_card(
    "my-agent",
    "A helpful research agent",
    "http://localhost:3001/jsonrpc",
    "1.0.0",
    vec![AgentSkill {
        id: "research".to_string(),
        name: "Research & Summarize".to_string(),
        description: "Researches topics and produces structured summaries".to_string(),
        tags: vec!["research".to_string()],
        examples: None,
        input_modes: None,
        output_modes: None,
        security_requirements: None,
    }],
    AgentCapabilities::none()
        .with_streaming(true)
        .with_push_notifications(true),
);

agent 卡片包括：

• Agent 名称、描述和版本
• 支持的接口，以及协议绑定和版本
• 能力：streaming、pushNotifications、extendedAgentCard
• 从 agent 配置中派生的技能
• 默认输入/输出模式

现在，能力是通过 AgentCapabilities 参数显式声明的——不再有硬编码的默认值。

通过 A2A v1 公开一个 Agent

构建一个完整的 A2A v1.0.0 server，并集成 LLM：

use std::sync::Arc;
use a2a_protocol_types::{AgentCapabilities, AgentSkill};
use adk_agent::LlmAgentBuilder;
use adk_server::a2a::v1::card::{CachedAgentCard, build_v1_agent_card};
use adk_server::a2a::v1::executor::V1Executor;
use adk_server::a2a::v1::jsonrpc_handler::jsonrpc_handler;
use adk_server::a2a::v1::push::NoOpPushNotificationSender;
use adk_server::a2a::v1::request_handler::RequestHandler;
use adk_server::a2a::v1::rest_handler::rest_router;
use adk_server::a2a::v1::task_store::InMemoryTaskStore;
use adk_server::a2a::v1::version::version_negotiation;
use adk_runner::RunnerConfig;
use adk_session::InMemorySessionService;
use axum::Router;
use axum::routing::post;
use tokio::sync::RwLock;

// 1. Create your agent
let model = adk_model::GeminiModel::new(&api_key, "gemini-2.5-flash")?;
let agent = LlmAgentBuilder::new("my-agent")
    .description("A helpful agent")
    .model(Arc::new(model))
    .instruction("You are a helpful assistant.")
    .build()?;

// 2. Set up A2A infrastructure
let task_store = Arc::new(InMemoryTaskStore::new());
let executor = Arc::new(V1Executor::new(task_store.clone()));
let push_sender = Arc::new(NoOpPushNotificationSender);

// 3. Build agent card with capabilities
let card = build_v1_agent_card(
    "my-agent", "A helpful agent",
    "http://localhost:3001/jsonrpc", "1.0.0",
    vec![/* skills */],
    AgentCapabilities::none().with_streaming(true),
);
let cached_card = Arc::new(RwLock::new(CachedAgentCard::new(card)));

// 4. Create runner config for LLM invocation
let session_service = Arc::new(InMemorySessionService::new());
let runner_config = Arc::new(RunnerConfig {
    app_name: "my-agent".to_string(),
    agent: Arc::new(agent),
    session_service,
    artifact_service: None,
    memory_service: None,
    plugin_manager: None,
    run_config: None,
    compaction_config: None,
    context_cache_config: None,
    cache_capable: None,
    request_context: None,
    cancellation_token: None,
});

// 5. Wire up the handler and routes
let handler = Arc::new(RequestHandler::with_runner(
    executor, task_store, push_sender, cached_card, runner_config,
));

let app = Router::new()
    .route("/jsonrpc", post(jsonrpc_handler))
    .with_state(handler.clone())
    .merge(rest_router(handler))
    .layer(axum::middleware::from_fn(version_negotiation));

// 6. Serve
let listener = tokio::net::TcpListener::bind("0.0.0.0:3001").await?;
axum::serve(listener, app).await?;

这会公开：

• GET /.well-known/agent-card.json — 带有 ETag 缓存的 Agent 卡片
• POST /jsonrpc — JSON-RPC 端点（全部 11 个 v1 操作；见 Operation coverage）
• REST 路由，覆盖所有操作
• 所有路由上的 A2A-Version header 协商

JSON-RPC Operations

支持全部 11 个 A2A v1.0.0 operations：

SendMessage

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "SendMessage",
  "params": {
    "message": {
      "messageId": "msg-123",
      "role": "ROLE_USER",
      "parts": [{"text": "Research quantum computing"}]
    }
  }
}

响应包含一个 Task 对象，其中包括 status、history 和 artifacts。该响应使用 Content-Type: application/a2a+json。

SendStreamingMessage

与 SendMessage 相同的请求格式。返回一个 SSE 流，其中：

1. 第一个事件是一个完整的 Task 对象（根据规范 §3.1.2）
2. 后续事件是 TaskStatusUpdateEvent（Working、Completed 等）
3. Artifact 事件是 TaskArtifactUpdateEvent

多轮对话

当任务进入 INPUT_REQUIRED 状态时，使用相同的 contextId 发送后续消息以恢复它：

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "SendMessage",
  "params": {
    "message": {
      "messageId": "msg-456",
      "role": "ROLE_USER",
      "contextId": "ctx-original",
      "parts": [{"text": "Yes, include more details on error correction"}]
    }
  }
}

处理程序会自动通过 contextId 找到现有任务，将其从 INPUT_REQUIRED 转换为 Working，将新消息追加到 history，并继续处理。

幂等性

带有相同 messageId 的重复 SendMessage 请求会返回先前创建的任务，而不会重新处理。这同时适用于 SendMessage 和 SendStreamingMessage。

推送通知认证

当客户端通过 CreateTaskPushNotificationConfig 注册 webhook 时，服务器会在 webhook 投递中包含认证头：

• Authorization: Bearer <credentials> — 当 authentication 字段具有 bearer 凭证时
• a2a-notification-token: <token> — 当 token 字段存在时

两个头可以同时设置。SSRF 防护会将 webhook URLs 验证为私有 IP 地址范围和 localhost。

输入验证

所有传入请求在处理前都会经过验证：

消费远程代理

使用 RemoteA2aAgent 与远程 A2A 代理通信：

use adk_server::a2a::RemoteA2aAgent;

let remote_agent = RemoteA2aAgent::builder("prime_checker")
    .description("Checks if numbers are prime")
    .agent_url("http://localhost:8001")
    .build()?;

// Use as a sub-agent in a local agent hierarchy
let root_agent = LlmAgentBuilder::new("root")
    .model(Arc::new(model))
    .sub_agent(Arc::new(remote_agent))
    .build()?;

A2A 客户端

用于直接的协议级通信：

use adk_server::a2a::client::v1_client::A2aV1Client;

// Discover agent card
let card = A2aV1Client::resolve_agent_card("http://localhost:3001").await?;
let client = A2aV1Client::new(card);

// Send message
let task = client.send_message(message).await?;

// Get task
let task = client.get_task(&task_id, Some(10)).await?;

// List tasks
let tasks = client.list_tasks(None, None, None, None).await?;

// Cancel task
client.cancel_task(&task_id).await?;

// Streaming
let response = client.send_streaming_message(message).await?;

// Push notification CRUD
let config = client.create_push_notification_config(config).await?;
client.delete_push_notification_config(&task_id, &config_id).await?;

错误处理

A2A 错误会映射到 JSON-RPC 代码和 HTTP 状态码：

运行示例

包含两个完整的 A2A v1.0.0 示例 agent：

cargo run --manifest-path examples/a2a-research-agent/Cargo.toml
cargo run --manifest-path examples/a2a-writing-agent/Cargo.toml --bin a2a-writing-agent
cargo run --manifest-path examples/a2a-writing-agent/Cargo.toml --bin client

客户端会验证：agent card 发现、SendMessage（两个带有真实 LLM 的 agent）、GetTask、ListTasks、CancelTask 错误路径、SendStreamingMessage、推送通知 CRUD、GetExtendedAgentCard、版本协商，以及错误路径。

最佳实践

1. 准确声明能力 — 根据你的 agent 实际支持的内容设置 streaming、pushNotifications
2. 长操作使用流式传输 — SendStreamingMessage 为客户端提供实时进度
3. 处理多轮流程 — 使用 contextId 在消息之间维护会话状态
4. 验证 webhook URLs — 内置了 SSRF 保护，但在生产环境中使用 HTTPS
5. 设置适当的超时 — 为远程 agent 调用配置请求超时
6. 使用幂等性 — 客户端可以安全地使用相同的 messageId 重试 SendMessage

相关内容

• LlmAgent — 创建 agent
• 多 agent 系统 — 子 agent 和层级结构
• 服务器部署 — 将 agent 作为 HTTP 服务器运行

---

上一个: ← 服务器 | 下一个: 评估 →

操作覆盖范围

全部 11 个 v1 JSON-RPC 操作都已分派并实现。

SubscribeToTask 是一个快照

该操作会返回任务及其当前状态，然后关闭流。它不会 传递后续更新，因此客户端不能等待它来获取进度。

实时重新连接需要一个按任务划分的事件队列，并且该队列必须比发起请求的生命周期更长。参考实现通过它们的 A2A SDKs 获取这一能力——adk-python 和 adk-go 都将 tasks/resubscribe 完全委托给 SDK 的队列管理器，而二者都没有在 ADK 代码中实现它。这个服务器是手写的，基于 a2a-protocol-types，它提供的是线上的类型而不是服务器运行时，因此队列目前还不存在。

当需要实时更新时，请使用 SendStreamingMessage。

流式事件契约

SendStreamingMessage 会在代理事件到达时对其进行转换：

同一响应的所有块共享一个 artifact ID，以便客户端可以重新组装它们。连接后的文本会被持久化，因此后续的 GetTask 会返回已流式传输的内容。这符合 adk-python 和 adk-go 在其 SDKs 上实现的契约。