解析端點並建構協調器

Agent Development Kit (ADK) 提供專屬的 AgentRegistry 用戶端，可讓您以程式輔助方式探索、查詢及連線至 Agent Registry 中編目的 AI 代理和 Model Context Protocol (MCP) 伺服器。

您可以使用 ADK 在執行階段解析這些端點，不必將端點網址硬式編碼到應用程式中。

Agent Registry 提供基礎端點，但實際工作環境部署作業通常會透過 Agent Gateway 轉送這些呼叫。 Agent Gateway 可協助您強制執行安全政策、執行通訊協定調解，以及對您探索到的工具套用內容過濾功能。

本文說明如何從 Agent Registry 擷取遠端代理和 MCP 工具組，並將其納入父項協調器代理。

事前準備

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

  1. 在專案中設定 Agent Registry。

  2. 安裝或升級至最新版 ADK，並加入必要的 A2A 依附元件：

    pip 

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

    uv 

    uv add "google-adk[a2a]"

    你必須升級至至少 google-adk>=1.29.0。

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

    gcloud auth application-default login

ADC 憑證必須具備代理程式或工具互動的基礎服務所需的 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 函式一樣。

請使用下列方法：

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

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

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

重複使用代理程式的最佳做法

為盡量縮短網路延遲時間，請在應用程式啟動時從登錄檔擷取代理程式和工具組一次，而不是在每次呼叫時呼叫 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()]
)

後續步驟