Déployer des serveurs A2A dans Runtime AgentCore

Amazon Bedrock AgentCore AgentCore Runtime vous permet de déployer et d'exécuter des serveurs Agent-to-Agent (A2A) dans le AgentCore Runtime. Ce guide vous explique comment créer, tester et déployer votre premier serveur A2A.

Dans cette section, vous allez apprendre :

Pour plus d'informations sur l'A2A, consultez le contrat de protocole A2A.

Comment Amazon Bedrock AgentCore prend en charge l'A2A

La prise en charge AgentCore du protocole A2A d'Amazon Bedrock permet une intégration parfaite avec les serveurs A2A en agissant comme une couche proxy transparente. Lorsqu'ils sont configurés pour A2A, Amazon Bedrock AgentCore s'attend à ce que les conteneurs exécutent des serveurs HTTP apatrides et streamables sur le port situé sur le chemin racine (0.0.0.0:9000/), ce qui correspond 9000 à la configuration du serveur A2A par défaut.

Le service permet d'isoler les sessions de niveau professionnel tout en préservant la transparence du protocole : les JSON-RPC charges utiles de l'InvokeAgentRuntimeAPI sont transmises directement au conteneur A2A sans modification. Cette architecture préserve les fonctionnalités standard du protocole A2A, telles que la découverte intégrée des agents par le biais de cartes d'agent /.well-known/agent-card.json et de JSON-RPC communication, tout en ajoutant l'authentification d'entreprise (SigV4/OAuth 2.0) et l'évolutivité.

Les principaux facteurs de différenciation par rapport aux autres protocoles sont le port (9000 contre 8080 pour HTTP), le chemin de montage (/vs/invocations) et le mécanisme standardisé de découverte des agents, faisant d'Amazon Bedrock AgentCore une plate-forme de déploiement idéale pour les agents A2A dans les environnements de production.

Principales différences par rapport aux autres protocoles :

Pour de plus amples informations, veuillez consulter https://a2a-protocol.org/.

Utilisation d'A2A avec Runtime AgentCore

Dans ce didacticiel, vous allez créer, tester et déployer un serveur A2A.

Conditions préalables

Étape 1 : Créez votre projet A2A

Cet exemple utilise des agents Strands, mais la AgentCore CLI prend également en charge les projets A2A avec LangChain/LangGraph Google ADK.

Échafaudage du projet

Exécutez la commande suivante et sélectionnez Strands comme framework lorsque vous y êtes invité :

agentcore create --protocol A2A

La CLI échafaude un projet complet avec toutes les dépendances et configurations requises. Le fichier généré main.py contient votre serveur 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))

Comprendre le code

Pour personnaliser cet agent, remplacez l'add_numbersoutil par vos propres outils et mettez à jour l'invite du système.

Étape 2 : Testez votre serveur A2A localement

Exécutez et testez votre serveur A2A dans un environnement de développement local.

Démarrez votre serveur A2A

Démarrez votre serveur A2A localement à l'aide de la AgentCore CLI :

agentcore dev

L'inspecteur des AgentCore agents s'ouvre alors dans votre navigateur Web. Pour utiliser le TUI basé sur un terminal à la place, utilisez. agentcore dev --no-browser

Vous pouvez également exécuter le serveur directement :

python main.py

Vous devriez voir une sortie indiquant que le serveur fonctionne sur le port9000.

Invoquer l'agent

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 .

Récupération de la carte d'agent de test

Vous pouvez tester le point de terminaison de la carte agent localement :

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

Vous pouvez également tester votre serveur déployé à l'aide de l'inspecteur A2A, comme décrit dans Tests à distance avec l'inspecteur A2A.

Étape 3 : Déployez votre serveur A2A sur Bedrock Runtime AgentCore

Configuration du groupe d'utilisateurs Cognito pour l'authentification

Avant le déploiement, configurez l'authentification pour un accès sécurisé à votre serveur déployé. Pour obtenir des instructions détaillées sur la configuration de Cognito, voir Configurer le groupe d'utilisateurs de Cognito pour l'authentification. Cela fournit les jetons OAuth nécessaires pour un accès sécurisé à votre serveur déployé.

Déployer vers AWS

Déployez votre agent :

agentcore deploy

Cette commande permettra de :

Après le déploiement, vous recevrez un ARN d'exécution de l'agent qui ressemble à ce qui suit :

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

Étape 4 : Obtenir la carte d'agent

Les cartes d'agent sont des documents de métadonnées JSON qui décrivent l'identité, les capacités, les compétences, le point de terminaison du service et les exigences d'authentification d'un serveur A2A. Ils permettent la découverte automatique des agents dans l'écosystème A2A.

Configurer les variables d’environnement

Configurer les variables d’environnement

Récupérez la carte d'agent

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

Une fois que vous avez obtenu l'URL depuis la carte d'agent, exportez-la AGENTCORE_RUNTIME_URL en tant que variable d'environnement :

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

Étape 5 : Invoquez votre serveur A2A déployé

Créez un code client pour appeler votre serveur Amazon Bedrock AgentCore A2A déployé et envoyez des messages pour tester les fonctionnalités.

Créez un nouveau fichier my_a2a_client_remote.py pour appeler votre serveur A2A déployé :

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

Annexe

Configuration du groupe d'utilisateurs Cognito pour l'authentification

Pour obtenir des instructions détaillées sur la configuration de Cognito, consultez la section Configurer le groupe d'utilisateurs Cognito pour l'authentification dans la documentation MCP.

Tests à distance avec l'inspecteur A2A

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

Résolution des problèmes

A2A-specific Problèmes courants

Les problèmes courants que vous pouvez rencontrer sont les suivants :

Gestion des exceptions

Spécifications A2A pour la gestion des erreurs : https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors

Les serveurs A2A renvoient les erreurs sous forme de réponses JSON-RPC d'erreur standard avec des codes d'état HTTP 200. Les erreurs d'exécution internes sont automatiquement converties en erreurs JSON-RPC internes afin de garantir la conformité au protocole.

Le service fournit désormais des réponses aux A2A-compliant erreurs appropriées avec des codes JSON-RPC d'erreur standardisés :