Menggunakan agen Agent2Agent (A2A)

Sebelum memulai

Tutorial ini mengasumsikan bahwa Anda telah membaca dan mengikuti petunjuk di:

Mendapatkan instance agen

Untuk membuat kueri A2aAgent, Anda harus terlebih dahulu membuat instance baru atau mendapatkan instance yang ada.

Untuk mendapatkan A2aAgent yang sesuai dengan ID resource tertentu:

SDK Agent Platform 

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)

di mana

A2A Python SDK

Metode ini menggunakan A2A Python SDK resmi, yang menyediakan library klien untuk berinteraksi dengan agen yang sesuai dengan A2A. Untuk mengetahui informasi selengkapnya, lihat dokumentasi A2A Python SDK.

Pertama, instal SDK:

pip install a2a-sdk>=0.3.4

Kemudian, dapatkan kartu agen untuk membuat instance klien. A2AClient menangani penemuan dan komunikasi untuk Anda.

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)

Library permintaan Python

Protokol A2A dibangun di atas endpoint HTTP standar. Anda dapat berinteraksi dengan endpoint ini menggunakan klien HTTP apa pun.

Ambil URL A2A dari kartu agen dan tentukan header permintaan.

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",
}

Saat menggunakan SDK Agent Platform, objek remote_agent sesuai dengan class AgentEngine yang berisi hal berikut:

Bagian selanjutnya dari bagian ini mengasumsikan bahwa Anda memiliki instance AgentEngine, yang diberi nama remote_agent.

Operasi yang didukung

Agen A2A yang dihosting di Agent Runtime mengekspos serangkaian operasi yang sesuai langsung dengan endpoint API protokol A2A.

Mengambil kartu agen

Perhatikan bahwa Agent Runtime tidak menayangkan kartu agen publik. Untuk mengambil kartu agen yang diautentikasi:

SDK Agent Platform 

response = await remote_agent.handle_authenticated_agent_card()

A2A Python SDK

response = await a2a_client.get_card()

Library permintaan Python

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

Mengirim pesan

Untuk mengirim pesan:

SDK Agent Platform 

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)

Anda dapat mengganti waktu tunggu yang digunakan untuk on_message_send dengan menetapkan kolom timeout dari HttpOptions saat membuat vertexai.Client.

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)

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

Mendapatkan tugas

Untuk mendapatkan tugas dan statusnya

SDK Agent Platform 

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

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

Membatalkan tugas

Untuk membatalkan tugas:

SDK Agent Platform 

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

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

Langkah berikutnya