Endpunkte auflösen und Orchestratoren erstellen

Das Agent Development Kit (ADK) bietet einen dedizierten AgentRegistry Client, mit dem Sie KI-Agenten und MCP-Server (Model Context Protocol), die in der Agent Registry katalogisiert sind, programmatisch ermitteln, nachschlagen und eine Verbindung zu ihnen herstellen können.

Anstatt Endpunkt-URLs fest in Ihre Anwendung zu codieren, können Sie diese Endpunkte mit dem ADK zur Laufzeit auflösen.

Die Agent Registry stellt die zugrunde liegenden Endpunkte bereit, aber bei Produktionsbereitstellungen werden diese Aufrufe in der Regel über Agent Gatewayweitergeleitet. Mit Agent Gateway können Sie Sicherheitsrichtlinien erzwingen, Protokollvermittlung durchführen und Inhaltsfilterung auf die von Ihnen ermittelten Tools anwenden.

In diesem Dokument wird beschrieben, wie Sie Remote-Agenten und MCP-Toolsets aus der Agent Registry abrufen und in einen übergeordneten Orchestrator-Agenten einbinden.

Hinweis

Bevor Sie das ADK in die Agent Registry einbinden, führen Sie die folgenden Schritte aus:

  1. Richten Sie die Agent Registry in Ihrem Projekt ein.

  2. Installieren Sie die neueste Version des ADK oder führen Sie ein Upgrade darauf durch. Die erforderlichen A2A-Abhängigkeiten müssen ebenfalls installiert sein:

    pip 

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

    uv 

    uv add "google-adk[a2a]"

    Sie müssen ein Upgrade auf mindestens google-adk>=1.29.0 durchführen.

  3. Konfigurieren Sie die Standardanmeldedaten für Anwendungen (Application Default Credentials, ADC):

    gcloud auth application-default login

Die ADC-Anmeldedaten müssen die erforderlichen IAM-Berechtigungen für die zugrunde liegenden Dienste haben, mit denen die Agenten oder Tools interagieren. Optional können Sie auch benutzerdefinierte Header für externe Toolsets verwenden. Weitere Informationen finden Sie unter Bei Tools und Ressourcen authentifizieren.

Umgebungsvariablen festlegen

Legen Sie die folgenden Umgebungsvariablen fest, um dieser Anleitung zu folgen:

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=LOCATION

Ersetzen Sie Folgendes:

  • PROJECT_ID: Ihre Projekt-ID.
  • LOCATION: die Registry-Region oder der Registry-Standort, z. B. us-central1.

Registry-Client initialisieren

Wenn Sie programmatisch mit Ihrer Registry interagieren möchten, initialisieren Sie den AgentRegistry-Client mit Ihrem Projekt und Standort:

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

Multi-Agenten-System erstellen

Das ADK abstrahiert die zugrunde liegenden Verbindungsmechanismen und ermöglicht Ihnen so, skalierbare Anwendungen zu entwerfen, indem Sie mehrere spezialisierte Agenten in flexiblen Hierarchien zusammenstellen.

Mit dem Registry-Client können Sie bestimmte Ressourcen abrufen und direkt in die Definition eines neuen LlmAgent-Agenten einfügen. Ihr Orchestrator kann Remote-Agenten als Sub-Agenten aufrufen und MCP-Tools ausführen, als wären sie lokale Python-Funktionen.

Verwenden Sie die folgenden Methoden:

  • Remote-Agent abrufen: Verwenden Sie get_remote_a2a_agent().
  • MCP-Toolset abrufen: Verwenden Sie get_mcp_toolset()

Das folgende Beispiel zeigt, wie Sie ein Multi-Agenten-System erstellen, indem Sie einen Orchestrator-Agenten erstellen, der einen registrierten Reiseagenten nutzt und einen registrierten Compute Engine-MCP-Server. In diesem Beispiel wird die Authentifizierung über die eigene Identität des Agenten abgewickelt. Sie können aber auch andere Methoden wie API-Schlüssel und OAuth verwenden. Weitere Informationen finden Sie unter Bei Tools und Ressourcen authentifizieren.

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 Practices für die Wiederverwendung von Agenten

Um die Netzwerklatenz zu minimieren, rufen Sie Ihre Agenten und Toolsets beim Start der Anwendung einmal aus der Registry ab, anstatt get_remote_a2a_agent() bei jedem Aufruf aufzurufen.

Ein Agent kann jeweils nur einen übergeordneten Agenten haben. Wenn Sie versuchen, dieselbe abgerufene Agenteninstanz mehreren Orchestratoren zuzuweisen, gibt das ADK möglicherweise einen Fehler aus, der darauf hinweist, dass der Agent bereits einen übergeordneten Agenten hat.

Wenn Sie einen ermittelten Agenten für mehrere übergeordnete Agenten wiederverwenden möchten, erstellen Sie mit der Methode .clone() eine neue Instanz des Agentenobjekts.

Das folgende Beispiel zeigt, wie Sie einen Agenten einmal abrufen und ihn für die Verwendung in verschiedenen Orchestratoren klonen:

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

Nächste Schritte