A2A-Server in AgentCore Runtime bereitstellen

Mit Amazon Bedrock AgentCore AgentCore Runtime können Sie Agent-to-Agent (A2A) -Server in der AgentCore Runtime bereitstellen und ausführen. Dieser Leitfaden führt Sie durch die Erstellung, das Testen und die Bereitstellung Ihres ersten A2A-Servers.

In diesem Abschnitt lernen Sie:

Weitere Informationen zu A2A finden Sie unter A2A-Protokollvertrag.

So AgentCore unterstützt Amazon Bedrock A2A

AgentCoreDie A2A-Protokollunterstützung von Amazon Bedrock ermöglicht eine nahtlose Integration mit A2A-Servern, indem sie als transparente Proxyschicht fungiert. Bei der Konfiguration für A2A AgentCore erwartet Amazon Bedrock, dass Container statuslose, streambare HTTP-Server auf dem Port 9000 im Root-Pfad (0.0.0.0:9000/) ausführen, was der standardmäßigen A2A-Serverkonfiguration entspricht.

Der Service bietet Sitzungsisolierung auf Unternehmensebene bei gleichzeitiger Wahrung der Protokolltransparenz — JSON-RPC Payloads von der InvokeAgentRuntimeAPI werden ohne Änderung direkt an den A2A-Container weitergeleitet. Bei dieser Architektur werden die standardmäßigen A2A-Protokollfunktionen wie die integrierte Agentenerkennung über Agentenkarten und die JSON-RPC Kommunikation beibehalten /.well-known/agent-card.json und gleichzeitig Unternehmensauthentifizierung (2.0) und Skalierbarkeit hinzugefügt. SigV4/OAuth

Die wichtigsten Unterscheidungsmerkmale zu anderen Protokollen sind der Port (9000 gegenüber 8080 für HTTP), der Bereitstellungspfad (/vs/invocations) und der standardisierte Agentenerkennungsmechanismus, was Amazon Bedrock zu AgentCore einer idealen Bereitstellungsplattform für A2A-Agenten in Produktionsumgebungen macht.

Hauptunterschiede zu anderen Protokollen:

Weitere Informationen finden Sie unter https://a2a-protocol.org/.

Verwenden von A2A mit Runtime AgentCore

In diesem Tutorial erstellen, testen und implementieren Sie einen A2A-Server.

Voraussetzungen

Schritt 1: Erstellen Sie Ihr A2A-Projekt

In diesem Beispiel werden Strands Agents verwendet, aber die AgentCore CLI unterstützt auch A2A-Projekte mit LangChain/LangGraph und Google ADK.

Richten Sie das Projekt ein Gerüst ein

Führen Sie den folgenden Befehl aus und wählen Sie Strands als Framework aus, wenn Sie dazu aufgefordert werden:

agentcore create --protocol A2A

Die CLI bildet das Gerüst für ein vollständiges Projekt mit allen erforderlichen Abhängigkeiten und Konfigurationen. Das generierte main.py enthält Ihren A2A-Server:

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

Den Code verstehen

Um diesen Agenten anzupassen, ersetzen Sie das add_numbers Tool durch Ihre eigenen Tools und aktualisieren Sie die Systemaufforderung.

Schritt 2: Testen Sie Ihren A2A-Server lokal

Führen Sie Ihren A2A-Server in einer lokalen Entwicklungsumgebung aus und testen Sie ihn.

Starten Sie Ihren A2A-Server

Starten Sie Ihren A2A-Server lokal mit der AgentCore CLI:

agentcore dev

Dadurch wird der AgentCore Agent Inspector in Ihrem Webbrowser geöffnet. Um stattdessen die terminalbasierte TUI zu verwenden, verwenden Sie. agentcore dev --no-browser

Alternativ können Sie den Server direkt ausführen:

python main.py

Sie sollten eine Ausgabe sehen, die besagt, dass der Server auf dem Port läuft9000.

Rufen Sie den Agenten auf

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 .

Testen Sie den Kartenabruf für den Agenten

Sie können den Endpunkt der Agentenkarte lokal testen:

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

Sie können Ihren bereitgestellten Server auch mit dem A2A Inspector testen, wie unter Remote Testing with A2A Inspector beschrieben.

Schritt 3: Stellen Sie Ihren A2A-Server auf Bedrock Runtime bereit AgentCore

Cognito-Benutzerpool für die Authentifizierung einrichten

Konfigurieren Sie vor der Bereitstellung die Authentifizierung für den sicheren Zugriff auf Ihren bereitgestellten Server. Eine ausführliche Anleitung zur Einrichtung von Cognito finden Sie unter Cognito-Benutzerpool für die Authentifizierung einrichten. Dadurch werden die OAuth-Token bereitgestellt, die für den sicheren Zugriff auf Ihren bereitgestellten Server erforderlich sind.

Bereitstellen auf AWS

Stellen Sie Ihren Agenten bereit:

agentcore deploy

Dieser Befehl wird:

Nach der Bereitstellung erhalten Sie einen Runtime-ARN für Agenten, der wie folgt aussieht:

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

Schritt 4: Besorgen Sie sich die Agentenkarte

Agentenkarten sind JSON-Metadatendokumente, die die Identität, Fähigkeiten, den Service-Endpunkt und die Authentifizierungsanforderungen eines A2A-Servers beschreiben. Sie ermöglichen die automatische Agentenerkennung im A2A-Ökosystem.

Umgebungsvariablen festlegen

Umgebungsvariablen festlegen

Agentenkarte abrufen

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

Nachdem Sie die URL von der Agent-Karte abgerufen haben, exportieren Sie AGENTCORE_RUNTIME_URL sie als Umgebungsvariable:

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

Schritt 5: Rufen Sie Ihren bereitgestellten A2A-Server auf

Erstellen Sie einen Client-Code, um Ihren bereitgestellten Amazon Bedrock AgentCore A2A-Server aufzurufen und Nachrichten zu senden, um die Funktionalität zu testen.

Erstellen Sie eine neue Datei, my_a2a_client_remote.py um Ihren bereitgestellten A2A-Server aufzurufen:

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

Anhang

Cognito-Benutzerpool für die Authentifizierung einrichten

Detaillierte Anweisungen zur Einrichtung von Cognito finden Sie in der MCP-Dokumentation unter Cognito-Benutzerpool für die Authentifizierung einrichten.

Ferntests mit A2A Inspector

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

Fehlerbehebung

Häufig auftretende Probleme A2A-specific

Im Folgenden sind häufig auftretende Probleme aufgeführt:

Behandlung von Ausnahmen

A2A-Spezifikationen für die Fehlerbehandlung: https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors

A2A-Server geben Fehler als JSON-RPC Standardfehlerantworten mit HTTP 200-Statuscodes zurück. Interne Laufzeitfehler werden automatisch in JSON-RPC interne Fehler umgewandelt, um die Einhaltung der Protokolle zu gewährleisten.

Der Service bietet jetzt korrekte A2A-compliant Fehlerantworten mit standardisierten JSON-RPC Fehlercodes: