Menggunakan agen Agent2Agent

Selain petunjuk umum untuk menggunakan agen, halaman ini menjelaskan fitur yang khusus untuk agen A2A.

Sebelum memulai

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

Operasi yang didukung

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

Penyiapan

Vertex AI SDK untuk Python

Anda dapat berinteraksi dengan agen A2A yang di-deploy ke Agent Engine menggunakan Vertex AI SDK untuk Python, menggunakan sintaksis yang sama.

Untuk menyiapkan A2A dengan Agent Engine, dapatkan instance agen yang di-deploy. Instance ini membungkus endpoint A2A pokok, sehingga Anda dapat memanggil endpoint sebagai metode 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

Metode ini menggunakan A2A Python SDK resmi, yang menyediakan library klien untuk berinteraksi dengan agen yang kompatibel 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 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",
}

Mengambil kartu agen

Perhatikan bahwa Agent Engine tidak menyajikan kartu agen publik. Untuk mengambil kartu agen yang diautentikasi:

Vertex AI SDK untuk Python 

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

Kirim pesan

Untuk mengirim pesan:

Vertex AI SDK untuk Python 

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)

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

Vertex AI SDK untuk Python 

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:

Vertex AI SDK untuk Python 

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