Agent2Agent 에이전트 사용

이 페이지에서는 에이전트 사용을 위한 일반 안내 외에도 A2A 에이전트와 관련된 기능을 설명합니다.

시작하기 전에

이 튜토리얼에서는 사용자가 다음 안내를 읽고 따랐다고 가정합니다.

지원되는 작업

Agent Engine에서 호스팅되는 A2A 에이전트는 A2A 프로토콜의 API 엔드포인트에 직접 해당하는 작업 집합을 노출합니다.

설정

Python용 Vertex AI SDK

동일한 구문을 사용하여 Vertex AI SDK for Python을 통해 Agent Engine에 배포된 A2A 에이전트와 상호작용할 수 있습니다.

Agent Engine으로 A2A를 설정하려면 배포된 에이전트의 인스턴스를 가져옵니다. 이 인스턴스는 기본 A2A 엔드포인트를 래핑하여 엔드포인트를 Python 메서드로 호출할 수 있도록 합니다.

import vertexai
from google.genai import types

# Replace with your actual values
PROJECT_ID = "your-project-id"
LOCATION = "your-location"
REASONING_ENGINE_ID = "your-reasoning-engine-id"
AGENT_ENGINE_RESOURCE = f"projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{REASONING_ENGINE_ID}"

client = vertexai.Client(
    project=PROJECT_ID,
    location=LOCATION,
    http_options=types.HttpOptions(
        api_version="v1beta1")
)

remote_agent = client.agent_engines.get(
    name=AGENT_ENGINE_RESOURCE,
)

A2A Python SDK

이 메서드는 A2A 규격 에이전트와 상호작용하기 위한 클라이언트 라이브러리를 제공하는 공식 A2A Python SDK를 사용합니다. 자세한 내용은 A2A Python SDK 문서를 참고하세요.

먼저 SDK를 설치합니다.

pip install a2a-sdk>=0.3.4

그런 다음 에이전트의 카드를 가져와 클라이언트 인스턴스를 만듭니다. A2AClient가 검색 및 통신을 처리합니다.

from google.auth import default
from google.auth.transport.requests import Request
from a2a.client import ClientConfig, ClientFactory
from a2a.types import TransportProtocol
import httpx

# We assume 'agent_card' is an existing AgentCard object.

# Fetch credentials for authentication for demo purpose. Use your own auth
credentials, _ = default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
credentials.refresh(Request())

# Create the client by chaining the factory and config initialization.
factory = ClientFactory(
    ClientConfig(
        supported_transports=[TransportProtocol.http_json], # only support http_json
        use_client_preference=True,
        httpx_client=httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {credentials.token}",
                "Content-Type": "application/json",
            }
        ),
    )
)
a2a_client = factory.create(agent_card)

Python requests 라이브러리

A2A 프로토콜은 표준 HTTP 엔드포인트에 기반합니다. HTTP 클라이언트를 사용하여 이러한 엔드포인트와 상호작용할 수 있습니다.

상담사 카드에서 A2A URL을 가져오고 요청 헤더를 정의합니다.

from google.auth import default
from google.auth.transport.requests import Request

# We assume 'agent_card' is an existing object
a2a_url = agent_card.url

# Get an authentication token for demonstration purposes. Use your own authentication mechanism.
credentials, _ = default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
credentials.refresh(Request())

headers = {
    "Authorization": f"Bearer {credentials.token}",
    "Content-Type": "application/json",
}

상담사 카드 가져오기

에이전트 엔진은 공개 에이전트 카드를 제공하지 않습니다. 인증된 에이전트 카드를 가져오려면 다음을 실행하세요.

Python용 Vertex AI SDK 

response = await remote_agent.handle_authenticated_agent_card()

A2A Python SDK

response = await a2a_client.get_card()

Python requests 라이브러리

card_endpoint = f"{a2a_url}/v1/card"
response = httpx.get(card_endpoint, headers=headers)
print(json.dumps(response.json(), indent=4))

메시지 보내기

메시지를 보내려면 다음 단계를 따르세요.

Python용 Vertex AI SDK 

message_data = {
  "messageId": "remote-agent-message-id",
  "role": "user",
  "parts": [{"kind": "text", "text": "What is the exchange rate from USD to EUR today?"}],
}

response = await remote_agent.on_message_send(**message_data)

A2A Python SDK

from a2a.types import Message, Part, TextPart
import pprint

message = Message(
    message_id="remote-agent-message-id",
    role="user",
    parts=[Part(root=TextPart(text="What's the currency rate of USD and EUR"))],
)

response_iterator = a2a_client.send_message(message)

async for chunk in response_iterator:
    pprint.pp(chunk)

Python requests 라이브러리

import httpx
import json

endpoint = f"{a2a_url}/v1/message:send"

payload = {
    "message": {
        "messageId": "remote-agent-message-id",
        "role": "1",
        "content": [{"text": "What is the exchange rate from USD to EUR today?"}],
    },
    "metadata": {"source": "python_script"},
}

response = httpx.post(endpoint, json=payload, headers=headers)
print(json.dumps(response.json(), indent=4))

작업 가져오기

작업 및 상태를 가져오려면 다음을 실행하세요.

Python용 Vertex AI SDK 

task_data = {
    "id": task_id,
}

response = await remote_agent.on_get_task(**task_data)

A2A Python SDK

from a2a.types import TaskQueryParams

task_data ={
    "id":task_id,
}
response = await a2a_client.get_task(TaskQueryParams(**task_data))

Python requests 라이브러리

task_end_point = f"{a2a_url}/v1/tasks/{task_id}"
response = httpx.get(task_end_point, headers=headers)
print(json.dumps(response.json(), indent=4))

작업 취소

작업을 취소하려면 다음 단계를 따르세요.

Python용 Vertex AI SDK 

task_data = {
    "id": task_id,
}
response = await remote_agent.on_cancel_task(**task_data)

A2A Python SDK

from a2a.types import TaskQueryParams

task_data ={
    "id":task_id,
}
response = await a2a_client.cancel_task(TaskQueryParams(**task_data))

Python requests 라이브러리

task_end_point = f"{a2a_url}/v1/tasks/{task_id}:cancel"
response = httpx.post(task_end_point, headers=headers)
print(json.dumps(response.json(), indent=4))

다음 단계