解析端點並建構協調器

Agent Development Kit (ADK) 提供專屬的 AgentRegistry 用戶端，可讓您透過程式輔助方式，探索、查詢及連結在 Agent Registry 註冊的 AI 代理和 Model Context Protocol (MCP) 伺服器。您可以使用 ADK 在執行階段解析這些端點，並將其組合成父項協調器代理，而不必將端點網址硬式編碼至應用程式。

Agent Registry 提供基礎端點。如果是正式版部署作業，您可以透過 Agent Gateway 轉送叫用，強制執行安全性政策、執行通訊協定中介，以及套用內容篩選。

事前準備

將 ADK 與 Agent Registry 整合前，請先完成下列事項：

  1. 設定 Agent Registry。

  2. 如果是單一專案設定，請取得已啟用 Agent Registry 的管理專案ID。如果您為多個專案設定了管理邊界，請使用管理專案的 ID 進行邊界範圍的探索和註冊，或指定成員專案 ID 進行單一專案作業。

  3. 安裝或升級至最新版本 ADK，並加入 A2A、Agent Identity 和 MCP 依附元件：

    pip 

    pip install --upgrade "google-adk[a2a,agent-identity,mcp]"

    uv 

    uv add "google-adk[a2a,agent-identity,mcp]"

    您必須升級至 google-adk>=1.29.0 以上版本。

  4. 設定應用程式預設憑證 (ADC)：

    gcloud auth application-default login

您的憑證必須具備必要的 Identity and Access Management (IAM) 權限，才能存取代理或工具互動的基礎服務。您也可以視需要為外部工具組使用自訂標頭。詳情請參閱「向工具和資源進行驗證」。

設定環境變數

設定下列環境變數：

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=LOCATION

更改下列內容：

  • PROJECT_ID：管理專案的專案 ID。
  • LOCATION：登錄檔區域或位置，例如 us-central1。

初始化登錄用戶端

如要透過程式與登錄檔互動，請使用管理專案和位置初始化 AgentRegistry 用戶端：

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

建構多代理系統

ADK 會抽象化底層連線機制，讓您將多個專用代理組合成彈性的階層結構，設計可擴充的應用程式。

您可以使用登錄用戶端擷取特定資源，並直接傳遞至新 LlmAgent 代理的定義。自動化調度管理工具可以叫用遠端代理做為子代理，並執行 MCP 工具，就像執行本機 Python 函式一樣。

如要擷取遠端代理或 MCP 工具組，請在 Python 代理程式碼中呼叫下列 AgentRegistry 用戶端方法：

  • 擷取遠端代理：使用 get_remote_a2a_agent()。
  • 擷取 MCP 工具集：使用 get_mcp_toolset()。

以下範例說明如何建構多代理系統，方法是建構一個協調器代理，利用已註冊的旅行社和已註冊的 Compute Engine MCP 伺服器。在本範例中，驗證作業是由代理的身分處理，但您也可以使用其他方法，例如 API 金鑰和 OAuth 2.0。詳情請參閱「向工具和資源進行驗證」。

如要使用視覺化低程式碼介面設計及自動化調度管理代理，請直接在 Gemini Enterprise Agent Platform 畫布的 Agent Studio 中設定已註冊的子代理。詳情請參閱「在 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.')

更改下列內容：

  • AGENT_ID：註冊代理人的 ID。
  • SERVER_ID：已註冊 MCP 伺服器的 ID。
  • MODEL_ID：自動調度管理代理的模型 ID，例如 gemini-1.5-flash。

重複使用代理的最佳做法

為盡量縮短網路延遲時間，請在應用程式啟動時，從登錄檔擷取代理和工具集一次，而非在每次叫用時呼叫 get_remote_a2a_agent()。

一個代理一次只能有一個上層代理。如果嘗試將同一個擷取的代理執行個體指派給多個協調器，ADK 可能會擲回錯誤，指出代理已有父項。

如要在多個上層代理中重複使用探索到的代理，請使用 .clone() 方法建立代理物件的新例項。

以下範例說明如何擷取代理一次，並複製該代理，以便在不同的自動調度管理工具中使用：

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

更改下列內容：

  • PROJECT_ID：管理專案的專案 ID。
  • LOCATION：登錄檔區域或位置，例如 us-central1。
  • AGENT_ID：註冊代理人的 ID。

後續步驟