Menyebarkan server A2A di Runtime AgentCore

Amazon Bedrock AgentCore AgentCore Runtime memungkinkan Anda menerapkan dan menjalankan server Agent-to-Agent (A2A) di Runtime. AgentCore Panduan ini memandu Anda melalui pembuatan, pengujian, dan penerapan server A2A pertama Anda.

Di bagian ini, Anda belajar:

Untuk informasi lebih lanjut tentang A2A, lihat kontrak protokol A2A.

Bagaimana Amazon Bedrock AgentCore mendukung A2A

Dukungan protokol AgentCore A2A Amazon Bedrock memungkinkan integrasi tanpa batas dengan server A2A dengan bertindak sebagai lapisan proxy transparan. Saat dikonfigurasi untuk A2A, Amazon Bedrock AgentCore mengharapkan container untuk menjalankan server HTTP stateless dan streamable pada port 9000 di root path (0.0.0.0:9000/), yang sejajar dengan konfigurasi server A2A default.

Layanan ini menyediakan isolasi sesi tingkat perusahaan sambil mempertahankan transparansi protokol - JSON-RPC muatan dari InvokeAgentRuntimeAPI diteruskan langsung ke wadah A2A tanpa modifikasi. Arsitektur ini mempertahankan fitur protokol A2A standar seperti penemuan agen bawaan melalui Kartu Agen /.well-known/agent-card.json dan JSON-RPC komunikasi, sambil menambahkan otentikasi perusahaan (SigV4/OAuth 2.0) dan skalabilitas.

Pembeda utama dari protokol lain adalah port (9000 vs 8080 untuk HTTP), jalur pemasangan (/vs/invocations), dan mekanisme penemuan agen standar, menjadikan Amazon Bedrock platform penyebaran yang AgentCore ideal untuk agen A2A di lingkungan produksi.

Perbedaan utama dari protokol lain:

Untuk informasi selengkapnya, lihat https://a2a-protocol.org/.

Menggunakan A2A dengan Runtime AgentCore

Dalam tutorial ini Anda membuat, menguji, dan menyebarkan server A2A.

Prasyarat

Langkah 1: Buat proyek A2A Anda

Contoh ini menggunakan Strands Agents, tetapi AgentCore CLI juga mendukung proyek A2A dengan LangChain/LangGraph dan Google ADK.

Perancah proyek

Jalankan perintah berikut dan pilih Strands sebagai framework Anda saat diminta:

agentcore create --protocol A2A

CLI membuat proyek lengkap dengan semua dependensi dan konfigurasi yang diperlukan. Yang dihasilkan main.py berisi server A2A Anda:

from strands import Agent, tool
from strands.multiagent.a2a.executor import StrandsA2AExecutor
from bedrock_agentcore.runtime import serve_a2a
from model.load import load_model

@tool
def add_numbers(a: int, b: int) -> int:
    """Return the sum of two numbers."""
    return a + b

tools = [add_numbers]

agent = Agent(
    model=load_model(),
    system_prompt="You are a helpful assistant. Use tools when appropriate.",
    tools=tools,
)

if __name__ == "__main__":
    serve_a2a(StrandsA2AExecutor(agent))

Memahami kode

Untuk menyesuaikan agen ini, ganti add_numbers alat dengan alat Anda sendiri dan perbarui prompt sistem.

Langkah 2: Uji server A2A Anda secara lokal

Jalankan dan uji server A2A Anda di lingkungan pengembangan lokal.

Mulai server A2A Anda

Mulai server A2A Anda secara lokal menggunakan CLI: AgentCore

agentcore dev

Ini membuka inspektur AgentCore agen di browser web Anda. Untuk menggunakan TUI berbasis terminal sebagai gantinya, gunakan. agentcore dev --no-browser

Atau, Anda dapat menjalankan server secara langsung:

python main.py

Anda akan melihat output yang menunjukkan server berjalan pada port9000.

Memohon agen

curl -X POST http://localhost:9000/ \
-H "Content-Type: application/json" \
-d '{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "what is 101 * 11?"
        }
      ],
      "messageId": "12345678-1234-1234-1234-123456789012"
    }
  }
}' | jq .

Pengambilan kartu agen uji

Anda dapat menguji titik akhir kartu agen secara lokal:

curl http://localhost:9000/.well-known/agent-card.json | jq.

Anda juga dapat menguji server yang digunakan menggunakan Inspector A2A seperti yang dijelaskan dalam Pengujian jarak jauh dengan inspektur A2A.

Langkah 3: Terapkan server A2A Anda ke Bedrock Runtime AgentCore

Siapkan kumpulan pengguna Cognito untuk otentikasi

Sebelum menerapkan, konfigurasikan otentikasi untuk akses aman ke server yang Anda gunakan. Untuk petunjuk penyiapan Cognito yang mendetail, lihat Mengatur kumpulan pengguna Cognito untuk autentikasi. Ini menyediakan token OAuth yang diperlukan untuk akses aman ke server yang Anda gunakan.

Menyebarkan ke AWS

Menyebarkan agen Anda:

agentcore deploy

Perintah ini akan:

Setelah penerapan, Anda akan menerima ARN runtime agen yang terlihat seperti:

arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123

Langkah 4: Dapatkan kartu agen

Kartu Agen adalah dokumen metadata JSON yang menggambarkan identitas, kemampuan, keterampilan, titik akhir layanan, dan persyaratan otentikasi server A2A. Mereka memungkinkan penemuan agen otomatis di ekosistem A2A.

Mengatur variabel lingkungan

Mengatur variabel lingkungan

Ambil kartu agen

import os
import json
import requests
from uuid import uuid4
from urllib.parse import quote

def fetch_agent_card():
    # Get environment variables
    agent_arn = os.environ.get('AGENT_ARN')
    bearer_token = os.environ.get('BEARER_TOKEN')

    if not agent_arn:
        print("Error: AGENT_ARN environment variable not set")
        return

    if not bearer_token:
        print("Error: BEARER_TOKEN environment variable not set")
        return

    # URL encode the agent ARN
    escaped_agent_arn = quote(agent_arn, safe='')

    # Construct the URL
    url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{escaped_agent_arn}/invocations/.well-known/agent-card.json"

    # Generate a unique session ID
    session_id = str(uuid4())
    print(f"Generated session ID: {session_id}")

    # Set headers
    headers = {
        'Accept': '*/*',
        'Authorization': f'Bearer {bearer_token}',
        'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id': session_id
    }

    try:
        # Make the request
        response = requests.get(url, headers=headers)
        response.raise_for_status()

        # Parse and pretty print JSON
        agent_card = response.json()
        print(json.dumps(agent_card, indent=2))

        return agent_card

    except requests.exceptions.RequestException as e:
        print(f"Error fetching agent card: {e}")
        return None

if __name__ == "__main__":
    fetch_agent_card()

Setelah Anda mendapatkan URL dari Kartu Agen, ekspor AGENTCORE_RUNTIME_URL sebagai variabel lingkungan:

export AGENTCORE_RUNTIME_URL="https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/<ARN>/invocations/"

Langkah 5: Panggil server A2A yang Anda gunakan

Buat kode klien untuk memanggil server Amazon Bedrock AgentCore A2A yang Anda gunakan dan kirim pesan untuk menguji fungsionalitas.

Buat file baru my_a2a_client_remote.py untuk memanggil server A2A yang Anda gunakan:

import asyncio
import logging
import os
from uuid import uuid4

import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, TextPart

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

DEFAULT_TIMEOUT = 300  # set request timeout to 5 minutes

def create_message(*, role: Role = Role.user, text: str) -> Message:
    return Message(
        kind="message",
        role=role,
        parts=[Part(TextPart(kind="text", text=text))],
        message_id=uuid4().hex,
    )

async def send_sync_message(message: str):
    # Get runtime URL from environment variable
    runtime_url = os.environ.get('AGENTCORE_RUNTIME_URL')

    # Generate a unique session ID
    session_id = str(uuid4())
    print(f"Generated session ID: {session_id}")

    # Add authentication headers for Amazon Bedrock AgentCore
    headers = {"Authorization": f"Bearer {os.environ.get('BEARER_TOKEN')}",
        'X-Amzn-Bedrock-AgentCore-Runtime-Session-Id': session_id}

    async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, headers=headers) as httpx_client:
        # Get agent card from the runtime URL
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=runtime_url)
        agent_card = await resolver.get_agent_card()

        # Agent card contains the correct URL (same as runtime_url in this case)
        # No manual override needed - this is the path-based mounting pattern

        # Create client using factory
        config = ClientConfig(
            httpx_client=httpx_client,
            streaming=False,  # Use non-streaming mode for sync response
        )
        factory = ClientFactory(config)
        client = factory.create(agent_card)

        # Create and send message
        msg = create_message(text=message)

        # With streaming=False, this will yield exactly one result
        async for event in client.send_message(msg):
            if isinstance(event, Message):
                logger.info(event.model_dump_json(exclude_none=True, indent=2))
                return event
            elif isinstance(event, tuple) and len(event) == 2:
                # (Task, UpdateEvent) tuple
                task, update_event = event
                logger.info(f"Task: {task.model_dump_json(exclude_none=True, indent=2)}")
                if update_event:
                    logger.info(f"Update: {update_event.model_dump_json(exclude_none=True, indent=2)}")
                return task
            else:
                # Fallback for other response types
                logger.info(f"Response: {str(event)}")
                return event

# Usage - Uses AGENTCORE_RUNTIME_URL environment variable
asyncio.run(send_sync_message("what is 101 * 11"))

Lampiran

Siapkan kumpulan pengguna Cognito untuk otentikasi

Untuk petunjuk penyiapan Cognito yang mendetail, lihat Mengatur kumpulan pengguna Cognito untuk autentikasi dalam dokumentasi MCP.

Pengujian jarak jauh dengan inspektur A2A

Lihat https://github.com/a2aproject/a2a-inspector.

Pemecahan masalah

A2A-specific Masalah umum

Berikut ini adalah masalah umum yang mungkin Anda temui:

Penanganan pengecualian

Spesifikasi A2A untuk penanganan Kesalahan: https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors

Server A2A mengembalikan kesalahan sebagai respons JSON-RPC kesalahan standar dengan kode status HTTP 200. Kesalahan Runtime internal secara otomatis diterjemahkan ke kesalahan JSON-RPC internal untuk mempertahankan kepatuhan protokol.

Layanan ini sekarang memberikan respons A2A-compliant kesalahan yang tepat dengan kode JSON-RPC kesalahan standar: