# 快速入门：通过 A2A 暴露远程智能体

Supported in ADKPythonExperimental

本快速入门涵盖了任何开发者最常见的起点：**"如何暴露我的 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 迁移 指南](https://github.com/a2aproject/a2a-python/tree/main/docs/migrations/v1_0)。

## 概览

本示例演示了智能体开发工具包（ADK）中的\*\*智能体到智能体（A2A）\*\*架构，展示了多个智能体如何协同工作来处理复杂任务。该示例实现了一个可以滚动骰子并检查数字是否为质数的智能体系统。

```text
┌─────────────────┐     ┌──────────────────┐     ┌────────────────────┐
│   根智能体       │───▶│   滚动智能体      │     │   远程质数          │
│    (本地)        │     │     (本地)        │     │   智能体            │
│                  │     │                   │     │   (localhost:8001)  │
│                  │───▶│                   │◀───│                     │
└─────────────────┘     └──────────────────┘     └────────────────────┘
```

A2A 基础示例包含：

- **根智能体**（`root_agent`）：将任务委托给专门子智能体的主协调器
- **滚动智能体**（`roll_agent`）：处理骰子滚动操作的本地子智能体
- **质数智能体**（`prime_agent`）：检查数字是否为质数的远程 A2A 智能体，此智能体在单独的 A2A 服务器上运行

## 使用 ADK 服务器暴露你的智能体

ADK 提供了一个内置的 CLI 命令 `adk api_server --a2a` 来使用 A2A 协议暴露你的智能体。

```python
# 你的智能体代码
root_agent = Agent(
    model='gemini-flash-latest',
    name='hello_world_agent',

    <...your agent code...>
)
```

### 1. 使用代码暴露智能体

```python
from google.adk.a2a.utils.agent_to_a2a import to_a2a

# 使你的智能体兼容 A2A
a2a_app = to_a2a(root_agent, port=8001)
```

`to_a2a()` 函数甚至会在后台自动生成一个智能体卡片，通过[从 ADK 智能体提取技能、能力和元数据](https://github.com/google/adk-python/blob/main/src/google/adk/a2a/utils/agent_card_builder.py)，以便在使用 `uvicorn` 提供智能体端点时，众所周知的智能体卡片可用。

你也可以通过 `agent_card` 参数提供自己的智能体卡片。该值可以是一个 `AgentCard` 对象或指向智能体卡片 JSON 文件的路径。

**使用 `AgentCard` 对象的示例：**

```python
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 文件路径的示例：**

```python
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`。如果省略，将创建一个由内存服务支持的默认运行器。
- **`lifespan`（可选）：** 用于 Starlette 生命周期事件的异步上下文管理器，用来运行启动和关闭逻辑。参见[管理应用程序生命周期](#manage-the-application-lifecycle)。

### 管理应用程序生命周期

使用 `to_a2a` 函数的 `lifespan` 参数来管理应用程序的生命周期。传入一个异步上下文管理器来执行设置和清理任务，例如在启动时建立数据库连接、在关闭时断开该连接。

上下文管理器会接收 `Starlette` 应用实例。使用 `app.state` 来存储你的应用程序需要全局访问的资源。

```python
from contextlib import asynccontextmanager
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from starlette.applications import Starlette


@asynccontextmanager
async def lifespan(app: Starlette):
  # 在应用程序启动时初始化资源
  app.state.db = await init_db()

  yield

  # 在应用程序关闭时清理资源
  await app.state.db.close()


a2a_app = to_a2a(agent, lifespan=lifespan)
```

### 获取示例代码

首先，确保你已经安装了必要的依赖项：

```bash
pip install google-adk[a2a]
```

你可以克隆并导航到 [**a2a_root** 示例](https://github.com/google/adk-python/tree/main/contributing/samples/a2a/a2a_root)：

```bash
git clone https://github.com/google/adk-python.git
```

正如你将看到的，文件夹结构如下：

```text
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` 目录下）。

```bash
# 确保当前工作目录为 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` 端口。

执行后，你应该看到类似以下内容：

```shell
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

```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>

你应该能看到智能体卡片的内容，看起来应该类似于：

```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" 作为你的智能体。

```bash
# 在另一个终端中，运行 adk web 服务器
adk web contributing/samples/a2a/
```

#### 工作原理

主智能体使用 `RemoteA2aAgent()` 来消费远程智能体。

a2a_basic/agent.py

```python
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 扩展](https://adk.wiki/a2a/a2a-extension/index.md)。

此交互通过 A2A 使用远程智能体——质数智能体：

```text
用户：7 是质数吗？
机器人：是的，7 是质数。
```

**组合操作：**

此交互同时使用本地滚动智能体和远程质数智能体：

```text
用户：滚动一个 10 面骰子并检查它是否是质数
机器人：我为你滚动了一个 8。
机器人：8 不是质数。
```

## 高级配置：自定义转换器与拦截器（暴露端）

在需要比 `to_a2a()` 提供的更细粒度控制的场景中，你可以实例化并直接将 [`A2aAgentExecutorConfig`](https://github.com/google/adk-python/blob/main/src/google/adk/a2a/executor/config.py) 传递给 `A2aAgentExecutor`。这允许你覆盖默认的数据转换器并注入执行中间件。

### 转换器

转换器负责在 A2A 协议载荷与 ADK 的原生 `Event` 或 `Part` 对象之间进行双向转换。你可以为以下钩子提供自己的映射函数：

- **`a2a_part_converter`**：将 A2A 消息部分转换为 ADK `Part` 对象。
- **`gen_ai_part_converter`**：将原生 ADK `Part` 对象转换为 A2A 消息部分。
- **`request_converter`**：将传入的 A2A 请求转换为 ADK `RunRequest`。
- **`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 功能相关的警告。这对于有意使用这些功能并希望获得更简洁日志的开发者很有用：

```bash
export ADK_SUPPRESS_A2A_EXPERIMENTAL_FEATURE_WARNINGS=true
```

### 管理应用程序生命周期

使用 `to_a2a` 函数的 `lifespan` 参数来管理应用程序的生命周期。传入一个异步上下文管理器来执行设置和清理任务，例如在启动时建立数据库连接、在关闭时断开该连接。

上下文管理器会接收 `Starlette` 应用实例。使用 `app.state` 来存储你的应用程序需要全局访问的资源。

```python
from contextlib import asynccontextmanager
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from starlette.applications import Starlette


@asynccontextmanager
async def lifespan(app: Starlette):
  # 在应用程序启动时初始化资源
  app.state.db = await init_db()

  yield

  # 在应用程序关闭时清理资源
  await app.state.db.close()


a2a_app = to_a2a(agent, lifespan=lifespan)
```

## 智能体执行器 V2

新版[智能体执行器](https://github.com/google/adk-python/blob/main/src/google/adk/a2a/executor/a2a_agent_executor_impl.py)通常在客户端发送所需的 [A2A 扩展](https://adk.wiki/a2a/a2a-extension/index.md)时启用。

但是，你也可以绕过扩展，在实例化 `A2aAgentExecutor` 时通过设置 `force_new_version=True` 标志来强制服务器使用新版执行器。这允许你使用新的执行器逻辑，而无需修改现有客户端以发送扩展。

```python
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 调用：

**质数检查：**

```text
用户：7 是质数吗？
机器人：是的，7 是质数。
```

**组合操作：**

```text
用户：滚动一个 10 面骰子并检查它是否是质数
机器人：我为你滚动了一个 8。
机器人：8 不是质数。
```

## 下一步

- [**A2A 快速入门（暴露）**](https://adk.wiki/a2a/quickstart-exposing/index.md)：了解如何暴露你的现有智能体。
- [**A2A 快速入门（消费）Go**](https://adk.wiki/a2a/quickstart-consuming-go/index.md)：学习如何使用 Go 语言实现消费。