Résoudre les points de terminaison et créer des orchestrateurs

L'Agent Development Kit (ADK) fournit un client AgentRegistry dédié qui vous permet de découvrir, de rechercher et de vous connecter de manière programmatique aux agents IA et aux serveurs MCP (Model Context Protocol) catalogués dans Agent Registry.

Au lieu de coder en dur les URL de point de terminaison dans votre application, vous pouvez utiliser l'ADK pour résoudre ces points de terminaison au moment de l'exécution.

Agent Registry fournit les points de terminaison sous-jacents, mais les déploiements de production routent généralement ces appels via Agent Gateway. Agent Gateway vous aide à appliquer des règles de sécurité, à effectuer la médiation de protocole et à appliquer le filtrage de contenu aux outils que vous découvrez.

Ce document explique comment récupérer des agents distants et des ensembles d'outils MCP à partir d'Agent Registry et les inclure dans un agent orchestrateur parent.

Avant de commencer

Avant d'intégrer l'ADK à Agent Registry, procédez comme suit :

  1. Configurez Agent Registry dans votre projet.

  2. Installez la dernière version de l'ADK ou mettez-la à niveau avec les dépendances A2A nécessaires :

    pip 

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

    uv 

    uv add "google-adk[a2a]"

    Vous devez passer à la version google-adk>=1.29.0 au minimum.

  3. Configurez les identifiants par défaut de l'application (ADC) :

    gcloud auth application-default login

Les identifiants ADC doivent disposer des autorisations IAM nécessaires pour les services sous-jacents avec lesquels les agents ou les outils interagissent. Vous pouvez également utiliser des en-têtes personnalisés pour les ensembles d'outils externes. Pour en savoir plus, consultez S'authentifier auprès des outils et des ressources.

Définir des variables d'environnement

Pour suivre ce guide, définissez les variables d'environnement suivantes :

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=LOCATION

Remplacez les éléments suivants :

  • PROJECT_ID : ID de votre projet.
  • LOCATION : région ou emplacement du registre, par exemple us-central1.

Initialiser le client du registre

Pour interagir avec votre registre de manière programmatique, initialisez le client AgentRegistry avec votre projet et votre emplacement :

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

Composer un système multi-agent

L'ADK fait abstraction des mécanismes de connexion sous-jacents, ce qui vous permet de concevoir des applications évolutives en composant plusieurs agents spécialisés dans des hiérarchies flexibles.

Vous pouvez utiliser le client du registre pour extraire des ressources spécifiques et les transmettre directement dans la définition d'un nouvel agent LlmAgent. Votre orchestrateur peut appeler des agents distants en tant que sous-agents et exécuter des outils MCP comme s'il s'agissait de fonctions Python locales.

Utilisez les méthodes suivantes :

  • Pour récupérer un agent distant : utilisez get_remote_a2a_agent().
  • Pour récupérer un ensemble d'outils MCP : utilisez get_mcp_toolset().

L'exemple suivant montre comment composer un système multi-agent en créant un agent orchestrateur qui tire parti d'un agent de voyage enregistré et d'un serveur MCP Compute Engine enregistré. Dans cet exemple, l'authentification est gérée par la propre identité de l'agent, mais vous pouvez utiliser d'autres méthodes, telles que les clés API et OAuth. Pour en savoir plus, consultez S'authentifier auprès des outils et des ressources.

Pour concevoir et orchestrer des agents à l'aide d'une interface visuelle low-code, configurez les sous-agents enregistrés directement dans le canevas Agent Studio. Pour en savoir plus, consultez Concevoir des agents dans 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.')

Bonnes pratiques pour réutiliser des agents

Pour minimiser la latence du réseau, récupérez vos agents et vos ensembles d'outils à partir du registre une seule fois au démarrage de l'application, au lieu d'appeler get_remote_a2a_agent() à chaque invocation.

Un agent ne peut avoir qu'un seul agent parent à la fois. Si vous tentez d'attribuer la même instance d'agent récupérée à plusieurs orchestrateurs, l'ADK peut générer une erreur indiquant que l'agent a déjà un parent.

Pour réutiliser un agent détecté dans plusieurs agents parents, utilisez la méthode .clone() afin de créer une instance de l'objet agent.

L'exemple suivant montre comment extraire un agent une seule fois et le cloner pour l'utiliser dans différents orchestrateurs :

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

Étapes suivantes