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:
Definir componentes do agente
Para criar um agente A2A, é necessário definir os seguintes componentes: um AgentCard, um AgentExecutor e um LlmAgent do ADK.
AgentCardconté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.AgentExecutorconté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:
LlmAgentdefine o agente do ADK, incluindo instruções do sistema, modelo generativo e 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, o AgentExecutor e o 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_cardon_message_sendon_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
Implantar agentes
Aprenda as cinco maneiras de implantar um agente no ambiente de execução do Agent Platform com base nas suas necessidades de desenvolvimento.
Usar um agente Agent2Agent
Use um agente Agent2Agent com o ambiente de execução do Agent Platform.
Avaliar seus agentes
Crie e implante um agente básico e use o serviço de avaliação de IA generativa para avaliar o agente.
Resolver problemas na criação de agentes
Aprenda a resolver erros comuns ao criar agentes personalizados.