Risolvi gli endpoint e crea orchestratori

Il Agent Development Kit (ADK) fornisce un client AgentRegistry dedicato che ti consente di scoprire, cercare e connetterti in modo programmatico agli agenti AI e ai server Model Context Protocol (MCP) catalogati in Agent Registry.

Invece di eseguire l'hard-coding degli URL endpoint nella tua applicazione, puoi utilizzare l'ADK per risolvere questi endpoint in fase di runtime.

Agent Registry fornisce gli endpoint sottostanti, ma in genere le implementazioni di produzione instradano queste chiamate tramite Agent Gateway. Agent Gateway ti aiuta ad applicare le norme di sicurezza, eseguire la mediazione del protocollo e applicare il filtro dei contenuti agli strumenti che scopri.

Questo documento descrive come recuperare gli agenti remoti e i set di strumenti MCP da Agent Registry e includerli in un agente orchestratore principale.

Prima di iniziare

Prima di integrare l'ADK con Agent Registry, completa i seguenti passaggi:

  1. Configura Agent Registry nel tuo progetto.

  2. Installa o esegui l'upgrade all'ultima versione dell'ADK con le dipendenze A2A necessarie:

    pip 

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

    uv 

    uv add "google-adk[a2a]"

    Devi eseguire l'upgrade ad almeno google-adk>=1.29.0.

  3. Configura le Credenziali predefinite dell'applicazione (ADC):

    gcloud auth application-default login

Le credenziali ADC devono disporre delle autorizzazioni IAM necessarie per i servizi sottostanti con cui interagiscono gli agenti o gli strumenti. Facoltativamente, puoi anche utilizzare intestazioni personalizzate per i set di strumenti esterni. Per ulteriori informazioni, consulta Autenticarsi a strumenti e risorse.

Imposta le variabili di ambiente

Per seguire questa guida, imposta le seguenti variabili di ambiente:

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=LOCATION

Sostituisci quanto segue:

  • PROJECT_ID: il tuo ID progetto.
  • LOCATION: la regione o la località del registro, ad esempio us-central1.

Inizializza il client del registro

Per interagire con il registro in modo programmatico, inizializza il client AgentRegistry con il tuo progetto e la tua località:

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

Componi un sistema multi-agente

L'ADK astrae i meccanismi di connessione sottostanti, consentendoti di progettare applicazioni scalabili componendo più agenti specializzati in gerarchie flessibili.

Puoi utilizzare il client del registro per recuperare risorse specifiche e passarle direttamente nella definizione di un nuovo agente LlmAgent. L'orchestratore può richiamare gli agenti remoti come sub-agenti ed eseguire gli strumenti MCP come se fossero funzioni Python locali.

Utilizza i seguenti metodi:

  • Per recuperare un agente remoto: utilizza get_remote_a2a_agent()
  • Per recuperare un set di strumenti MCP: utilizza get_mcp_toolset()

L'esempio seguente mostra come comporre un sistema multi-agente creando un agente orchestratore che sfrutta un'agenzia viaggi registrata e un server MCP Compute Engine registrato. In questo esempio, l'autenticazione viene gestita dalla stessa identità dell'agente, ma puoi utilizzare altri metodi, come le chiavi API e OAuth. Per ulteriori informazioni, consulta Autenticarsi a strumenti e risorse.

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
                   sub-agents 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.')

Best practice per il riutilizzo degli agenti

Per ridurre al minimo la latenza di rete, recupera gli agenti e i set di strumenti dal registro una sola volta all'avvio dell'applicazione anziché chiamare get_remote_a2a_agent() a ogni chiamata.

Un agente può avere un solo agente principale alla volta. Se tenti di assegnare la stessa istanza dell'agente recuperata a più orchestratori, l'ADK potrebbe generare un errore che indica che l'agente ha già un agente principale.

Per riutilizzare un agente scoperto in più agenti principali, utilizza il metodo .clone() per creare una nuova istanza dell'oggetto agente.

L'esempio seguente mostra come recuperare un agente una sola volta e clonarlo per utilizzarlo in diversi orchestratori:

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

Passaggi successivi