Agent2Agent 에이전트 사용

시작하기 전에

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

에이전트 인스턴스 가져오기

A2aAgent를 쿼리하려면 먼저 새 인스턴스를 만들거나 기존 인스턴스를 가져와야 합니다.

특정 리소스 ID에 해당하는 A2aAgent를 가져오려면 다음 안내를 따르세요.

Python용 Vertex AI SDK 

import vertexai
from google.genai import types

PROJECT_ID = "PROJECT_ID"
LOCATION = "LOCATION"
RESOURCE_ID = "RESOURCE_ID"
RESOURCE_NAME = f"projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{RESOURCE_ID}"

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

remote_agent = client.agent_engines.get(name=RESOURCE_NAME)

print(remote_agent)

각 항목의 의미는 다음과 같습니다.

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 요청 라이브러리

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를 사용하는 경우 remote_agent 객체는 다음을 포함하는 AgentEngine 클래스에 해당합니다.

  • 배포된 에이전트에 관한 정보가 포함된 agent.api_resource agent.operation_schemas()를 호출하여 에이전트가 지원하는 작업 목록을 반환할 수도 있습니다. 자세한 내용은 지원되는 작업을 참고하세요.
  • 동기 서비스 상호작용을 허용하는 agent.api_client
  • 비동기 서비스 상호작용을 허용하는 agent.async_api_client

이 섹션의 나머지 부분에서는 remote_agent라는 이름의 AgentEngine 인스턴스가 있다고 가정합니다.

지원되는 작업

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

에이전트 카드 가져오기

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

Python용 Vertex AI SDK 

response = await remote_agent.handle_authenticated_agent_card()

A2A Python SDK

response = await a2a_client.get_card()

Python 요청 라이브러리

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 요청 라이브러리

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 요청 라이브러리

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 요청 라이브러리

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

다음 단계