快速入门：通过 A2A 暴露远程智能体¶
本快速入门涵盖了任何开发者最常见的起点："如何暴露我的 ADK 智能体，以便其他智能体可以通过 A2A 使用它？"。这对于构建复杂的多智能体系统至关重要，其中不同的智能体需要协作和交互。
A2A Python SDK 版本兼容性
ADK 的 A2A 集成同时兼容 A2A SDK 的两个主要版本
（a2a-sdk 0.3.x 和 1.x.x）。安装的 A2A SDK 版本会自动检测，
因此无需对你的 ADK 应用程序代码做任何修改。
尽管 a2a-sdk 0.3.x 在兼容模式下受支持，但新的
集成应以 1.x.x 为目标。如果你的代码直接引用了 a2a-sdk 类型
（例如自定义执行器或手动构建的 AgentCard
实例），请在迁移到 1.x.x 时参阅 A2A SDK v1.0 迁移
指南。
概览¶
本示例演示了智能体开发工具包（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',
<...your agent code...>
)
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 路由。
参数¶
agent（必填）： 你希望通过 A2A 协议暴露的主要 ADK 智能体实例。host（可选）： 用于构建生成的智能体卡片中公布的 A2A RPC URL 的主机。默认为"localhost"。protocol（可选）： 该 URL 中使用的协议。默认为"http"。port（可选）： 该 URL 中使用的端口。默认为8000。to_a2a()本身不会绑定端口，因此此值必须与你实际提供服务的端口匹配（参见下面的uvicorn --port标志），否则公布的智能体卡片将指向不可达的位置。push_config_store（可选）： 用于管理 A2A 推送通知的自定义存储实现。如果未提供，系统默认使用内存存储（InMemoryPushNotificationConfigStore）。agent_card（可选）： 一个AgentCard对象或指向 JSON 文件的路径。如果省略，ADK 会自动从你的智能体代码生成智能体卡片。runner（可选）： 一个预构建的Runner。如果省略，将创建一个由内存服务支持的默认运行器。
获取示例代码¶
首先，确保你已经安装了必要的依赖项：
你可以克隆并导航到 a2a_root 示例：
正如你将看到的，文件夹结构如下：
a2a_basic/
├── remote_a2a/
│ └── hello_world/
│ ├── __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 目录下）。
# 确保当前工作目录为 adk-python/
# 使用 uvicorn 启动远程智能体
uvicorn contributing.samples.a2a.a2a_root.remote_a2a.hello_world.agent:a2a_app --host localhost --port 8001
使用 --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 已经提供了一个智能体卡片：
{
"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() 来消费远程智能体。
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 扩展。
此交互通过 A2A 使用远程智能体——质数智能体：
组合操作：
此交互同时使用本地滚动智能体和远程质数智能体：
高级配置：自定义转换器与拦截器（暴露端）¶
在需要比 to_a2a() 提供的更细粒度控制的场景中，你可以实例化并直接将 A2aAgentExecutorConfig 传递给 A2aAgentExecutor。这允许你覆盖默认的数据转换器并注入执行中间件。
转换器¶
转换器负责在 A2A 协议载荷与 ADK 的原生 Event 或 Part 对象之间进行双向转换。你可以为以下钩子提供自己的映射函数：
a2a_part_converter：将 A2A 消息部分转换为 ADKPart对象。gen_ai_part_converter：将原生 ADKPart对象转换为 A2A 消息部分。request_converter：将传入的 A2A 请求转换为 ADKRunRequest。event_converter：（旧版） 将 ADK 事件转换为 A2A 事件，用于旧版执行器实现。adk_event_converter：（新版） 将 ADK 事件转换为 A2A 事件，用于新的更新版执行器实现。
执行拦截器¶
你可以注入一个 execute_interceptors 列表，以为 A2aAgentExecutor 载荷处理添加中间件逻辑：
before_agent：在智能体开始处理请求之前执行。允许你检查或修改传入的RequestContext。after_event：在 ADK 事件转换为 A2A 事件之后执行。允许你在事件入队前修改发出的事件，或返回None以过滤并完全丢弃该事件。after_agent：在智能体处理完成且最终事件准备好后执行。用于在发送之前检查或修改终端状态事件（例如completed或failed）。
抑制实验性警告
可以将 ADK_SUPPRESS_A2A_EXPERIMENTAL_FEATURE_WARNINGS 环境变量设置为 true，以抑制与实验性 A2A 功能相关的警告。这对于有意使用这些功能并希望获得更简洁日志的开发者很有用：
智能体执行器 V2¶
新版智能体执行器通常在客户端发送所需的 A2A 扩展时启用。
但是，你也可以绕过扩展，在实例化 A2aAgentExecutor 时通过设置 force_new_version=True 标志来强制服务器使用新版执行器。这允许你使用新的执行器逻辑，而无需修改现有客户端以发送扩展。
from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
executor = A2aAgentExecutor(
...,
force_new_version=True
)
高级配置：自定义转换器与拦截器¶
在内部，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 语言实现消费。