Resolver endpoints e criar orquestradores

O Kit de Desenvolvimento de Agente (ADK) oferece um cliente AgentRegistry dedicado que permite descobrir, pesquisar e se conectar de forma programática a agentes de IA e servidores do Protocolo de Contexto de Modelo (MCP) catalogados no Agent Registry.

Em vez de fixar no código os URLs de endpoint no aplicativo, use o ADK para resolver esses endpoints durante a execução.

O Agent Registry fornece os endpoints subjacentes, mas as implantações de produção normalmente roteiam essas invocações pelo Gateway de Agente. O Gateway de Agente ajuda você a aplicar políticas de segurança, realizar mediação de protocolo e aplicar filtragem de conteúdo às ferramentas descobertas.

Este documento descreve como recuperar agentes remotos e conjuntos de ferramentas do MCP do Agent Registry e incluí-los em um agente orquestrador principal.

Antes de começar

Antes de integrar o ADK ao Agent Registry, conclua o seguinte:

  1. Configure o Agent Registry no projeto.

  2. Instale ou faça upgrade para a versão mais recente do ADK com as dependências A2A necessárias:

    pip 

    pip install --upgrade "google-adk[a2a]"

    uv 

    uv add "google-adk[a2a]"

    Faça upgrade para pelo menos google-adk>=1.29.0.

  3. Configure as Application Default Credentials (ADC):

    gcloud auth application-default login

As credenciais do ADC precisam ter as permissões do IAM necessárias para os serviços subjacentes com que os agentes ou ferramentas interagem. Também é possível usar cabeçalhos personalizados para conjuntos de ferramentas externas. Para mais informações, consulte Autenticar ferramentas e recursos.

Defina as variáveis de ambiente

Para seguir este guia, defina as seguintes variáveis de ambiente:

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=LOCATION

Substitua:

  • PROJECT_ID: o ID do projeto.
  • LOCATION: a região ou o local do registro, como us-central1.

Inicializar o cliente do registro

Para interagir com seu registro de maneira programática, inicialize o cliente AgentRegistry com seu projeto e local:

import os
from google.adk.integrations.agent_registry import AgentRegistry

project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
location = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")

if not project_id:
    raise ValueError("GOOGLE_CLOUD_PROJECT environment variable not set.")

# Initialize the client
registry = AgentRegistry(
    project_id=project_id,
    location=location,
)

Criar um sistema multiagente

O ADK abstrai a mecânica de conexão subjacente, permitindo que você crie aplicativos escalonáveis ao compor vários agentes especializados em hierarquias flexíveis.

É possível usar o cliente do registro para buscar recursos específicos e transmiti-los diretamente para a definição de um novo agente LlmAgent. Seu orquestrador pode invocar agentes remotos como subagentes e executar ferramentas do MCP como se fossem funções Python locais.

Use os seguintes métodos:

  • Para buscar um agente remoto: use get_remote_a2a_agent()
  • Para buscar um conjunto de ferramentas do MCP: use get_mcp_toolset()

O exemplo a seguir demonstra como criar um sistema multiagente criando um agente orquestrador que aproveita um agente de viagens registrado e um servidor MCP do Compute Engine registrado. Neste exemplo, a autenticação é processada pela própria identidade do agente, mas você pode usar outros métodos, como chaves de API e OAuth. Para mais informações, consulte Autenticar ferramentas e recursos.

Para projetar e orquestrar agentes usando uma interface visual com pouco código, configure subagentes registrados diretamente na tela do Agent Studio. Para mais informações, consulte Projetar agentes no Agent Studio.

import httpx
import google.auth
from google.auth.transport.requests import Request
from google.adk.agents.llm_agent import LlmAgent

# Define the GoogleAuth class for the HTTP client
class GoogleAuth(httpx.Auth):
    def __init__(self):
        self.creds, _ = google.auth.default()
    def auth_flow(self, request):
        if not self.creds.valid:
            self.creds.refresh(Request())
        request.headers["Authorization"] = f"Bearer {self.creds.token}"
        yield request

# Connect to a remote A2A agent using its resource name in short or full format
# Short formats automatically imply the client's configured project and location
# Short format: "agents/AGENT_ID"
# Full format: f"projects/{project_id}/locations/{location}/agents/AGENT_ID"
agent_name = "agents/AGENT_ID"

# Configure the HTTP client with GoogleAuth and a 60-second timeout
httpx_client = httpx.AsyncClient(auth=GoogleAuth(), timeout=httpx.Timeout(60.0))
my_remote_agent = registry.get_remote_a2a_agent(
    agent_name=agent_name,
    httpx_client=httpx_client
)

# Retrieve an MCP toolset using its resource name in short or full format
# Short formats automatically imply the client's configured project and location
# Short format: "mcpServers/SERVER_ID"
# Full format: f"projects/{project_id}/locations/{location}/mcpServers/SERVER_ID"
mcp_server_name = "mcpServers/SERVER_ID"
my_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)

# Compose the orchestrator agent
main_agent = LlmAgent(
    model="MODEL_ID", # Replace with a model such as gemini-1.5-flash
    name="travel_orchestrator",
    instruction="""You are a travel coordinator. You can use your
                   subagents to book travel and your tools to query
                   historical travel data.""",
    tools=[my_mcp_toolset],
    sub_agents=[my_remote_agent],
)

# You can now run your orchestrator agent
# response = await main_agent.run('Book a flight to Paris and check my past trips.')

Práticas recomendadas para reutilizar agentes

Para minimizar a latência da rede, busque seus agentes e conjuntos de ferramentas no registro uma vez na inicialização do aplicativo em vez de chamar get_remote_a2a_agent() em cada invocação.

Um agente só pode ter um agente principal por vez. Se você tentar atribuir a mesma instância de agente buscada a vários orquestradores, o ADK poderá gerar um erro indicando que o agente já tem um pai.

Para reutilizar um agente descoberto em vários agentes principais, use o método .clone() para criar uma nova instância do objeto de agente.

O exemplo a seguir mostra como buscar um agente uma vez e cloná-lo para uso em diferentes orquestradores:

import httpx
import google.auth
from google.auth.transport.requests import Request
from google.adk.agents.llm_agent import LlmAgent

# Define the GoogleAuth class for the HTTP client
class GoogleAuth(httpx.Auth):
    def __init__(self):
        self.creds, _ = google.auth.default()
    def auth_flow(self, request):
        if not self.creds.valid:
            self.creds.refresh(Request())
        request.headers["Authorization"] = f"Bearer {self.creds.token}"
        yield request

# Configure the HTTP client with GoogleAuth and a 60-second timeout
httpx_client = httpx.AsyncClient(auth=GoogleAuth(), timeout=httpx.Timeout(60.0))

# Fetch the remote agent once during startup
# Use the resource name in short or full format
# Short formats automatically imply the client's configured project and location
# Short format: "agents/AGENT_ID"
# Full format: f"projects/{project_id}/locations/{location}/agents/AGENT_ID"
agent_name = f"projects/PROJECT_ID/locations/LOCATION/agents/AGENT_ID"
base_remote_agent = registry.get_remote_a2a_agent(
    agent_name=agent_name,
    httpx_client=httpx_client
)

# Use .clone() to assign the agent to different parent orchestrators
flight_orchestrator = LlmAgent(
    model="gemini-1.5-flash",
    name="flight_orchestrator",
    sub_agents=[base_remote_agent.clone()]
)

hotel_orchestrator = LlmAgent(
    model="gemini-1.5-flash",
    name="hotel_orchestrator",
    sub_agents=[base_remote_agent.clone()]
)

A seguir