Implemente servidores A2A en tiempo de ejecución AgentCore

Amazon Bedrock AgentCore AgentCore Runtime le permite implementar y ejecutar servidores Agent-to-Agent (A2A) en Runtime. AgentCore Esta guía explica cómo crear, probar e implementar su primer servidor A2A.

En esta sección, aprenderá lo siguiente:

Para obtener más información sobre la A2A, consulte el contrato de protocolo A2A.

Cómo Amazon Bedrock AgentCore apoya el A2A

La compatibilidad con el protocolo A2A AgentCore de Amazon Bedrock permite una integración perfecta con los servidores A2A al actuar como una capa de proxy transparente. Cuando se configura para A2A, Amazon Bedrock AgentCore espera que los contenedores ejecuten servidores HTTP sin estado y con capacidad de transmisión en el puerto de la ruta raíz (0.0.0.0:9000/), lo que se 9000 alinea con la configuración predeterminada del servidor A2A.

El servicio proporciona un aislamiento de sesiones de nivel empresarial y, al mismo tiempo, mantiene la transparencia del protocolo: JSON-RPC las cargas útiles de la InvokeAgentRuntimeAPI se transfieren directamente al contenedor A2A sin modificaciones. Esta arquitectura conserva las funciones estándar del protocolo A2A, como la detección de agentes integrada mediante tarjetas de agente /.well-known/agent-card.json y la JSON-RPC comunicación, al tiempo que añade la autenticación empresarial (SigV4/OAuth 2.0) y la escalabilidad.

Los principales diferenciadores de otros protocolos son el puerto (9000 frente a 8080 para HTTP), la ruta de montaje y el mecanismo estandarizado de descubrimiento de agentes, lo que convierte a Amazon Bedrock en AgentCore una plataforma de implementación ideal para los agentes de A2A en entornos de producción. / /invocations

Principales diferencias con respecto a otros protocolos:

Para obtener más información, consulte https://a2a-protocol.org/.

Uso de A2A con Runtime AgentCore

En este tutorial, creará, probará e implementará un servidor A2A.

Requisitos previos

Paso 1: Crea tu proyecto A2A

Este ejemplo usa Strands Agents, pero la AgentCore CLI también admite proyectos A2A con LangChain/LangGraph Google ADK.

Organiza el proyecto

Ejecuta el siguiente comando y selecciona Strands como marco cuando se te pida:

agentcore create --protocol A2A

La CLI estructura un proyecto completo con todas las dependencias y configuraciones requeridas. El generado main.py contiene su servidor 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))

Entendiendo el código

Para personalizar este agente, sustituya la add_numbers herramienta por las suyas propias y actualice el indicador del sistema.

Paso 2: Pruebe su servidor A2A localmente

Ejecute y pruebe su servidor A2A en un entorno de desarrollo local.

Inicie su servidor A2A

Inicie su servidor A2A localmente mediante la AgentCore CLI:

agentcore dev

Esto abre el inspector de AgentCore agentes en su navegador web. Para utilizar en su lugar la TUI basada en el terminal, utilice. agentcore dev --no-browser

Como alternativa, puede ejecutar el servidor directamente:

python main.py

Debería ver un resultado que indica que el servidor se está ejecutando en el puerto9000.

Invoca al 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 .

Pruebe la recuperación de la tarjeta del agente

Puede probar el punto final de la tarjeta de agente de forma local:

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

También puede probar el servidor desplegado con el Inspector A2A, tal y como se describe en Pruebas remotas con el inspector A2A.

Paso 3: Implemente su servidor A2A en Bedrock Runtime AgentCore

Configurar el grupo de usuarios de Cognito para la autenticación

Antes de la implementación, configure la autenticación para un acceso seguro al servidor implementado. Para obtener instrucciones detalladas de configuración de Cognito, consulte Configurar el grupo de usuarios de Cognito para la autenticación. Esto proporciona los tokens de OAuth necesarios para un acceso seguro al servidor implementado.

Implemente en AWS

Despliegue a su agente:

agentcore deploy

Este comando hará lo siguiente:

Tras la implementación, recibirá un ARN de tiempo de ejecución del agente con el siguiente aspecto:

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

Paso 4: Obtenga la tarjeta de agente

Las tarjetas de agente son documentos de metadatos JSON que describen la identidad, las capacidades, las habilidades, el punto final del servicio y los requisitos de autenticación de un servidor A2A. Permiten la detección automática de agentes en el ecosistema A2A.

Configure las variables de entorno

Configure las variables de entorno

Recupera la tarjeta de 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()

Después de obtener la URL de la tarjeta de agente, expórtela AGENTCORE_RUNTIME_URL como una variable de entorno:

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

Paso 5: invoque el servidor A2A desplegado

Cree un código de cliente para invocar el servidor Amazon Bedrock AgentCore A2A implementado y envíe mensajes para probar la funcionalidad.

Cree un archivo nuevo my_a2a_client_remote.py para invocar el servidor A2A implementado:

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

Apéndice

Configurar el grupo de usuarios de Cognito para la autenticación

Para obtener instrucciones detalladas de configuración de Cognito, consulte Configurar el grupo de usuarios de Cognito para la autenticación en la documentación de MCP.

Pruebas remotas con el inspector A2A

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

Resolución de problemas

Problemas comunes A2A-specific

Los siguientes son problemas comunes que pueden surgir:

Gestión de excepciones

Especificaciones A2A para el manejo de errores: https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors

Los servidores A2A devuelven los errores como respuestas de JSON-RPC error estándar con códigos de estado HTTP 200. Los errores de tiempo de ejecución internos se traducen automáticamente en errores JSON-RPC internos para mantener el cumplimiento del protocolo.

El servicio ahora proporciona respuestas de A2A-compliant error adecuadas con códigos de JSON-RPC error estandarizados: