快速入门：通过 A2A 暴露远程智能体¶
Supported in ADKPythonExperimental
本快速入门涵盖了任何开发者最常见的起点："如何暴露我的 ADK 智能体，以便其他智能体可以通过 A2A 使用它？"。这对于构建复杂的多智能体系统至关重要，其中不同的智能体需要协作和交互。
概述¶
本示例演示了智能体开发工具包（ADK）中的智能体到智能体（A2A）架构，展示了多个智能体如何协同工作来处理复杂任务。该示例实现了一个可以滚动骰子并检查数字是否为质数的智能体系统。
┌─────────────────┐ ┌──────────────────┐ ┌────────────────────┐
│ 根智能体 │───▶│ 滚动智能体 │ │ 远程质数 │
│ (本地) │ │ (本地) │ │ 智能体 │
│ │ │ │ │ (localhost:8001) │
│ │───▶│ │◀───│ │
└─────────────────┘ └──────────────────┘ └────────────────────┘
A2A 基础示例包含：
- 根智能体（
root_agent）：将任务委托给专门子智能体的主协调器 - 滚动智能体（
roll_agent）：处理骰子滚动操作的本地子智能体 - 质数智能体（
prime_agent）：检查数字是否为质数的远程 A2A 智能体，此智能体在单独的 A2A 服务器上运行
使用 ADK 服务器暴露你的智能体¶
ADK 提供了一个内置的 CLI 命令 adk api_server --a2a 来使用 A2A 协议暴露你的智能体。
# 你的智能体代码
root_agent = Agent(
model='gemini-flash-latest',
name='hello_world_agent',
<...你的智能体代码...>
)
1. 使用代码暴露智能体¶
from google.adk.a2a.utils.agent_to_a2a import to_a2a
# 使你的智能体兼容 A2A
a2a_app = to_a2a(root_agent, port=8001)
to_a2a() 函数甚至会在后台自动生成一个智能体卡片，通过从 ADK 智能体提取技能、能力和元数据，以便在使用 uvicorn 提供智能体端点时，众所周知的智能体卡片可用。
你也可以通过 agent_card 参数提供自己的智能体卡片。该值可以是一个 AgentCard 对象或指向智能体卡片 JSON 文件的路径。
使用 AgentCard 对象的示例：
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from a2a.types import AgentCard
# 定义 A2A 智能体卡片
my_agent_card = AgentCard(
name="file_agent",
url="http://example.com",
description="来自文件的测试智能体",
version="1.0.0",
capabilities={},
skills=[],
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
supports_authenticated_extended_card=False,
)
a2a_app = to_a2a(root_agent, port=8001, agent_card=my_agent_card)
使用 JSON 文件路径的示例：
from google.adk.a2a.utils.agent_to_a2a import to_a2a
# 从文件加载 A2A 智能体卡片
a2a_app = to_a2a(root_agent, port=8001, agent_card="/path/to/your/agent-card.json")
深入了解：to_a2a() 方法¶
当你调用 to_a2a() 时，ADK 会自动处理多个设置步骤来暴露你的智能体：
- A2aAgentExecutor 设置：
A2aAgentExecutor充当 A2A 协议和你的 ADK 智能体之间的桥梁。如果你不提供自定义Runner，它会自动创建一个由内存服务支持的默认运行器（用于制品、会话、记忆和凭据）。 - 状态管理： 创建
InMemoryTaskStore来跟踪 A2A 任务，以及InMemoryPushNotificationConfigStore用于处理推送通知。 - 请求处理： 创建
DefaultRequestHandler将传入的 A2A HTTP 请求路由到A2aAgentExecutor和状态存储。 - Starlette 应用和智能体卡片： 创建 Starlette 应用程序。在启动阶段，它要么加载你提供的智能体卡片，要么使用
AgentCardBuilder从你的智能体配置自动构建一个。然后挂载所有必要的 A2A API 路由。
参数¶
root_agent（必填）： 你想要通过 A2A 协议暴露的主要 ADK 智能体实例。port（可选）： 应用程序运行的端口号。push_config_store（可选）： 用于管理 A2A 推送通知的自定义存储实现。如果未提供，系统默认使用内存存储（InMemoryPushNotificationConfigStore）。agent_card（可选）： 一个AgentCard对象或 JSON 文件的路径。如果省略，ADK 会从你的智能体代码自动生成智能体卡片。
获取示例代码¶
首先，确保你已经安装了必要的依赖项：
你可以克隆并导航到 a2a_root 示例：
正如你将看到的，文件夹结构如下：
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])：质数检查算法
启动远程 A2A 智能体服务器¶
为了展示你的 ADK 智能体如何通过 A2A 消费 (Consuming) 远程智能体，你首先需要启动一个服务器，它将托管质数智能体（位于 check_prime_agent 目录下）。
# 启动 A2A 服务器，在 8001 端口上提供 check_prime_agent 服务
adk api_server --a2a --port 8001 contributing/samples/a2a_basic/remote_a2a
使用 --log_level debug 查看详细日志
要启用调试级别的日志记录，你可以在 adk api_server 命令中添加 --log_level debug。这将在测试智能体时提供更丰富的调试信息。
为什么使用 8001 端口？
在本地测试时，暴露智能体（远程质数智能体）的 A2A 服务器端口必须与消费智能体的端口不同。默认情况下，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 已经提供了一个智能体卡片：
a2a_basic/remote_a2a/check_prime_agent/agent-card.json
{
"capabilities": {},
"defaultInputModes": ["text/plain"],
"defaultOutputModes": ["application/json"],
"description": "专门用于检查数字是否为质数的智能体。它可以高效地确定单个数字或列表的质数属性。",
"name": "check_prime_agent",
"skills": [
{
"id": "prime_checking",
"name": "Prime Number Checking",
"description": "使用高效的数学算法检查列表中的数字是否为质数",
"tags": ["mathematical", "computation", "prime", "numbers"]
}
],
"url": "http://localhost:8001/a2a/check_prime_agent",
"version": "1.0.0"
}
检查远程智能体是否正在运行¶
你可以通过访问之前作为 to_a2a() 函数一部分自动生成的智能体卡片来检查你的智能体是否已启动并正在运行：
http://localhost:8001/.well-known/agent-card.json
你应该能看到智能体卡片的内容，看起来应该类似于：
{
"capabilities": {},
"defaultInputModes": ["text/plain"],
"defaultOutputModes": ["text/plain"],
"description": "hello world agent that can roll a dice of 8 sides and check prime numbers.",
"name": "hello_world_agent",
"protocolVersion": "0.2.6",
"skills": [
{
"description": "hello world agent that can roll a dice of 8 sides and check prime numbers. \n I roll dice and answer questions about the outcome of the dice rolls.\n I can roll dice of different sizes.\n I can use multiple tools in parallel by calling functions in parallel(in one request and in one round).\n It is ok to discuss previous dice roles, and comment on the dice rolls.\n When I are asked to roll a die, I must call the roll_die tool with the number of sides. Be sure to pass in an integer. Do not pass in a string.\n I should never roll a die on my own.\n When checking prime numbers, call the check_prime tool with a list of integers. Be sure to pass in a list of integers. I should never pass in a string.\n I should not check prime numbers before calling the tool.\n When I are asked to roll a die and check prime numbers, I should always make the following two function calls:\n 1. I should first call the roll_die tool to get a roll. Wait for the function response before calling the check_prime tool.\n 2. After I get the function response from roll_die tool, I should call the check_prime tool with the roll_die result.\n 2.1 If user asks I to check primes based on previous rolls, make sure I include the previous rolls in the list.\n 3. When I respond, I must include the roll_die result from step 1.\n I should always perform the previous 3 steps when asking for a roll and checking prime numbers.\n I should not rely on the previous history on prime results.\n ",
"id": "hello_world_agent",
"name": "model",
"tags": ["llm"]
},
{
"description": "Roll a die and return the rolled result.\n\nArgs:\n sides: The integer number of sides the die has.\n tool_context: the tool context\nReturns:\n An integer of the result of rolling the die.",
"id": "hello_world_agent-roll_die",
"name": "roll_die",
"tags": ["llm", "tools"]
},
{
"description": "Check if a given list of numbers are prime.\n\nArgs:\n nums: The list of numbers to check.\n\nReturns:\n A str indicating which number is prime.",
"id": "hello_world_agent-check_prime",
"name": "check_prime",
"tags": ["llm", "tools"]
}
],
"supportsAuthenticatedExtendedCard": false,
"url": "http://localhost:8001",
"version": "0.0.1"
}
运行主（消费）智能体¶
现在你的远程智能体正在运行，你可以启动开发 UI 并选择 "a2a_root" 作为你的智能体。
工作原理¶
主智能体使用 RemoteA2aAgent() 来消费远程智能体。
a2a_basic/agent.py
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,
)
注意
设置 use_legacy=False 会启用 A2A 扩展。
随后，你只需将 prime_agent 注册为 root_agent 的子智能体即可：
root_agent = Agent(
model="gemini-2.0-flash",
name="root_agent",
instruction="...",
sub_agents=[roll_agent, prime_agent],
)
高级配置：自定义转换器与拦截器¶
在内部，RemoteA2aAgent 在 A2A 协议格式与 ADK 的原生 Event 系统之间执行转换。你可以通过 config 参数传入 A2aRemoteAgentConfig 来自定义此行为。
转换器¶
a2a_message_converter：转换 A2A 消息。a2a_task_converter：转换 A2A 任务。a2a_part_converter：底层 Parts 转换。
请求拦截器¶
before_request：请求处理前执行。after_request：请求处理后执行。
交互示例¶
一旦你的主智能体和远程智能体都在运行，你就可以观察到它们之间的 A2A 调用：
质数检查：
组合操作：
下一步¶
- A2A 快速入门（暴露）：了解如何暴露你的现有智能体。
- A2A 快速入门（消费）Go：学习如何使用 Go 语言实现消费。