Implementazione di server A2A in Runtime AgentCore

Amazon Bedrock AgentCore AgentCore Runtime consente di distribuire ed eseguire server Agent-to-Agent (A2A) nel Runtime. AgentCore Questa guida ti guida nella creazione, nel test e nella distribuzione del tuo primo server A2A.

In questa sezione, imparerai:

Per ulteriori informazioni su A2A, vedere Contratto di protocollo A2A.

In che modo Amazon Bedrock AgentCore supporta A2A

Il supporto AgentCore del protocollo A2A di Amazon Bedrock consente una perfetta integrazione con i server A2A fungendo da livello proxy trasparente. Una volta configurato per A2A, Amazon Bedrock AgentCore prevede che i container eseguano server HTTP stateless e in streaming sulla porta nel percorso principale (0.0.0.0:9000/), 9000 in linea con la configurazione predefinita del server A2A.

Il servizio offre un isolamento delle sessioni di livello aziendale mantenendo al contempo la trasparenza del protocollo: i JSON-RPC payload dell'API vengono trasferiti direttamente al contenitore A2A senza modifiche. InvokeAgentRuntime Questa architettura preserva le funzionalità standard del protocollo A2A, come il rilevamento integrato degli agenti tramite Agent Cards at /.well-known/agent-card.json and JSON-RPC communication, aggiungendo al contempo l'autenticazione aziendale (2.0) e la scalabilità. SigV4/OAuth

I principali fattori di differenziazione dagli altri protocolli sono la porta (9000 vs 8080 per HTTP), il mount path (/vs/invocations) e il meccanismo standardizzato di rilevamento degli agenti, che rendono Amazon Bedrock AgentCore una piattaforma di distribuzione ideale per gli agenti A2A negli ambienti di produzione.

Principali differenze rispetto ad altri protocolli:

Per ulteriori informazioni, consulta https://a2a-protocol.org/.

Utilizzo di A2A con Runtime AgentCore

In questo tutorial creerai, testerai e distribuirai un server A2A.

Prerequisiti

Fase 1: Crea il tuo progetto A2A

Questo esempio utilizza Strands Agents, ma la AgentCore CLI supporta anche progetti LangChain/LangGraph A2A con Google ADK.

Scaffold: il progetto

Esegui il seguente comando e seleziona Strands come framework quando richiesto:

agentcore create --protocol A2A

La CLI supporta un progetto completo con tutte le dipendenze e le configurazioni richieste. Il file generato main.py contiene il tuo server A2A:

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

Comprendere il codice

Per personalizzare questo agente, sostituite add_numbers lo strumento con i vostri strumenti e aggiornate il prompt di sistema.

Fase 2: Esegui il test del server A2A a livello locale

Esegui e testa il tuo server A2A in un ambiente di sviluppo locale.

Avvia il tuo server A2A

Avvia il tuo server A2A localmente utilizzando la CLI: AgentCore

agentcore dev

Questo apre l' AgentCore Agent Inspector nel tuo browser web. Per utilizzare invece la TUI basata su terminale, usa. agentcore dev --no-browser

In alternativa, puoi eseguire direttamente il server:

python main.py

Dovresti vedere un output che indica che il server è in esecuzione sulla porta9000.

Invoca l'agente

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 .

Recupero della carta dell'agente di test

Puoi testare localmente l'endpoint della carta agente:

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

È inoltre possibile testare il server distribuito utilizzando A2A Inspector come descritto in Test remoti con A2A Inspector.

Fase 3: Implementa il server A2A su Bedrock Runtime AgentCore

Configura il pool di utenti di Cognito per l'autenticazione

Prima della distribuzione, configura l'autenticazione per un accesso sicuro al server distribuito. Per istruzioni dettagliate sulla configurazione di Cognito, consulta Configurare il pool di utenti di Cognito per l'autenticazione. Ciò fornisce i token OAuth necessari per un accesso sicuro al server distribuito.

Esegui la distribuzione su AWS

Implementa il tuo agente:

agentcore deploy

Questo comando consentirà di:

Dopo la distribuzione, riceverai un ARN di runtime dell'agente simile a:

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

Fase 4: Ottieni la carta dell'agente

Le Agent Card sono documenti di metadati JSON che descrivono l'identità, le capacità, le competenze, l'endpoint di servizio e i requisiti di autenticazione di un server A2A. Consentono l'individuazione automatica degli agenti nell'ecosistema A2A.

Impostazione delle variabili di ambiente

Impostazione delle variabili di ambiente

Recupera la carta dell'agente

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

Dopo aver ottenuto l'URL dalla Agent Card, esporta AGENTCORE_RUNTIME_URL come variabile di ambiente:

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

Fase 5: richiama il server A2A distribuito

Crea codice client per richiamare il server Amazon Bedrock AgentCore A2A distribuito e invia messaggi per testarne la funzionalità.

Crea un nuovo file per richiamare il server my_a2a_client_remote.py A2A distribuito:

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

Appendice

Configura il pool di utenti di Cognito per l'autenticazione

Per istruzioni dettagliate sulla configurazione di Cognito, consulta Configurare il pool di utenti Cognito per l'autenticazione nella documentazione MCP.

Test a distanza con A2A Inspector

Per informazioni, consulta https://github.com/a2aproject/a2a-inspector.

Risoluzione dei problemi

Problemi comuni A2A-specific

Di seguito sono riportati i problemi più comuni che potresti riscontrare:

Gestione delle eccezioni

Specifiche A2A per la gestione degli errori: https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors

I server A2A restituiscono gli errori come risposte di JSON-RPC errore standard con codici di stato HTTP 200. Gli errori interni di runtime vengono automaticamente tradotti in errori JSON-RPC interni per mantenere la conformità del protocollo.

Il servizio ora fornisce risposte di A2A-compliant errore corrette con codici di JSON-RPC errore standardizzati: