快速入门：通过 A2A 消费远程智能体¶
本快速入门涵盖了任何开发者最常见的起点：“有一个远程智能体，如何让我的 ADK 智能体通过 A2A 使用它？”。这对于构建复杂的多智能体系统至关重要，其中不同的智能体需要协作和交互。
概述¶
本示例演示了智能体开发工具包（ADK）中的智能体到智能体（A2A）架构，展示了多个智能体如何协同工作来处理复杂任务。该示例实现了一个可以滚动骰子并检查数字是否为质数的智能体。
┌─────────────────┐ ┌──────────────────┐ ┌────────────────────┐
│ 根智能体 │───▶│ 滚动智能体 │ │ 远程质数 │
│ (本地) │ │ (本地) │ │ 智能体 │
│ │ │ │ │ (localhost:8001) │
│ │───▶│ │◀───│ │
└─────────────────┘ └──────────────────┘ └────────────────────┘
A2A 基础示例包含：
- 根智能体（
root_agent）：将任务委托给专门子智能体的主协调器 - 滚动智能体（
roll_agent）：处理骰子滚动操作的本地子智能体 - 质数智能体（
prime_agent）：检查数字是否为质数的远程 A2A 智能体，此智能体在单独的 A2A 服务器上运行
使用 ADK 服务器暴露你的智能体¶
ADK 提供了一个内置的 CLI 命令 adk api_server --a2a 来使用 A2A 协议暴露你的智能体。
在 a2a_basic 示例中，你首先需要通过 A2A 服务器暴露 check_prime_agent，以便本地根智能体可以使用它。
1. 获取示例代码¶
首先，确保安装了必要的依赖项：
你可以克隆并导航到 a2a_basic 示例：
正如你将看到的，文件夹结构如下：
a2a_basic/
├── remote_a2a/
│ └── check_prime_agent/
│ ├── __init__.py
│ ├── agent.json
│ └── agent.py
├── README.md
├── __init__.py
└── agent.py # 本地根智能体
主智能体 (a2a_basic/agent.py)¶
roll_die(sides: int)：用于滚动骰子的函数工具roll_agent：专门从事骰子滚动的本地智能体prime_agent：远程 A2A 智能体配置root_agent：具有委托逻辑的主协调器
远程质数智能体 (a2a_basic/remote_a2a/check_prime_agent/)¶
agent.py：质数检查服务的实现agent.json：A2A 智能体的智能体卡片check_prime(nums: list[int])：质数检查算法
2. 启动远程质数智能体服务器¶
为了展示你的 ADK 智能体如何通过 A2A 消费（Consuming）远程智能体，你首先需要启动一个远程智能体服务器，它将托管质数智能体（在 check_prime_agent 下）。
# 启动在端口 8001 上提供 check_prime_agent 的远程 a2a 服务器
adk api_server --a2a --port 8001 contributing/samples/a2a_basic/remote_a2a
使用 --log_level debug 添加日志进行调试
要启用调试级日志记录，你可以在 adk api_server 中添加 --log_level debug，如下所示：
为什么使用端口 8001？
在本快速入门中，当在本地测试时，你的智能体将使用 localhost，因此暴露智能体（远程质数智能体）的 A2A 服务器的 port 必须与消费智能体的端口不同。你将与之交互的消费智能体的 adk web 的默认端口是 8000，这就是为什么 A2A 服务器使用单独的端口 8001 创建的原因。
执行后，你应该看到类似以下内容：
INFO: Started server process [56558]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8001 (Press CTRL+C to quit)
3. 查找远程智能体所需的智能体卡片 (agent-card.json)¶
A2A 协议要求每个智能体都必须有一个描述其功能的智能体卡片。
如果其他人已经构建了你想要在你的智能体中消费（Consuming）的远程 A2A 智能体，那么你应该确认他们有一个智能体卡片（agent-card.json）。
在示例中，check_prime_agent 已经提供了一个智能体卡片：
{
"capabilities": {},
"defaultInputModes": ["text/plain"],
"defaultOutputModes": ["application/json"],
"description": "An agent specialized in checking whether numbers are prime. It can efficiently determine the primality of individual numbers or lists of numbers.",
"name": "check_prime_agent",
"skills": [
{
"id": "prime_checking",
"name": "Prime Number Checking",
"description": "Check if numbers in a list are prime using efficient mathematical algorithms",
"tags": ["mathematical", "computation", "prime", "numbers"]
}
],
"url": "http://localhost:8001/a2a/check_prime_agent",
"version": "1.0.0"
}
关于 ADK 中智能体卡片的更多信息
在 ADK 中，你可以使用 to_a2a(root_agent) 包装器，它会自动为你生成智能体卡片。如果你有兴趣了解更多关于如何暴露现有智能体以便其他人可以使用它的信息，请查看 A2A 快速入门（暴露） 教程。
4. 运行主（消费）智能体¶
工作原理¶
主智能体使用 RemoteA2aAgent() 函数来消费远程智能体（在我们的示例中为 prime_agent）。正如你在下面看到的，RemoteA2aAgent() 需要 name、description 和 agent_card 的 URL。
# ...省略部分代码...
from google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
# 配置远程 A2A 智能体
prime_agent = RemoteA2aAgent(
name="prime_agent",
description="处理质数检查任务的智能体。",
agent_card=(
f"http://localhost:8001/a2a/check_prime_agent{AGENT_CARD_WELL_KNOWN_PATH}"
),
use_legacy=False,
)
# ...省略部分代码...
使用新的 A2A 集成
通过设置 use_legacy=False，智能体将使用新的 ADK-A2A 集成，因为它将发送 A2A 扩展 给远程智能体。
然后，你只需在智能体中使用 RemoteA2aAgent。在本例中，prime_agent 被作为下面 root_agent 的子智能体之一使用：
from google.adk.agents.llm_agent import Agent
from google.genai import types
root_agent = Agent(
model="gemini-2.0-flash",
name="root_agent",
instruction="""
你是一个得力的助手，可以滚动骰子并检查数字是否为质数。
你将滚动骰子的任务委托给 roll_agent，将质数检查任务委托给 prime_agent。
请遵循以下步骤：
1. 如果用户要求滚动骰子，请委托给 roll_agent。
2. 如果用户要求检查质数，请委托给 prime_agent。
3. 如果用户要求先滚动骰子然后检查结果是否为质数，请先调用 roll_agent，然后将结果传递给 prime_agent。
在继续之前，请务必澄清结果。
""",
global_instruction=(
"你是 DicePrimeBot，随时准备滚动骰子并检查质数。"
),
sub_agents=[roll_agent, prime_agent],
tools=[example_tool],
generate_content_config=types.GenerateContentConfig(
safety_settings=[
types.SafetySetting( # 避免关于滚动骰子的虚假警报
category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
threshold=types.HarmBlockThreshold.OFF,
),
]
),
)
高级配置：自定义转换器与拦截器¶
在内部，RemoteA2aAgent 在 A2A 协议格式与 ADK 的原生 Event 系统之间进行相互转换。你可以通过向 RemoteA2aAgent 的 config 参数传递一个 A2aRemoteAgentConfig 对象来自定义此行为。
这允许你定义自定义类型映射、注入请求参数以及拦截请求或响应。
转换器¶
转换器负责将传入的 A2A 响应翻译为原生的 ADK 对象。你可以为以下钩子提供自己的映射函数：
a2a_message_converter：将标准 A2A 消息转换为 ADKEvent对象。a2a_task_converter：将 A2A 任务转换为 ADKEvent。a2a_status_update_converter：将 A2ATaskStatusUpdateEvent转换为 ADKEvent对象。a2a_artifact_update_converter：将 A2ATaskArtifactUpdateEvent转换为 ADKEvent对象。a2a_part_converter：一个基础的底层钩子，由其他转换器在内部使用，用于将单个 A2A 消息部分 (Parts) 转换为 GenAIPart对象。
注意
这些自定义客户端转换器仅在响应来自 智能体执行器 (agent executor) 的新实现时才会使用。有关更多详细信息，请参见 A2A 扩展。
请求拦截器¶
你可以注入一个 request_interceptors 列表，以为 A2A 请求添加中间件逻辑：
before_request：在智能体开始处理之前执行。你可以修改A2AMessage，或返回一个 ADKEvent以立即中止请求并将该事件返回给调用者。after_request：在智能体处理完请求后执行。你可以修改生成的 ADKEvent，或者返回None以过滤掉并完全丢弃该事件。
请求参数配置¶
通过拦截器，你还可以修改 A2A 请求的 ParametersConfig 以注入：
request_metadata：将自定义元数据字典传入请求头。client_call_context：为底层传输注入特定的客户端调用上下文。
# ...省略部分代码...
from google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
# 带高级配置的远程 A2A 智能体
prime_agent = RemoteA2aAgent(
name="prime_agent",
description="处理质数检查任务的智能体。",
agent_card=(
f"http://localhost:8001/a2a/check_prime_agent{AGENT_CARD_WELL_KNOWN_PATH}"
),
use_legacy=False,
config=A2aRemoteAgentConfig(
a2a_message_converter=my_a2a_message_converter,
request_interceptors=[my_request_interceptor],
),
)
# ...省略部分代码...
交互示例¶
一旦你的主智能体和远程智能体都在运行，你就可以与根智能体交互，看看它如何通过 A2A 调用远程智能体：
简单骰子滚动：
此交互使用本地子智能体 roll_agent：
质数检查：
此交互通过 A2A 使用远程智能体 prime_agent：
组合操作：
此交互同时使用本地滚动智能体和远程质数智能体：
下一步¶
现在你已经创建了一个通过 A2A 服务器使用远程智能体的智能体，下一步是学习如何从另一个智能体连接到它。
- A2A 快速入门（暴露）：了解如何暴露你的现有智能体，以便其他智能体通过 A2A 协议使用它。
- A2A 快速入门（消费）Go：了解如何使用 Go 消费 A2A 远程智能体。