Criar um agente Agent2Agent

O Agent Runtime permite desenvolver e implantar agentes usando o protocolo Agent2Agent (A2A). O A2A é um padrão aberto criado para permitir a comunicação e a colaboração perfeitas entre agentes de IA.

Este documento explica como desenvolver e testar um agente A2A localmente, incluindo a definição de componentes como AgentCard e AgentExecutor.

Para mais informações sobre como gerenciar os agentes implantados, consulte Gerenciar agentes implantados.

O fluxo de trabalho principal envolve as seguintes etapas:

  1. Definir componentes principais
  2. Criar um agente local
  3. Testar o agente local

Definir componentes do agente

Para criar um agente A2A, é necessário definir os seguintes componentes: um AgentCard, um AgentExecutor e um LlmAgent do ADK.

  • AgentCard contém um documento de metadados que descreve os recursos do agente. AgentCard é como um cartão de visitas que outros agentes podem usar para descobrir o que seu agente pode fazer. Para mais detalhes, consulte a especificação do card do agente.
  • AgentExecutor contém a lógica principal do agente e define como ele processa as tarefas. É aqui que você implementa o comportamento do agente. Saiba mais na especificação do protocolo A2A.
  • Opcional: LlmAgent define o agente do ADK, incluindo as instruções do sistema, o modelo generativo e as ferramentas.

Definir um AgentCard

O exemplo de código a seguir define um AgentCard para um agente de taxa de câmbio:

from a2a.types import AgentCard, AgentSkill
from vertexai.agent_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]
)

Definir um AgentExecutor

O exemplo de código a seguir define um AgentExecutor que responde com a taxa de câmbio. Ele usa uma instância de CurrencyAgent e inicializa o ADK Runner para executar solicitações.

import requests
from a2a.server.agent_execution.agent_executor import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a import types as a2a_types
from a2a.types import Part

from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.genai import types as genai_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) -> None:
        task_id = context.task_id
        updater = TaskUpdater(
            event_queue=event_queue,
            task_id=task_id or "",
            context_id=context.context_id or "",
        )
        await updater.cancel()

    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)
        
        task = a2a_types.Task(
            id=context.task_id,
            context_id=context.context_id,
            status=a2a_types.TaskStatus(state=a2a_types.TaskState.TASK_STATE_SUBMITTED),
            history=[context.message] if context.message else [],
        )
        await event_queue.enqueue_event(task)

        await updater.start_work()

        query = context.get_user_input()
        content = genai_types.Content(role='user', parts=[genai_types.Part.from_text(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(
                        [Part(text=response_text)],
                        name='result',
                        last_chunk=True,
                    )
                    await updater.complete()
                    return

            await updater.update_status(
                a2a_types.TaskState.TASK_STATE_FAILED,
                message=updater.new_agent_message([Part(text='Failed to generate a final response with text content.')]),
            )

        except Exception as e:
            await updater.update_status(
                a2a_types.TaskState.TASK_STATE_FAILED,
                message=updater.new_agent_message([Part(text=f"An error occurred: {str(e)}")]),
            )

Definir um LlmAgent

Primeiro, defina uma ferramenta de câmbio para o LlmAgent usar:

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)}

Em seguida, defina um LlmAgent do ADK que usa a ferramenta.

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],
)

Criar um agente local

Depois de definir os componentes do agente, crie uma instância da A2aAgent classe que usa o AgentCard, AgentExecutor e LlmAgent para iniciar os testes locais.

from vertexai.agent_engines.templates.a2a 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()

O modelo de agente A2A ajuda a criar um serviço compatível com A2A. O serviço atua como um wrapper, abstraindo a camada de conversão.

Testar o agente local

O agente de taxa de câmbio oferece suporte aos três métodos a seguir:

  • handle_authenticated_agent_card
  • on_message_send
  • on_get_task

Testar handle_authenticated_agent_card

O código a seguir recupera o card autenticado do agente, que descreve os recursos dele.

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

Testar on_message_send

O código a seguir simula um cliente enviando uma nova mensagem ao agente. O A2aAgent cria uma nova tarefa e retorna o ID dela.

from a2a.types import SendMessageRequest, Message, Part
from a2a.server.context import ServerCallContext

# 1. Define the message
message = Message(
    role="ROLE_USER",
    message_id="local-test-message-id",
    parts=[Part(text="What is the exchange rate from USD to EUR today?")]
)

# 2. Construct the request
request = SendMessageRequest(message=message)

# 3. Construct context
context = ServerCallContext()

# 4. Call the agent
send_message_response = await a2a_agent.on_message_send(request=request, context=context)

print(send_message_response)

Testar on_get_task

O código a seguir recupera o status e o resultado de uma tarefa. A saída mostra que a tarefa foi concluída e inclui o artefato de resposta "Hello World".

from a2a.types import GetTaskRequest

# 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.id

# 2. Construct the request
request = GetTaskRequest(id=task_id_to_get)

# 3. Call the agent's handler to get the task status.
# Reusing the context constructed in the previous step
task_status_response = await a2a_agent.on_get_task(request=request, context=context)

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

A seguir

Guia

Aprenda as cinco maneiras de implantar um agente no Agent Platform Runtime com base nas suas necessidades de desenvolvimento.

Guia

Use um agente Agent2Agent com o Agent Platform Runtime.

Guia

Crie e implante um agente básico e use o serviço de avaliação de IA generativa para avaliar o agente.

Solução de problemas

Aprenda a resolver erros comuns ao criar agentes personalizados.

Recurso

Encontre recursos e suporte para a Google Agent Platform.

Recurso

Confira exemplos do Agent2Agent em Python no GitHub.