Agent Development Kit（ADK）は、Agent Registry 内でカタログ化された AI エージェントと Model Context Protocol（MCP）サーバーをプログラムで検出、検索、接続できる専用の AgentRegistry クライアントを提供します。
エンドポイント URL をアプリケーションにハードコードする代わりに、ADK を使用して実行時にこれらのエンドポイントを解決できます。
Agent Registry は基盤となるエンドポイントを提供しますが、通常、本番環境のデプロイではこれらの呼び出しは Agent Gateway 経由でルーティングされます。Agent Gateway を使用すると、セキュリティ ポリシーの適用、プロトコル メディエーションの実行、検出したツールへのコンテンツ フィルタリングの適用を行うことができます。
このドキュメントでは、Agent Registry からリモート エージェントと MCP ツールセットを取得して、親オーケストレーター エージェントに含める方法について説明します。
始める前に
ADK を Agent Registry と統合する前に、次の操作を行います。
- プロジェクトで Agent Registry を設定します。
必要な A2A 依存関係を含む ADK の最新バージョンをインストールするか、最新バージョンにアップグレードします。
pip
pip install --upgrade "google-adk[a2a]"uv
uv add "google-adk[a2a]"少なくとも
google-adk>=1.29.0にアップグレードする必要があります。アプリケーションのデフォルト認証情報（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 などの他の方法を使用することもできます。詳細については、ツールとリソースに対して認証するをご覧ください。
ビジュアルなローコード インターフェースを使用してエージェントを設計、オーケストレートするには、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.')
エージェントを再利用するためのベスト プラクティス
ネットワーク レイテンシを最小限に抑えるには、呼び出しごとに get_remote_a2a_agent() を呼び出すのではなく、アプリケーションの起動時にレジストリからエージェントとツールセットを 1 回だけ取得します。
エージェントが一度に持つことのできる親エージェントは 1 つのみです。同じ取得済みエージェント インスタンスを複数のオーケストレーターに割り当てようとすると、ADK から、エージェントにすでに親があることを示すエラーが返されることがあります。
検出されたエージェントを複数の親エージェントで再利用するには、.clone() メソッドを使用してエージェント オブジェクトの新しいインスタンスを作成します。
次の例は、エージェントを 1 回取得し、さまざまなオーケストレータで使用するためにクローンを作成する方法を示しています。
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()]
)
次のステップ
- 解決済みのエンドポイントのトラフィックを転送するように Agent Gateway を設定する方法を学習する。
- カスタム ヘッダーまたはバインディングを使用してツールとリソースに対して認証を行う方法を学習する。
- Agent Registry でスキルを登録してスキルを管理する方法について説明します。