开发 Agent2Agent 代理

本页介绍了如何开发和测试 Agent2Agent (A2A) 智能体。A2A 协议是一种开放标准，旨在实现 AI 智能体之间的无缝通信和协作。本指南侧重于本地工作流，可让您在部署之前定义和验证代理的功能。

核心工作流程涉及以下步骤：

  1. 定义关键组件
  2. 创建本地代理
  3. 测试本地代理

定义代理组件

如需创建 A2A 智能体，您需要定义以下组件：AgentCard、AgentExecutor 和 ADK LlmAgent。

  • AgentCard 包含一个描述代理功能的元数据文档。AgentCard 就像一张名片，其他代理可以使用它来了解您的代理可以做什么。如需了解详情，请参阅代理卡片规范。
  • AgentExecutor 包含代理的核心逻辑，并定义了代理如何处理任务。您可以在此处实现代理的行为。如需详细了解，请参阅 A2A 协议规范。
  • （可选）LlmAgent 定义 ADK 代理，包括其系统指令、生成模型和工具。

定义 AgentCard

以下代码示例为货币汇率代理定义了 AgentCard：

from a2a.types import AgentCard, AgentSkill
from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card

# Define the skill for the CurrencyAgent
currency_skill = AgentSkill(
    id='get_exchange_rate',
    name='Get Currency Exchange Rate',
    description='Retrieves the exchange rate between two currencies on a specified date.',
    tags=['Finance', 'Currency', 'Exchange Rate'],
    examples=[
        'What is the exchange rate from USD to EUR?',
        'How many Japanese Yen is 1 US dollar worth today?',
    ],
)

# Create the agent card using the utility function
agent_card = create_agent_card(
    agent_name='Currency Exchange Agent',
    description='An agent that can provide currency exchange rates',
    skills=[currency_skill]
)

定义 AgentExecutor

以下代码示例定义了一个 AgentExecutor，用于返回货币兑换汇率。它接受 CurrencyAgent 实例并初始化 ADK Runner 以执行请求。

import requests
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState, TextPart, UnsupportedOperationError, Part
from a2a.utils import new_agent_text_message
from a2a.utils.errors import ServerError
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.sessions import InMemorySessionService
from google.genai import types

class CurrencyAgentExecutorWithRunner(AgentExecutor):
    """Executor that takes an LlmAgent instance and initializes the ADK Runner internally."""

    def __init__(self, agent: LlmAgent):
        self.agent = agent
        self.runner = None

    def _init_adk(self):
        if not self.runner:
            self.runner = Runner(
                app_name=self.agent.name,
                agent=self.agent,
                artifact_service=InMemoryArtifactService(),
                session_service=InMemorySessionService(),
                memory_service=InMemoryMemoryService(),
            )

    async def cancel(self, context: RequestContext, event_queue: EventQueue):
        raise ServerError(error=UnsupportedOperationError())

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        self._init_adk() # Initialize on first execute call

        if not context.message:
            return

        user_id = context.message.metadata.get('user_id') if context.message and context.message.metadata else 'a2a_user'

        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        if not context.current_task:
            await updater.submit()
        await updater.start_work()

        query = context.get_user_input()
        content = types.Content(role='user', parts=[types.Part(text=query)])

        try:
            session = await self.runner.session_service.get_session(
                app_name=self.runner.app_name,
                user_id=user_id,
                session_id=context.context_id,
            ) or await self.runner.session_service.create_session(
                app_name=self.runner.app_name,
                user_id=user_id,
                session_id=context.context_id,
            )

            final_event = None
            async for event in self.runner.run_async(
                session_id=session.id,
                user_id=user_id,
                new_message=content
            ):
                if event.is_final_response():
                    final_event = event

            if final_event and final_event.content and final_event.content.parts:
                response_text = "".join(
                    part.text for part in final_event.content.parts if hasattr(part, 'text') and part.text
                )
                if response_text:
                    await updater.add_artifact(
                        [TextPart(text=response_text)],
                        name='result',
                    )
                    await updater.complete()
                    return

            await updater.update_status(
                TaskState.failed,
                message=new_agent_text_message('Failed to generate a final response with text content.'),
                final=True
            )

        except Exception as e:
            await updater.update_status(
                TaskState.failed,
                message=new_agent_text_message(f"An error occurred: {str(e)}"),
                final=True,
            )

定义 LlmAgent

首先，为 LlmAgent 定义要使用的币种兑换工具：

def get_exchange_rate(
    currency_from: str = "USD",
    currency_to: str = "EUR",
    currency_date: str = "latest",
):
    """Retrieves the exchange rate between two currencies on a specified date.
    Uses the Frankfurter API (https://api.frankfurter.app/) to obtain
    exchange rate data.
    """
    try:
        response = requests.get(
            f"https://api.frankfurter.app/{currency_date}",
            params={"from": currency_from, "to": currency_to},
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        return {"error": str(e)}

然后，定义使用该工具的 ADK LlmAgent。

my_llm_agent = LlmAgent(
    model='gemini-2.0-flash',
    name='currency_exchange_agent',
    description='An agent that can provide currency exchange rates.',
    instruction="""You are a helpful currency exchange assistant.
                   Use the get_exchange_rate tool to answer user questions.
                   If the tool returns an error, inform the user about the error.""",
    tools=[get_exchange_rate],
)

创建本地代理

定义代理的组件后，创建一个使用 AgentCard、AgentExecutor 和 LlmAgent 的 A2aAgent 类实例，以开始本地测试。

from vertexai.preview.reasoning_engines import A2aAgent

a2a_agent = A2aAgent(
    agent_card=agent_card, # Assuming agent_card is defined
    agent_executor_builder=lambda: CurrencyAgentExecutorWithRunner(
        agent=my_llm_agent,
    )
)
a2a_agent.set_up()

A2A 代理模板可帮助您创建符合 A2A 标准的服务。该服务充当封装容器，可为您抽象出转换层。

测试本地代理

汇率代理支持以下三种方法：

  • handle_authenticated_agent_card
  • on_message_send
  • on_get_task

测试 handle_authenticated_agent_card

以下代码会检索代理的已验证卡片，其中描述了代理的功能。

# Test the `authenticated_agent_card` endpoint.
response_get_card = await a2a_agent.handle_authenticated_agent_card(request=None, context=None)
print(response_get_card)

测试 on_message_send

以下代码模拟了客户向代理发送新消息。A2aAgent 可创建新任务并返回任务的 ID。

import json
from starlette.requests import Request
import asyncio

# 1. Define the message payload you want to send.
message_data = {
    "message": {
        "messageId": "local-test-message-id",
        "content":[
              {
                  "text": "What is the exchange rate from USD to EUR today?"
              }
          ],
        "role": "ROLE_USER",
    },
}

# 2. Construct the request
scope = {
    "type": "http",
    "http_version": "1.1",
    "method": "POST",
    "headers": [(b"content-type", b"application/json")],
}

async def receive():
    byte_data = json.dumps(message_data).encode("utf-8")
    return {"type": "http.request", "body": byte_data, "more_body": False}

post_request = Request(scope, receive=receive)

# 3. Call the agent
send_message_response = await a2a_agent.on_message_send(request=post_request, context=None)

print(send_message_response)

测试 on_get_task

以下代码用于检索任务的状态和结果。输出显示任务已完成，并包含“Hello World”响应制品。


from starlette.requests import Request
import asyncio

# 1. Provide the task_id from the previous step.
# In a real application, you would store and retrieve this ID.
task_id_to_get = send_message_response['task']['id']

# 2. Define the path parameters for the request.
task_data = {"id": task_id_to_get}

# 3. Construct the starlette.requests.Request object directly.
scope = {
    "type": "http",
    "http_version": "1.1",
    "method": "GET",
    "headers": [],
    "query_string": b'',
    "path_params": task_data,
}

async def empty_receive():
    return {"type": "http.disconnect"}

get_request = Request(scope, empty_receive)

# 4. Call the agent's handler to get the task status.
task_status_response = await a2a_agent.on_get_task(request=get_request, context=None)

print(f"Successfully retrieved status for Task ID: {task_id_to_get}")
print("\nFull task status response:")
print(task_status_response)

后续步骤