Comment Amazon Bedrock AgentCore prend en charge l'A2A Utiliser A2A avec Runtime AgentCore Annexe

Déploiement de serveurs A2A dans Runtime AgentCore

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

Dans cette section, vous allez apprendre :

Comment Amazon Bedrock AgentCore prend en charge l'A2A
Comment créer un serveur A2A doté de fonctionnalités d'agent
Comment tester votre serveur localement
Comment déployer votre serveur sur AWS
Comment invoquer votre serveur déployé
Comment récupérer les cartes d'agent à des fins de découverte

Pour plus d'informations sur le protocole A2A, consultez la section Contrat de protocole A2A.

Rubriques

Comment Amazon Bedrock AgentCore prend en charge l'A2A
Utiliser A2A avec Runtime AgentCore
Annexe

Comment Amazon Bedrock AgentCore prend en charge l'A2A

La prise en charge AgentCore du protocole A2A d'Amazon Bedrock permet une intégration fluide 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 sans état et diffusables sur le port situé 9000 au niveau du chemin racine (0.0.0.0:9000/), ce qui correspond à la configuration par défaut du serveur A2A.

Le service fournit une isolation de session de niveau entreprise 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 via des cartes d'agent /.well-known/agent-card.json et JSON-RPC la 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 plateforme de déploiement idéale pour les agents A2A dans les environnements de production.

Principales différences par rapport aux autres protocoles :

Port: Les serveurs A2A fonctionnent sur le port 9000 (contre 8080 pour HTTP, 8000 pour MCP)
Chemin: Les serveurs A2A sont montés sur / (vs /invocations pour HTTP, /mcp pour MCP)
Cartes d'agent: A2A fournit une découverte intégrée des agents via des cartes d'agent sur /.well-known/agent-card.json
Protocole: Utilisations JSON-RPC de la communication d'agent à agent
Authentification: Supporte les schémas d'authentification SIGv4 et OAuth 2.0

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

Utiliser A2A avec Runtime AgentCore

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

Rubriques

Conditions préalables
Étape 1 : Créez votre projet A2A
Étape 2 : Testez votre serveur A2A localement
Étape 3 : Déployez votre serveur A2A sur Bedrock Runtime AgentCore
Étape 4 : Obtenir la carte d'agent
Étape 5 : Invoquez votre serveur A2A déployé

Conditions préalables

Python 3.10 ou supérieur installé et compréhension de base de Python
Node.js 20 ou version ultérieure installée (requise pour la AgentCore CLI)
L' AgentCore interface de ligne de commande installée : npm install -g @aws/agentcore
Un AWS compte avec les autorisations appropriées et les informations d'identification locales configurées
Compréhension du protocole A2A et des concepts de communication agent-agent

É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.

Échafaudez le projet

Exécutez la commande suivante :


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

La CLI propose 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

Strands Agent: Crée un agent doté d'outils et de fonctionnalités spécifiques
Executeur Strand SA2A: Encapsule l'agent Strands pour assurer la compatibilité du protocole A2A
serve_a2a: L'assistant du AgentCore SDK Amazon Bedrock qui démarre un Bedrock-compatible serveur A2A. Il gère le point de terminaison de /ping santé, le service de la carte d'agent, la variable d'AGENTCORE_RUNTIME_URLenvironnement, la propagation de l'en-tête Bedrock et s'exécute sur le port 9000 par défaut.
Portée 9000: Les serveurs A2A s'exécutent sur le port 9000 par défaut dans Runtime AgentCore

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 l' AgentCore interface de ligne de commande :


agentcore dev

Cela ouvre l'inspecteur de l' AgentCore agent dans votre navigateur Web. Pour utiliser le TUI basé sur le 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 un 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 de l'agent de test

Vous pouvez tester le point de terminaison de la carte d'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

Configurer le groupe d'utilisateurs Cognito pour l'authentification

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

Déployez vers AWS

Déployez votre agent :


agentcore deploy

Cette commande va :

Regroupez le code de votre agent et ses dépendances
Téléchargez l'artefact de déploiement sur Amazon S3
Création d'un environnement d'exécution Amazon Bedrock AgentCore
Déployez votre agent pour AWS

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


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

Exportez le jeton du porteur en tant que variable d'environnement. Pour la configuration du jeton porteur, voir Configuration du jeton porteur.
```
export BEARER_TOKEN="<BEARER_TOKEN>"
```

Exportez l'ARN de l'agent.


export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_a2a_server-xyz123"

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

Après avoir obtenu l'URL de la carte d'agent, exportez 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 envoyer des messages pour tester la fonctionnalité.

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

Rubriques

Configurer le groupe d'utilisateurs Cognito pour l'authentification
Test à distance avec l'inspecteur A2A
Résolution des problèmes

Configurer le groupe d'utilisateurs Cognito pour l'authentification

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

Test à 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 :

Conflits portuaires: Les serveurs A2A doivent fonctionner sur le port 9000 dans l' AgentCore environnement d'exécution
JSON-RPC erreurs: Vérifiez que votre client envoie des messages JSON-RPC 2.0 correctement formatés
Incompatibilité entre les méthodes d'autorisation: Assurez-vous que votre demande utilise la même méthode d'authentification (OAuth ou Sigv4) que celle avec laquelle l'agent a été configuré

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 la plupart des erreurs sous forme de réponses JSON-RPC d'erreur standard. Le service renvoie les échecs d'authentification et d'autorisation (par exempleAccessDeniedException) sous forme d'erreurs HTTP natives avec leurs propres codes d'état, comme indiqué dans le tableau suivant. Le service traduit automatiquement les erreurs d'exécution internes en erreurs JSON-RPC internes afin de garantir la conformité du protocole.

Le service fournit des réponses aux A2A-compliant erreurs à l'aide de codes JSON-RPC d'erreur standardisés :

JSON-RPC Code d'erreur	Exception d'exécution	Code d'erreur HTTP	JSON-RPC Message d'erreur
Non applicable	`AccessDeniedException`	403	Accès refusé (renvoyé sous la forme d'une erreur HTTP standard, pas d' JSON-RPC erreur)
-32051	`ResourceNotFoundException`	404	Ressource introuvable — La ressource demandée n'existe pas
-32052	`ValidationException`	400	Erreur de validation — Données de demande non valides
-32053	`ThrottlingException`	429	Limite de débit dépassée : trop de demandes
-32053	`ServiceQuotaExceededException`	429	Limite de débit dépassée : trop de demandes
-32054	`ConflictException`	409	Conflit de ressources — La ressource existe déjà
-32054	`RetryableConflictException`	409	Fonctionnement de la session en cours, veuillez réessayer
-32055	`RuntimeClientError`	424	Erreur du client d'exécution : consultez vos CloudWatch journaux pour plus d'informations.
-32603	`Any other exception`	500	Erreur interne : une erreur inattendue s'est produite lors du traitement de la demande

Avertissement JavaScript est désactivé ou n'est pas disponible dans votre navigateur.

Pour que vous puissiez utiliser la documentation AWS, Javascript doit être activé. Vous trouverez des instructions sur les pages d'aide de votre navigateur.

Conventions de rédaction

Fonctionnalités du Stateful MCP

Déployer AG-UI des serveurs