엔드포인트 확인 및 오케스트레이터 빌드

에이전트 개발 키트 (ADK)는 Agent Registry 내에 카탈로그화된 AI 에이전트 및 모델 컨텍스트 프로토콜 (MCP) 서버를 프로그래매틱 방식으로 검색, 조회, 연결할 수 있는 전용 AgentRegistry 클라이언트를 제공합니다.

애플리케이션에 엔드포인트 URL을 하드코딩하는 대신 ADK를 사용하여 런타임에 이러한 엔드포인트를 확인할 수 있습니다.

Agent Registry는 기본 엔드포인트를 제공하지만 프로덕션 배포는 일반적으로 Agent Gateway를 통해 이러한 호출을 라우팅합니다. Agent Gateway를 사용하면 보안 정책을 적용하고, 프로토콜 중재를 실행하고, 검색한 도구에 콘텐츠 필터링을 적용할 수 있습니다.

이 문서에서는 에이전트 레지스트리에서 원격 에이전트 및 MCP 도구 모음을 가져와 상위 오케스트레이터 에이전트에 포함하는 방법을 설명합니다.

시작하기 전에

ADK를 Agent Registry와 통합하기 전에 다음을 완료하세요.

  1. 프로젝트에서 Agent Registry를 설정합니다.

  2. 필요한 A2A 종속 항목을 사용하여 ADK를 최신 버전으로 설치하거나 업그레이드합니다.

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

다음 단계