Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Implemente servidores A2A en AgentCore tiempo de ejecución

Amazon Bedrock AgentCore Runtime le permite implementar y ejecutar servidores Agent-to-Agent (A2A) en tiempo de ejecución. AgentCore Esta guía le 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 el 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 transmisibles en el puerto 9000 de la ruta raíz (0.0.0.0:9000/), que se alinea con la configuración predeterminada del servidor A2A.

El servicio proporciona un aislamiento de sesión de nivel empresarial y, al mismo tiempo, mantiene la transparencia del protocolo: JSON-RPC las cargas útiles de la InvokeAgentRuntime API se transfieren directamente al contenedor A2A sin modificaciones. Esta arquitectura conserva las funciones estándar del protocolo A2A, como la detección integrada de agentes 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 factores que lo diferencian de otros protocolos son el puerto (9000 frente al 8080 para HTTP), la ruta de montaje (/frente /invocations a) y el mecanismo estandarizado de detección de agentes, lo que convierte a Amazon Bedrock en AgentCore una plataforma de implementación ideal para los agentes A2A en entornos de producción.

Diferencias clave 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 Google LangChain/LangGraph ADK.

Organiza el proyecto

Use el siguiente comando:

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

La CLI estructura un proyecto completo con todas las dependencias y configuraciones necesarias. 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))

Comprender el código

Para personalizar este agente, sustituya la add_numbers herramienta por sus propias herramientas 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 terminales, 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 el 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 terminal 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

Configure 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 de configuración detalladas 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 su agente:

agentcore deploy

Este comando hará lo siguiente:

Tras la implementación, recibirá un ARN en 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 el descubrimiento automático de agentes en el ecosistema A2A.

Configure las variables de entorno

Configure las variables de entorno

Recupera la tarjeta del 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: Invoca tu servidor A2A desplegado

Cree un código de cliente para invocar su 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 su 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

Configure 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 la gestión de errores: https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors

Los servidores A2A devuelven la mayoría de los errores como respuestas de JSON-RPC error estándar. El servicio devuelve los errores de autenticación y autorización (por ejemploAccessDeniedException) como errores HTTP nativos con sus propios códigos de estado, como se muestra en la tabla siguiente. El servicio traduce automáticamente los errores de tiempo de ejecución internos en errores JSON-RPC internos para mantener el cumplimiento del protocolo.

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