1. Introdução
À medida que os agentes de IA assumem mais responsabilidades, fica difícil manter, escalonar e evoluir um único agente que faz tudo. Recursos diferentes geralmente precisam de estratégias de implantação, ciclos de atualização ou até mesmo equipes diferentes.
- O protocolo A2A (Agent2Agent) resolve o lado da comunicação, padronizando como os agentes descobrem os recursos uns dos outros e colaboram em frameworks e organizações.
- O tempo de execução da plataforma de agentes do Gemini Enterprise resolve o lado da implantação: uma plataforma sem servidor totalmente gerenciada que hospeda seus agentes com suporte A2A integrado, escalonamento automático, endpoints seguros, sessões persistentes e gerenciamento de infraestrutura zero.
Juntos, eles permitem criar agentes especializados, implantá-los como serviços A2A detectáveis e compô-los em sistemas multiagentes.
O que você vai criar
Um agente de reservas que gerencia reservas de mesa em restaurantes (criação, verificação e cancelamento) usando o estado da sessão do ADK, que é gerenciado pelas sessões da plataforma de agentes do Gemini Enterprise. Você implanta esse agente no tempo de execução da plataforma de agentes do Gemini Enterprise, onde ele pode ser descoberto pelo card do agente do protocolo A2A. Em seguida , faça upgrade do agente concierge de restaurantes Foodie Finds (do codelab de pré-requisitos. Não se preocupe se você não tiver acessado o codelab. Preparamos um repositório inicial para você) para consumir o agente de reservas como um subagente A2A remoto. O resultado é um sistema multiagente em que o orquestrador encaminha consultas de menu para a caixa de ferramentas do MCP e solicitações de reserva para o agente A2A remoto.
O que você vai aprender
- Criar um agente do ADK que usa o serviço de sessão gerenciada para gerenciar dados de reserva
- Expor um agente do ADK como um servidor A2A com cards e habilidades do agente
- Implantar um agente A2A no Gemini Enterprise Agent Runtime
- Consumir um agente A2A remoto de outro agente do ADK usando
RemoteA2aAgente processando solicitações autenticadas - Testar sistemas multiagente de forma incremental: A2A local, A2A implantado, integração parcial, implantação completa
Pré-requisitos
- (Recomendado) Conclua os seguintes codelabs :
- Como criar agentes de IA persistentes com o ADK e o CloudSQL -> mais detalhes sobre a sessão e o estado do ADK
- RAG agêntico com ADK, MCP Toolbox e Cloud SQL: você pode continuar criando seu agente com base neste codelab. O código inicial fornecido é idêntico.
- Uma conta do Google Cloud com uma conta de faturamento ativa
- Conhecimento básico de Python e conceitos do ADK
2. Configuração do ambiente: continuando do codelab anterior
As narrativas fornecidas neste codelab são a continuação do codelab de pré-requisito: RAG agêntico com ADK, MCP Toolbox e Cloud SQL . Você pode continuar seu trabalho do codelab anterior
Podemos começar a criar no diretório de trabalho do codelab anterior ( o diretório de trabalho deve ser build-agent-adk-toolbox-cloudsql). Para evitar confusão, vamos renomear o diretório com o mesmo nome que usamos quando começamos do zero.
mv ~/build-agent-adk-toolbox-cloudsql ~/adk-a2a-agent-runtime-starter
cloudshell workspace ~/adk-a2a-agent-runtime-starter && cd ~/adk-a2a-agent-runtime-starter
source .env
Verifique se os arquivos de chave do codelab anterior estão no lugar:
echo "--- Restaurant Agent ---"
cat restaurant_agent/agent.py | head -5
echo ""
echo "--- Toolbox Config ---"
cat tools.yaml | head -5
Você vai encontrar o arquivo restaurant_agent/agent.py com a importação LlmAgent e o tools.yaml com a configuração da caixa de ferramentas.
Em seguida, vamos reinicializar o ambiente Python.
rm -rf .venv
uv sync
Além disso, verifique se o banco de dados foi propagado e está pronto:
uv run python scripts/verify_seed.py
Se você seguir todos os detalhes de teste do codelab anterior, poderá ver uma saída como esta:
Menu Items: 16/15 Embeddings: 16/15 ✗ Database not ready
Tudo bem! A verificação do banco de dados não considera os dados adicionais inseridos na verificação de ingestão de dados. Desde que você tenha 15 ou mais dados, tudo bem.
Ativar a API necessária
Em seguida, vamos ativar a API necessária para interagir com a plataforma de agentes do Gemini Enterprise.
gcloud services enable \
cloudresourcemanager.googleapis.com
Você já deve ter os arquivos e a infraestrutura necessários para continuar na próxima seção: A2A Protocol and Gemini Enterprise Agent Runtime!
3. Configuração do ambiente: comece do zero com o repositório inicial
Esta etapa prepara seu ambiente do Cloud Shell, configura seu projeto do Google Cloud e clona o repositório inicial.
Abra o Cloud Shell
Abra o Cloud Shell no navegador. O Cloud Shell oferece um ambiente pré-configurado com todas as ferramentas necessárias para este codelab. Clique em Autorizar quando solicitado a
Em seguida, clique em Visualizar -> Terminal para abrir o terminal.Sua interface vai ficar parecida com esta:
Essa será nossa interface principal, com o IDE na parte de cima e o terminal na parte de baixo.
Configurar seu diretório de trabalho
Clone o repositório inicial. Todo o código que você escrever neste codelab vai ficar aqui:
rm -rf ~/adk-a2a-agent-runtime-starter
git clone https://github.com/alphinside/adk-a2a-agent-runtime-starter.git
cloudshell workspace ~/adk-a2a-agent-runtime-starter && cd ~/adk-a2a-agent-runtime-starter
Crie o arquivo .env com base no modelo fornecido:
cp .env.example .env
Para simplificar a configuração do projeto no terminal, baixe este script de configuração no diretório de trabalho:
curl -sL https://raw.githubusercontent.com/alphinside/cloud-trial-project-setup/main/setup_verify_trial_project.sh -o setup_verify_trial_project.sh
Execute o script. Ele verifica sua conta de faturamento de teste, cria um novo projeto (ou valida um existente), salva o ID do projeto em um arquivo .env no diretório atual e define o projeto ativo em gcloud.
bash setup_verify_trial_project.sh && source .env
O script faz o seguinte:
- Verifique se você tem uma conta de faturamento de teste ativa
- Verifique se há um projeto em
.env(se houver). - Crie um novo projeto ou reutilize o atual
- Vincular a conta de faturamento de teste ao projeto
- Salve o ID do projeto em
.env - Defina o projeto como o projeto
gcloudativo
Verifique se o projeto está configurado corretamente conferindo o texto amarelo ao lado do diretório de trabalho no prompt do terminal do Cloud Shell. Ele vai mostrar o ID do projeto.
Ativar a API necessária
Em seguida, vamos ativar a API necessária para interagir com a plataforma de agentes do Gemini Enterprise.
gcloud services enable \
aiplatform.googleapis.com \
cloudresourcemanager.googleapis.com
Configuração inicial da infraestrutura
Primeiro, vamos instalar as dependências do Python usando uv, um gerenciador de projetos e pacotes Python rápido escrito em Rust ( documentações do uv). Este codelab usa o uv para velocidade e simplicidade na manutenção do projeto Python.
uv sync
Em seguida, execute o script de configuração completa, que cria a instância do Cloud SQL, insere dados e implanta o serviço da caixa de ferramentas, que vai atuar como estado inicial do nosso agente de restaurante.
bash scripts/full_setup.sh > logs/full_setup.log 2>&1 &
4. Conceito: protocolo Agent2Agent (A2A) e tempo de execução do agente do Gemini Enterprise
Antes de criar, vamos entender brevemente as duas tecnologias principais apresentadas neste codelab para dimensionar nosso aplicativo de agente.
O protocolo Agent2Agent (A2A)
O protocolo Agent2Agent (A2A) é um padrão aberto criado para permitir a comunicação e a colaboração perfeitas entre agentes de IA. Enquanto o MCP (Protocolo de Contexto de Modelo) conecta agentes a ferramentas e dados, o A2A conecta agentes a outros agentes, permitindo que eles descubram os recursos uns dos outros, deleguem tarefas e colaborem em frameworks e organizações.
A principal diferença entre encapsular um agente como uma ferramenta (via MCP) e expô-lo via A2A: as ferramentas não têm estado e executam funções únicas, enquanto os agentes A2A podem raciocinar, manter o estado e lidar com interações de várias rodadas, como negociação ou esclarecimento. Um agente exposto via A2A mantém todos os recursos, em vez de ser reduzido a uma chamada de função.
O A2A define três conceitos principais:
- Card do agente: um documento JSON que descreve o que um agente faz, suas habilidades e o endpoint dele. Outros agentes buscam esse card para descobrir recursos.
- Mensagem: uma solicitação de usuário ou agente enviada a um endpoint A2A, acionando uma tarefa.
- Tarefa: uma unidade de trabalho com um ciclo de vida (enviada → em andamento → concluída/com falha) e artefatos que contêm os resultados.
Para mais detalhes, consulte O que é A2A?
Tempo de execução da plataforma de agentes do Gemini Enterprise
O Agent Runtime é um serviço totalmente gerenciado no Google Cloud para implantar, escalonar e gerenciar agentes de IA em produção com recursos de segurança empresarial (por exemplo, VPC Service Controls, CMEK). Ele cuida da infraestrutura para que você possa se concentrar na lógica do agente.
O Agent Runtime oferece:
- Implantação gerenciada: implante agentes criados com ADK, LangGraph ou qualquer framework Python com uma única chamada de SDK.
- Hospedagem A2A: implante agentes como endpoints compatíveis com A2A com exibição automática de cards de agentes e acesso autenticado.
- Sessões persistentes: o
VertexAiSessionServicearmazena o histórico e o estado da conversa em todas as solicitações. - Escalonamento automático: escalona do zero para lidar com o tráfego, sem gerenciamento de infraestrutura.
- Observabilidade: rastreamento, geração de registros e monitoramento integrados pela pilha de observabilidade do Google Cloud.
- e muitos outros recursos. Consulte esta documentação para mais detalhes.
Neste codelab, você vai implantar o agente de reservas no Agent Runtime. O processo de implantação serializa (faz pickle) e envia o código do agente. O Agent Runtime provisiona um endpoint sem servidor que atende ao protocolo A2A. Outros agentes (ou clientes) interagem com ele por chamadas HTTP padrão, autenticadas com credenciais do Google Cloud.
5. Criar o agente de reservas
Esta etapa cria um novo agente do ADK que processa reservas de restaurantes usando o estado da sessão. O agente oferece suporte a três operações (criar, verificar e cancelar) com o número de telefone como chave de pesquisa. Todos os dados de reserva ficam no estado da sessão do ADK.
Estruturar o agente
Use adk create para gerar a estrutura de diretório do agente com a configuração correta de modelo e projeto:
source .env
uv run adk create reservation_agent \
--model gemini-2.5-flash \
--project ${GOOGLE_CLOUD_PROJECT} \
--region ${GOOGLE_CLOUD_LOCATION}
Isso cria um diretório reservation_agent/ com __init__.py, agent.py e .env pré-configurados para o modelo do Gemini na Agent Platform.
adk-a2a-agent-runtime-starter/ ├── reservation_agent/ │ ├── __init__.py │ ├── agent.py │ └── .env ├── logs ├── scripts └── ...
Em seguida, vamos atualizar o código do agente
Escrever o código do agente
Abra o arquivo do agente gerado:
cloudshell edit reservation_agent/agent.py
Em seguida, substitua o conteúdo pelo seguinte:
# reservation_agent/agent.py
from google.adk.agents import LlmAgent
from google.adk.tools import ToolContext
# App-scoped state prefix ensures reservations persist across all sessions.
# See https://adk.dev/sessions/state/ for state scope details.
STATE_PREFIX = "app:reservation:"
def create_reservation(
phone_number: str,
name: str,
party_size: int,
date: str,
time: str,
tool_context: ToolContext,
) -> dict:
"""Create a new restaurant reservation.
Args:
phone_number: Customer's phone number, used as the reservation ID.
name: Name for the reservation.
party_size: Number of guests.
date: Reservation date (e.g., '2025-07-15' or 'this Friday').
time: Reservation time (e.g., '7:00 PM').
Returns:
Confirmation of the reservation.
"""
reservation = {
"name": name,
"party_size": party_size,
"date": date,
"time": time,
"status": "confirmed",
}
tool_context.state[f"{STATE_PREFIX}{phone_number}"] = reservation
return {
"status": "confirmed",
"message": f"Reservation created for {name}, party of {party_size} on {date} at {time}. Phone: {phone_number}.",
}
def check_reservation(phone_number: str, tool_context: ToolContext) -> dict:
"""Look up an existing reservation by phone number.
Args:
phone_number: The phone number used when the reservation was created.
tool_context: ADK tool context for state access.
Returns:
The reservation details, or a message if not found.
"""
reservation = tool_context.state.get(f"{STATE_PREFIX}{phone_number}")
if reservation:
return {"found": True, "reservation": reservation}
return {"found": False, "message": f"No reservation found for {phone_number}."}
def cancel_reservation(phone_number: str, tool_context: ToolContext) -> dict:
"""Cancel an existing reservation by phone number.
Args:
phone_number: The phone number used when the reservation was created.
tool_context: ADK tool context for state access.
Returns:
Confirmation of cancellation, or a message if not found.
"""
key = f"{STATE_PREFIX}{phone_number}"
reservation = tool_context.state.get(key)
if not reservation:
return {"success": False, "message": f"No reservation found for {phone_number}."}
if reservation.get("status") == "cancelled":
return {"success": False, "message": f"Reservation for {phone_number} is already cancelled."}
reservation["status"] = "cancelled"
tool_context.state[key] = reservation
return {"success": True, "message": f"Reservation for {reservation['name']} ({phone_number}) has been cancelled."}
root_agent = LlmAgent(
name="reservation_agent",
model="gemini-2.5-flash",
instruction="""You are a friendly reservation assistant for "Foodie Finds" restaurant.
You help diners create, check, and cancel table reservations.
When a diner wants to make a reservation, collect these details:
- Name for the reservation
- Phone number (used as the reservation ID)
- Party size (number of guests)
- Date
- Time
Always confirm the details before creating the reservation.
When checking or cancelling, ask for the phone number if not provided.
Be concise and professional.""",
tools=[create_reservation, check_reservation, cancel_reservation],
)
6. Preparar a configuração do servidor A2A
Definir o card do agente A2A
O card do agente é uma descrição estruturada dos recursos dele. Outros agentes e clientes usam esse card para descobrir o que seu agente faz. Crie a configuração do card:
cloudshell edit reservation_agent/a2a_config.py
Copie o seguinte para reservation_agent/a2a_config.py:
# reservation_agent/a2a_config.py
from a2a.types import AgentSkill
from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card
reservation_skill = AgentSkill(
id="manage_reservations",
name="Restaurant Reservations",
description="Create, check, and cancel table reservations at Foodie Finds restaurant",
tags=["reservations", "restaurant", "booking"],
examples=[
"Book a table for 4 on Friday at 7pm",
"Check reservation for 555-0101",
"Cancel my reservation, phone number 555-0101",
],
input_modes=["text/plain"],
output_modes=["text/plain"],
)
agent_card = create_agent_card(
agent_name="Reservation Agent",
description="Handles restaurant table reservations — create, check, and cancel bookings for Foodie Finds restaurant.",
skills=[reservation_skill],
)
Criar o executor A2A
O executor faz a ponte entre o protocolo A2A e o agente do ADK. Ele recebe solicitações de A2A, as executa pelo agente do ADK e retorna resultados como tarefas de A2A:
cloudshell edit reservation_agent/executor.py
Copie o seguinte para reservation_agent/executor.py:
# reservation_agent/executor.py
import os
from typing import NoReturn
import vertexai
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
from a2a.utils import new_agent_text_message
from a2a.utils.errors import ServerError
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService, VertexAiSessionService
from google.genai import types
from reservation_agent.agent import root_agent as reservation_agent
class ReservationAgentExecutor(AgentExecutor):
"""Bridge between the A2A protocol and the ADK reservation agent.
Uses InMemorySessionService for local testing, VertexAiSessionService
when deployed to Agent Runtime (detected via GOOGLE_CLOUD_AGENT_ENGINE_ID).
"""
def __init__(self) -> None:
self.agent = None
self.runner = None
def _init_agent(self) -> None:
if self.agent is not None:
return
self.agent = reservation_agent
engine_id = os.environ.get("GOOGLE_CLOUD_AGENT_ENGINE_ID")
if engine_id:
project = os.environ.get("GOOGLE_CLOUD_PROJECT")
location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
vertexai.init(project=project, location=location)
session_service = VertexAiSessionService(
project=project, location=location, agent_engine_id=engine_id,
)
app_name = engine_id
else:
session_service = InMemorySessionService()
app_name = self.agent.name
self.runner = Runner(
app_name=app_name,
agent=self.agent,
artifact_service=InMemoryArtifactService(),
session_service=session_service,
memory_service=InMemoryMemoryService(),
)
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if self.agent is None:
self._init_agent()
query = context.get_user_input()
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
user_id = context.message.metadata.get("user_id", "a2a-user") if context.message.metadata else "a2a-user"
if not context.current_task:
await updater.submit()
await updater.start_work()
try:
session = await self._get_or_create_session(context.context_id, user_id)
content = types.Content(role="user", parts=[types.Part(text=query)])
async for event in self.runner.run_async(
session_id=session.id, user_id=user_id, new_message=content,
):
if event.is_final_response():
parts = event.content.parts
answer = " ".join(p.text for p in parts if p.text) or "No response."
await updater.add_artifact([TextPart(text=answer)], name="answer")
await updater.complete()
break
except Exception as e:
await updater.update_status(
TaskState.failed, message=new_agent_text_message(f"Error: {e!s}"),
)
raise
async def _get_or_create_session(self, context_id: str, user_id: str):
app_name = self.runner.app_name
if context_id:
session = await self.runner.session_service.get_session(
app_name=app_name, session_id=context_id, user_id=user_id,
)
if session:
return session
session = await self.runner.session_service.create_session(
app_name=app_name, user_id=user_id, session_id=context_id,
)
return session
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> NoReturn:
raise ServerError(error=UnsupportedOperationError())
O executor detecta automaticamente o ambiente: quando GOOGLE_CLOUD_AGENT_ENGINE_ID é definido (o tempo de execução do agente injeta isso no momento da implantação), ele usa VertexAiSessionService para sessões persistentes. Localmente, ele volta para InMemorySessionService.
Seu diretório reservation_agent agora deve conter:
reservation_agent/ ├── __init__.py ├── agent.py ├── a2a_config.py ├── executor.py └── .env
7. Preparar um agente A2A usando o SDK da plataforma de agentes e testar localmente
Esta etapa encapsula o agente de reservas como um agente compatível com A2A usando a classe A2aAgent do SDK da plataforma de agente ( o nome do SDK ainda usa o termo vertex para compatibilidade com versões anteriores) e testa o fluxo completo do protocolo A2A localmente: recuperação do card do agente, envio de mensagens e recuperação de tarefas. Esse é o mesmo objeto A2aAgent que você implanta no tempo de execução do agente na próxima etapa.
Adicionar dependências
Instale o SDK da Plataforma de Agentes com suporte ao Agent Runtime e ao ADK, além do SDK A2A:
uv add "google-cloud-aiplatform[agent_engines,adk]==1.149.0" "a2a-sdk==0.3.26"
Entenda os componentes do A2A
Para encapsular um agente do ADK para A2A, são necessários três componentes:
- Card do agente: um "cartão de visitas" que descreve os recursos, as habilidades e o URL do endpoint do agente. Outros agentes usam isso para descobrir o que seu agente faz.
- Executor de agente: a ponte entre o protocolo A2A e a lógica do seu agente do ADK. Ele recebe solicitações de A2A, as executa pelo agente do ADK e retorna resultados como tarefas de A2A.
- A2aAgent: a classe do SDK da plataforma de agentes que combina o card e o executor em uma unidade implantável.
Criar o script de teste
Crie o script a seguir para testar localmente
cloudshell edit scripts/test_a2a_agent_local.py
Copie o seguinte para scripts/test_a2a_agent_local.py:
# scripts/test_a2a_agent_local.py
import asyncio
import json
import os
from pprint import pprint
from dotenv import load_dotenv
from starlette.requests import Request
from vertexai.preview.reasoning_engines import A2aAgent
from reservation_agent.a2a_config import agent_card
from reservation_agent.executor import ReservationAgentExecutor
load_dotenv()
# --- Helper functions for building mock requests ---
def receive_wrapper(data: dict):
async def receive():
byte_data = json.dumps(data).encode("utf-8")
return {"type": "http.request", "body": byte_data, "more_body": False}
return receive
def build_post_request(data: dict = None, path_params: dict = None) -> Request:
scope = {"type": "http", "http_version": "1.1", "headers": [(b"content-type", b"application/json")], "app": None}
if path_params:
scope["path_params"] = path_params
return Request(scope, receive_wrapper(data))
def build_get_request(path_params: dict) -> Request:
scope = {"type": "http", "http_version": "1.1", "query_string": b"", "app": None}
if path_params:
scope["path_params"] = path_params
async def receive():
return {"type": "http.disconnect"}
return Request(scope, receive)
# --- Helper: poll for task completion ---
async def wait_for_task(a2a_agent, task_id, max_retries=30):
"""Poll on_get_task until the task reaches a terminal state."""
for _ in range(max_retries):
request = build_get_request({"id": task_id})
result = await a2a_agent.on_get_task(request=request, context=None)
state = result.get("status", {}).get("state", "")
if state in ["completed", "failed"]:
return result
await asyncio.sleep(1)
return result
def print_task_answer(result):
"""Extract and print the answer from task artifacts."""
print(f"Status: {result.get('status', {}).get('state')}")
for artifact in result.get("artifacts", []):
if artifact.get("parts") and "text" in artifact["parts"][0]:
print(f"Answer: {artifact['parts'][0]['text']}")
# --- Local test ---
async def main():
# Create and set up the A2A agent locally
a2a_agent = A2aAgent(agent_card=agent_card, agent_executor_builder=ReservationAgentExecutor)
a2a_agent.set_up()
# 1. Get agent card
print("=" * 50)
print("1. Retrieving agent card...")
print("=" * 50)
request = build_get_request(None)
card_response = await a2a_agent.handle_authenticated_agent_card(request=request, context=None)
print(f"Agent: {card_response.get('name')}")
print(f"Skills: {[s.get('name') for s in card_response.get('skills', [])]}")
# 2. Create a reservation
print("\n" + "=" * 50)
print("2. Creating a reservation...")
print("=" * 50)
message_data = {
"message": {
"messageId": f"msg-{os.urandom(4).hex()}",
"content": [{"text": "Book a table for 2 on Saturday at 6pm. Name: Bob, Phone: 555-0202"}],
"role": "ROLE_USER",
},
}
request = build_post_request(message_data)
response = await a2a_agent.on_message_send(request=request, context=None)
task_id = response["task"]["id"]
context_id = response["task"].get("contextId")
print(f"Task ID: {task_id}")
# 3. Wait for result
print("\n" + "=" * 50)
print("3. Waiting for task result...")
print("=" * 50)
result = await wait_for_task(a2a_agent, task_id)
print_task_answer(result)
# 4. Check the reservation (same context for session continuity)
print("\n" + "=" * 50)
print("4. Checking the reservation...")
print("=" * 50)
check_data = {
"message": {
"messageId": f"msg-{os.urandom(4).hex()}",
"content": [{"text": "Check the reservation for 555-0202"}],
"role": "ROLE_USER",
"contextId": context_id,
},
}
request = build_post_request(check_data)
check_response = await a2a_agent.on_message_send(request=request, context=None)
check_result = await wait_for_task(a2a_agent, check_response["task"]["id"])
print_task_answer(check_result)
# 5. Cancel the reservation
print("\n" + "=" * 50)
print("5. Cancelling the reservation...")
print("=" * 50)
cancel_data = {
"message": {
"messageId": f"msg-{os.urandom(4).hex()}",
"content": [{"text": "Cancel the reservation for 555-0202"}],
"role": "ROLE_USER",
"contextId": context_id,
},
}
request = build_post_request(cancel_data)
cancel_response = await a2a_agent.on_message_send(request=request, context=None)
cancel_result = await wait_for_task(a2a_agent, cancel_response["task"]["id"])
print_task_answer(cancel_result)
print("\n" + "=" * 50)
print("All tests passed!")
print("=" * 50)
if __name__ == "__main__":
asyncio.run(main())
O script de teste importa o card do agente e o executor que você criou na etapa anterior, sem duplicação. Ele vai criar um A2aAgent local, simular chamadas de protocolo A2A usando solicitações HTTP simuladas e verificar todas as três operações de reserva.
Como nenhum GOOGLE_CLOUD_AGENT_ENGINE_ID está definido localmente, o executor usa InMemorySessionService. Quando implantado no Agent Runtime, o mesmo executor muda automaticamente para VertexAiSessionService em sessões persistentes.
Executar o teste
PYTHONPATH=. uv run python scripts/test_a2a_agent_local.py
A saída passa por cinco etapas:
- Card do agente: recupera os recursos e as habilidades do agente.
- Criar reserva: reserva uma mesa e retorna uma tarefa com a confirmação.
- Receber resultado da tarefa: recupera a tarefa concluída com a resposta.
- Verificar reserva: pesquisa a reserva por número de telefone.
- Cancelar reserva: cancela a reserva e confirma
Exemplo de saída, como mostrado abaixo
================================================== 1. Retrieving agent card... ================================================== Agent: Reservation Agent Skills: ['Restaurant Reservations'] ================================================== 2. Creating a reservation... ================================================== Task ID: f7f7004d-cfea-49c2-b57d-5bca9959e193 ================================================== 3. Waiting for task result... ================================================== Status: TASK_STATE_COMPLETED Answer: Your reservation for Bob, party of 2, on Saturday at 6:00 PM has been confirmed. The phone number associated is 555-0202. ================================================== 4. Checking the reservation... ================================================== Status: TASK_STATE_COMPLETED Answer: I found a reservation for Bob, party of 2, on Saturday at 6:00 PM. The reservation status is confirmed. ================================================== 5. Cancelling the reservation... ================================================== Status: TASK_STATE_COMPLETED Answer: Your reservation for Bob (555-0202) has been cancelled. ================================================== All tests passed! ==================================================
Neste ponto, você verificou: o card do agente A2A descreve as habilidades corretas, todas as três operações de reserva funcionam pelo fluxo de mensagens/tarefas do protocolo A2A e o estado persiste nas mensagens dentro do mesmo contexto.
8. Implantar o agente de reserva no Agent Runtime
Esta etapa implanta o agente de reserva no ambiente de execução da plataforma de agentes do Gemini Enterprise, uma plataforma sem servidor totalmente gerenciada que hospeda seu agente e o expõe como um endpoint A2A seguro. Após a implantação, qualquer cliente autorizado pode descobrir e interagir com o agente usando endpoints HTTP A2A padrão.
Criar o bucket de preparo
Crie um bucket do Cloud Storage para o preparo do Agent Runtime. O ambiente de execução do agente usa esse bucket para fazer upload do código e das dependências do agente durante a implantação:
STAGING_BUCKET="${GOOGLE_CLOUD_PROJECT}-adk-a2a-agent-runtime"
gsutil mb -l $REGION -p $GOOGLE_CLOUD_PROJECT gs://$STAGING_BUCKET 2>/dev/null || echo "Bucket already exists"
echo "STAGING_BUCKET=$STAGING_BUCKET" >> .env
source .env
Criar o script de implantação
Em seguida, precisamos preparar o script de implantação
cloudshell edit scripts/deploy_a2a_agent_runtime.py
Copie o seguinte para scripts/deploy_a2a_agent_runtime.py:
# scripts/deploy_a2a_agent_runtime.py
import os
from pathlib import Path
import vertexai
from dotenv import load_dotenv
from google.genai import types
from vertexai.preview.reasoning_engines import A2aAgent
from reservation_agent.a2a_config import agent_card
from reservation_agent.executor import ReservationAgentExecutor
load_dotenv()
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ["REGION"]
STAGING_BUCKET = os.environ.get("STAGING_BUCKET", f"{PROJECT_ID}-adk-a2a-agent-runtime")
BUCKET_URI = f"gs://{STAGING_BUCKET}"
a2a_agent = A2aAgent(
agent_card=agent_card,
agent_executor_builder=ReservationAgentExecutor,
)
def main():
vertexai.init(project=PROJECT_ID, location=REGION, staging_bucket=BUCKET_URI)
client = vertexai.Client(
project=PROJECT_ID,
location=REGION,
http_options=types.HttpOptions(api_version="v1beta1"),
)
print("Deploying Reservation Agent to Agent Runtime...")
print("This may take 3-5 minutes.")
remote_agent = client.agent_engines.create(
agent=a2a_agent,
config={
"display_name": agent_card.name,
"description": agent_card.description,
"requirements": [
"google-cloud-aiplatform[agent_engines,adk]==1.149.0",
"a2a-sdk==0.3.26",
"google-adk==1.29.0",
"cloudpickle",
"pydantic"
],
"extra_packages": [
"./reservation_agent",
],
"http_options": {
"api_version": "v1beta1",
},
"staging_bucket": BUCKET_URI,
},
)
resource_name = remote_agent.api_resource.name
print(f"\nDeployment complete!")
print(f"Resource name: {resource_name}")
env_path = Path(".env")
lines = env_path.read_text().splitlines() if env_path.exists() else []
lines = [l for l in lines if not l.startswith("RESERVATION_AGENT_RESOURCE_NAME=")]
lines.append(f"RESERVATION_AGENT_RESOURCE_NAME={resource_name}")
env_path.write_text("\n".join(lines) + "\n")
print("Written RESERVATION_AGENT_RESOURCE_NAME to .env")
if __name__ == "__main__":
main()
O script de implantação importa os mesmos agent_card e ReservationAgentExecutor usados no teste local, sem duplicação de código. O Agent Runtime serializa (faz pickle) o objeto A2aAgent com as dependências dele para implantação. No final do script de implantação, ele vai gravar o valor RESERVATION_AGENT_RESOURCE_NAME no arquivo .env.
Implantar no Agent Runtime
Execute o script de implantação:
PYTHONPATH=. uv run python scripts/deploy_a2a_agent_runtime.py
A implantação leva de 3 a 5 minutos. O script provisiona um endpoint sem servidor no Agent Runtime que hospeda o agente de reserva. Após a implantação bem-sucedida, você vai ver uma saída semelhante à mostrada abaixo.
Deploying Reservation Agent to Agent Runtime... This may take 3-5 minutes. Deployment complete! Resource name: projects/your-project-number/locations/us-central1/reasoningEngines/your-agent-deployment-unique-id Written RESERVATION_AGENT_RESOURCE_NAME to .env
É possível conferir o agente implantado no console do Cloud. Pesquise Agent Platform na barra de pesquisa do console.
Em seguida, na guia à esquerda, passe o cursor sobre Agents e selecione Deployments.
Você vai encontrar o Reservation Agent na lista de implantações, como mostrado abaixo.
Testar o agente implantado
Agora, vamos testar o agente implantado e criar um script de teste para ele:
cloudshell edit scripts/test_a2a_agent_runtime.py
Copie o seguinte para scripts/test_a2a_agent_runtime.py:
# scripts/test_a2a_agent_runtime.py
import asyncio
import os
import time
import vertexai
from a2a.types import TaskState
from dotenv import load_dotenv
from google.genai import types
load_dotenv()
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ["REGION"]
RESOURCE_NAME = os.environ["RESERVATION_AGENT_RESOURCE_NAME"]
async def main():
vertexai.init(project=PROJECT_ID, location=REGION)
client = vertexai.Client(
project=PROJECT_ID, location=REGION,
http_options=types.HttpOptions(api_version="v1beta1"),
)
agent = client.agent_engines.get(name=RESOURCE_NAME)
# 1. Get agent card
print("=" * 50)
print("1. Retrieving agent card...")
print("=" * 50)
card = await agent.handle_authenticated_agent_card()
print(f"Agent: {card.name}")
print(f"URL: {card.url}")
print(f"Skills: {[s.name for s in card.skills]}")
# 2. Send a reservation request
print("\n" + "=" * 50)
print("2. Sending reservation request...")
print("=" * 50)
message_data = {
"messageId": "msg-remote-001",
"role": "user",
"parts": [{"kind": "text", "text": "Book a table for 3 on Sunday at noon. Name: Carol, Phone: 555-0303"}],
}
response = await agent.on_message_send(**message_data)
task_object = None
for chunk in response:
if isinstance(chunk, tuple) and len(chunk) > 0 and hasattr(chunk[0], "id"):
task_object = chunk[0]
break
task_id = task_object.id
print(f"Task ID: {task_id}")
print(f"Status: {task_object.status.state}")
# 3. Poll for result
print("\n" + "=" * 50)
print("3. Waiting for result...")
print("=" * 50)
result = None
for _ in range(30):
try:
result = await agent.on_get_task(id=task_id)
if result.status.state in [TaskState.completed, TaskState.failed]:
break
except Exception:
pass
time.sleep(1)
print(f"Final status: {result.status.state}")
if result.artifacts:
for artifact in result.artifacts:
if artifact.parts and hasattr(artifact.parts[0], "root") and hasattr(artifact.parts[0].root, "text"):
print(f"Answer: {artifact.parts[0].root.text}")
print("\n" + "=" * 50)
print("Remote agent test passed!")
print("=" * 50)
if __name__ == "__main__":
asyncio.run(main())
Em seguida, vamos executar o teste.
source .env
uv run python scripts/test_a2a_agent_runtime.py
A saída mostra o card do agente com a skill "Reservas de restaurantes", seguida da conclusão da tarefa com uma confirmação de reserva.
================================================== 1. Retrieving agent card... ================================================== Agent: Reservation Agent URL: https://us-central1-aiplatform.googleapis.com/v1beta1/projects/your-project-id/locations/us-central1/reasoningEngines/your-agent-unique-id/a2a Skills: ['Restaurant Reservations'] ================================================== 2. Sending reservation request... ================================================== Task ID: b34585d0-5f03-4cb0-85a3-40710a0d224d Status: TaskState.completed ================================================== 3. Waiting for result... ================================================== Final status: TaskState.completed Answer: Your reservation for Carol, party of 3 on Sunday at noon with phone number 555-0303 is confirmed. ================================================== Remote agent test passed! ==================================================
O agente de reserva agora está sendo executado como um endpoint A2A gerenciado no ambiente de execução do agente.
9. Integrar o agente de reservas A2A ao agente raiz do restaurante
Esta etapa faz upgrade do agente de restaurante para usar o agente de reserva implantado como um subagente A2A remoto. O orquestrador é executado localmente, enquanto o agente de reserva é executado no Agent Runtime, uma integração parcial que valida a conexão A2A antes da implantação completa.
Resolver o URL do cartão do agente A2A
O RemoteA2aAgent precisa do URL do card do agente de reservas implantado para descobrir os recursos dele. Crie um script que busque esse URL do ambiente de execução do agente e o grave no .env do agente do restaurante:
cloudshell edit scripts/resolve_agent_card_url.py
Copie o seguinte para scripts/resolve_agent_card_url.py:
# scripts/resolve_agent_card_url.py
import asyncio
import os
from pathlib import Path
import vertexai
from dotenv import load_dotenv
from google.genai import types
load_dotenv()
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ["REGION"]
RESOURCE_NAME = os.environ["RESERVATION_AGENT_RESOURCE_NAME"]
async def main():
vertexai.init(project=PROJECT_ID, location=REGION)
client = vertexai.Client(
project=PROJECT_ID, location=REGION,
http_options=types.HttpOptions(api_version="v1beta1"),
)
agent = client.agent_engines.get(name=RESOURCE_NAME)
card = await agent.handle_authenticated_agent_card()
card_url = f"{card.url}/v1/card"
print(f"Agent: {card.name}")
print(f"Card URL: {card_url}")
# Write to restaurant_agent/.env
# Write to both restaurant_agent/.env (for adk web) and root .env (for Cloud Run deploy)
for env_path in [Path("restaurant_agent/.env"), Path(".env")]:
lines = env_path.read_text().splitlines() if env_path.exists() else []
lines = [l for l in lines if not l.startswith("RESERVATION_AGENT_CARD_URL=")]
lines.append(f"RESERVATION_AGENT_CARD_URL={card_url}")
env_path.write_text("\n".join(lines) + "\n")
print(f"Written RESERVATION_AGENT_CARD_URL to {env_path}")
if __name__ == "__main__":
asyncio.run(main())
Execute o script para preencher o arquivo .env com o URL do card do agente.
uv run python scripts/resolve_agent_card_url.py
source .env
Atualizar o agente do restaurante
Abra o arquivo do agente de restaurante:
cloudshell edit restaurant_agent/agent.py
Em seguida, substitua o conteúdo pela versão atualizada que inclui o agente de reserva remota como um subagente:
# restaurant_agent/agent.py
import os
import httpx
from google.adk.agents import LlmAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.auth import default
from google.auth.transport.requests import Request as AuthRequest
from toolbox_adk import ToolboxToolset
TOOLBOX_URL = os.environ.get("TOOLBOX_URL", "http://127.0.0.1:5000")
RESERVATION_AGENT_CARD_URL = os.environ.get("RESERVATION_AGENT_CARD_URL", "")
toolbox = ToolboxToolset(TOOLBOX_URL)
class GoogleCloudAuth(httpx.Auth):
"""Auto-refreshing Google Cloud authentication for httpx.
Refreshes the access token before each request if expired,
so long-running agents never hit 401 errors.
"""
def __init__(self):
self.credentials, _ = default(
scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
def auth_flow(self, request):
# Refresh the token if it is expired or missing
if not self.credentials.valid:
self.credentials.refresh(AuthRequest())
request.headers["Authorization"] = f"Bearer {self.credentials.token}"
yield request
reservation_remote_agent = RemoteA2aAgent(
name="reservation_agent",
description="Handles restaurant table reservations — create, check, and cancel bookings. Delegate to this agent when the user wants to book a table, check a reservation, or cancel a reservation.",
agent_card=RESERVATION_AGENT_CARD_URL,
httpx_client=httpx.AsyncClient(auth=GoogleCloudAuth(), timeout=60),
)
root_agent = LlmAgent(
name="restaurant_agent",
model="gemini-2.5-flash",
instruction="""You are a friendly and knowledgeable concierge at "Foodie Finds," a restaurant. Your job:
- Help diners browse the menu by category or cuisine type.
- Provide full details about specific dishes, including ingredients, price, and dietary information.
- Recommend dishes based on natural language descriptions of what the diner is craving.
- Add new menu items when asked.
- For reservation requests (booking, checking, or cancelling tables), delegate to the reservation_agent.
When a diner asks about a specific dish by name or cuisine, use the get-item-details tool.
When a diner asks for a specific category or cuisine type, use the search-menu tool.
When a diner describes what kind of food they want — by flavor, texture, dietary needs, or cravings — use the search-menu-by-description tool for semantic search.
When in doubt between search-menu and search-menu-by-description, prefer search-menu-by-description — it searches dish descriptions and finds more relevant matches.
If a dish is not available (available is false), let the diner know and suggest similar alternatives from the search results.
Be conversational, knowledgeable, and concise.""",
tools=[toolbox],
sub_agents=[reservation_remote_agent],
)
As principais mudanças em relação à versão anterior são:
GoogleCloudAuth: um manipuladorhttpx.Authpersonalizado que atualiza o token de acesso do Google Cloud antes de cada solicitação. O Agent Runtime exige chamadas A2A autenticadas, e os tokens expiram após um período.- O
RemoteA2aAgentlêRESERVATION_AGENT_CARD_URLdo.env(escrito pelo script de resolução) e usa ohttpx_clientautenticado. - Registrado como um subagente: o orquestrador do ADK delega automaticamente solicitações de reserva a ele.
- Instrução atualizada para mencionar a delegação de reserva
Testar o agente integrado localmente
O agente inicial exigia integração com a caixa de ferramentas do MCP. O arquivo necessário já deveria ter sido fornecido no codelab anterior ou no repositório inicial. Só precisamos garantir que o processo da caixa de ferramentas seja executado corretamente.
Se TOOLBOX_URL no seu .env já apontar para um serviço do Cloud Run (do codelab anterior ou talvez do full_setup.sh do repositório inicial), pule esta etapa. O agente vai se conectar à caixa de ferramentas implantada.
Se você precisar de uma caixa de ferramentas local, verifique se uma já está em execução antes de iniciar uma nova instância:
if curl -s http://127.0.0.1:5000/api/toolsets > /dev/null 2>&1; then
echo "Toolbox already running on port 5000"
else
set -a; source .env; set +a
./toolbox --config=tools.yaml > logs/toolbox.log 2>&1 &
echo "Toolbox started"
fi
Em seguida, podemos tentar interagir com o agente de restaurante pela interface de desenvolvimento da Web do ADK.
uv run adk web --allow_origins "regex:https://.*\.cloudshell\.dev" --port 8080
Abra a interface da Web do ADK usando a Visualização da Web do Cloud Shell (clique no botão "Visualização da Web" e mude a porta para 8080). Em seguida, selecione restaurant_agent.
Teste uma conversa mista:
Consulta de menu
What Italian dishes do you have?
Pedido de reserva
I want to create reservation under name Bob, phone number 123456
Verificar reserva
Criar uma nova sessão ( iniciar uma conversa do zero):
Check the reservation for 123456
Interrompa o processo adk web com Ctrl+C duas vezes. Em seguida, vamos concluir o sistema implantando totalmente o agente
10. Implantar o agente de restaurante atualizado no Cloud Run
Esta etapa reimplanta o agente de restaurante no Cloud Run com a integração A2A, concluindo o sistema multiagente totalmente implantado.
Conceder permissões para acessar o ambiente de execução do agente
A conta de serviço do Cloud Run precisa de permissão para chamar o ambiente de execução do agente. Conceda o papel roles/aiplatform.user à conta de serviço padrão do Compute Engine:
PROJECT_NUMBER=$(gcloud projects describe $GOOGLE_CLOUD_PROJECT --format='value(projectNumber)')
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \
--member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
--role="roles/aiplatform.user"
Implantar no Cloud Run
Nesta configuração, presumimos que o serviço de agente de restaurante já existe do codelab anterior ou executando o scripts/full_setup.sh se você começar do zero. Isso faz uma nova implantação com o código atualizado (nova integração do RemoteA2aAgent) e adiciona o URL do card do agente de reservas como uma nova variável de ambiente. As variáveis de ambiente atuais (TOOLBOX_URL, GOOGLE_CLOUD_PROJECT etc.) são preservadas:
gcloud run deploy restaurant-agent \
--source . \
--region=$REGION \
--allow-unauthenticated \
--update-env-vars="RESERVATION_AGENT_CARD_URL=$RESERVATION_AGENT_CARD_URL" \
--min-instances=0 \
--max-instances=1 \
--memory=1Gi \
--port=8080
Testar o sistema totalmente implantado
Consiga o URL do serviço implantado:
AGENT_URL=$(gcloud run services describe restaurant-agent --region=$REGION --format='value(status.url)')
echo "Agent URL: $AGENT_URL"
Abra o URL no seu navegador. A interface da Web do ADK é carregada. Essa é a mesma interface que você usou localmente, agora em execução no Cloud Run.
Você pode conversar com o agente
Consulta de menu
What spicy dishes do you have?
Pedido de reserva
Book a table for 4 on Friday at 7pm. Name: Eve, Phone: 555-0505
Verificar reserva
Criar uma nova sessão ( iniciar uma conversa do zero):
Check reservation for 555-0505
O sistema multiagente está totalmente implantado. O agente do restaurante no Cloud Run orquestra dois serviços de back-end: a caixa de ferramentas do MCP para operações de cardápio e o agente de reservas A2A no Agent Runtime.
11. Parabéns!
Você criou e implantou um sistema multiagente usando o protocolo A2A no Google Cloud.
O que você aprendeu
- Criou um agente do ADK que usa o estado da sessão (
ToolContext) para gerenciar dados de reserva sem um banco de dados - Implantou um agente A2A no Agent Runtime usando o SDK da plataforma de agentes
- Consumiu um agente A2A remoto de outro agente do ADK usando
RemoteA2aAgentcomo um subagente - Testamos o sistema de forma incremental: A2A local → A2A implantado → integração parcial → implantação completa
Limpeza
Para evitar cobranças na sua conta do Google Cloud, exclua os recursos criados neste codelab.
Opção 1: excluir o projeto (recomendado)
gcloud projects delete $GOOGLE_CLOUD_PROJECT
Opção 2: excluir recursos individuais
# Delete the Agent Runtime deployment
uv run python -c "
import vertexai
from google.genai import types
vertexai.init(project='$GOOGLE_CLOUD_PROJECT', location='$REGION')
client = vertexai.Client(
project='$GOOGLE_CLOUD_PROJECT', location='$REGION',
http_options=types.HttpOptions(api_version='v1beta1'),
)
agent = client.agent_engines.get(name='$RESERVATION_AGENT_RESOURCE_NAME')
agent.delete(force=True)
print('Agent Runtime deployment deleted.')
"
# Delete Cloud Run services
gcloud run services delete restaurant-agent --region=$REGION --quiet
gcloud run services delete toolbox-service --region=$REGION --quiet
# Delete Cloud SQL instance
gcloud sql instances delete $DB_INSTANCE --quiet
# Delete GCS staging bucket
gsutil rm -r gs://$STAGING_BUCKET