Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Implementa i server A2A in Runtime AgentCore

Amazon Bedrock AgentCore Runtime ti consente di distribuire ed eseguire server Agent-to-Agent (A2A) nel Runtime. AgentCore Questa guida illustra come creare, testare e distribuire il tuo primo server A2A.

In questa sezione, imparerai:

Per ulteriori informazioni su A2A, vedere il contratto di protocollo A2A.

In che modo Amazon Bedrock supporta A2A AgentCore

Il supporto del protocollo A2A AgentCore di Amazon Bedrock consente una perfetta integrazione con i server A2A agendo come un livello proxy trasparente. Se configurato per A2A, Amazon Bedrock AgentCore prevede che i contenitori eseguano server HTTP stateless e streamable sulla porta nel percorso principale (), in linea con la configurazione predefinita del server 9000 A2A. 0.0.0.0:9000/

Il servizio fornisce un isolamento delle sessioni di livello aziendale mantenendo la trasparenza del protocollo: i JSON-RPC payload dall'API vengono trasmessi 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 e la JSON-RPC comunicazione, aggiungendo al contempo l'autenticazione aziendale (2.0) e la scalabilità. SigV4/OAuth

I principali elementi 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 agli altri protocolli:

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

Utilizzo di A2A con Runtime AgentCore

In questo tutorial crei, testa e distribuisci un server A2A.

Prerequisiti

Fase 1: Crea il tuo progetto A2A

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

Assembla il progetto

Esegui il comando seguente:

agentcore create \
  --project-name A2AProject \
  --name A2AAgent \
  --language Python \
  --framework Strands \
  --model-provider Bedrock \
  --memory none \
  --protocol A2A
cd A2AProject

La CLI impalca un progetto completo con tutte le dipendenze e le configurazioni richieste. Il 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))

Comprensione del codice

Per personalizzare questo agente, sostituisci add_numbers lo strumento con i tuoi strumenti e aggiorna il prompt di sistema.

Passaggio 2: testa il tuo server A2A localmente

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.

Richiama 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 da parte dell'agente di test

Puoi testare localmente l'endpoint della carta dell'agente:

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

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

Fase 3: Distribuisci il tuo server A2A su Bedrock Runtime AgentCore

Configura il pool di utenti 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 l'accesso sicuro al server distribuito.

Distribuisci 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: Ottenete 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 il rilevamento automatico 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/"

Passaggio 5: richiama il server A2A distribuito

Crea un codice client per richiamare il server Amazon Bedrock AgentCore A2A distribuito e inviare 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 Cognito per l'autenticazione

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

Test remoto con A2A Inspector

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

Risoluzione dei problemi

A2A-specific Problemi comuni

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 la maggior parte degli errori come risposte di errore standard. JSON-RPC Il servizio restituisce gli errori di autenticazione e autorizzazione (ad esempioAccessDeniedException) come errori HTTP nativi con i propri codici di stato, come mostrato nella tabella seguente. Il servizio traduce automaticamente gli errori interni di runtime in errori JSON-RPC interni per mantenere la conformità del protocollo.

Il servizio fornisce risposte agli errori con codici di A2A-compliant JSON-RPC errore standardizzati: