Implemente servidores A2A em Runtime AgentCore

O Amazon Bedrock AgentCore AgentCore Runtime permite que você implante e execute servidores Agent-to-Agent (A2A) no Runtime. AgentCore Este guia explica como criar, testar e implantar seu primeiro servidor A2A.

Nesta seção, você aprende:

Para obter mais informações sobre A2A, consulte Contrato de protocolo A2A.

Como o Amazon Bedrock AgentCore oferece suporte ao A2A

O suporte ao protocolo AgentCore A2A do Amazon Bedrock permite uma integração perfeita com servidores A2A, atuando como uma camada de proxy transparente. Quando configurado para A2A, o Amazon Bedrock AgentCore espera que os contêineres executem servidores HTTP sem estado e streamáveis 9000 na porta do caminho raiz (0.0.0.0:9000/), que se alinha à configuração padrão do servidor A2A.

O serviço fornece isolamento de sessão de nível corporativo, mantendo a transparência do protocolo - JSON-RPC as cargas da InvokeAgentRuntimeAPI são passadas diretamente para o contêiner A2A sem modificação. Essa arquitetura preserva os recursos padrão do protocolo A2A, como descoberta integrada de agentes por meio de cartões de agente /.well-known/agent-card.json e JSON-RPC comunicação, ao mesmo tempo em que adiciona autenticação corporativa (SigV4/OAuth 2.0) e escalabilidade.

Os principais diferenciais de outros protocolos são a porta (9000 versus 8080 para HTTP), o caminho de montagem (/vs/invocations) e o mecanismo padronizado de descoberta de agentes, tornando o Amazon Bedrock AgentCore uma plataforma de implantação ideal para agentes A2A em ambientes de produção.

Principais diferenças em relação a outros protocolos:

Para obter mais informações, consulte https://a2a-protocol.org/.

Usando A2A com Runtime AgentCore

Neste tutorial, você cria, testa e implanta um servidor A2A.

Pré-requisitos

Etapa 1: Crie seu projeto A2A

Este exemplo usa Strands Agents, mas a AgentCore CLI também oferece suporte a projetos A2A com LangChain/LangGraph o Google ADK.

Anime o projeto

Execute o comando a seguir e selecione Strands como sua estrutura quando solicitado:

agentcore create --protocol A2A

A CLI estrutura um projeto completo com todas as dependências e configurações necessárias. O gerado main.py contém seu 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))

Entendendo o código

Para personalizar esse agente, substitua a add_numbers ferramenta por suas próprias ferramentas e atualize o prompt do sistema.

Etapa 2: Teste seu servidor A2A localmente

Execute e teste seu servidor A2A em um ambiente de desenvolvimento local.

Inicie seu servidor A2A

Inicie seu servidor A2A localmente usando a CLI AgentCore :

agentcore dev

Isso abre o inspetor de AgentCore agentes em seu navegador da web. Para usar a TUI baseada em terminal em vez disso, use. agentcore dev --no-browser

Como alternativa, você pode executar o servidor diretamente:

python main.py

Você deve ver a saída indicando que o servidor está sendo executado na porta9000.

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

Recuperação do cartão do agente de teste

Você pode testar o endpoint da placa do agente localmente:

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

Você também pode testar seu servidor implantado usando o Inspetor A2A, conforme descrito em Teste remoto com o inspetor A2A.

Etapa 3: Implantar seu servidor A2A no Bedrock Runtime AgentCore

Configurar o grupo de usuários do Cognito para autenticação

Antes da implantação, configure a autenticação para acesso seguro ao seu servidor implantado. Para obter instruções detalhadas de configuração do Cognito, consulte Configurar o grupo de usuários do Cognito para autenticação. Isso fornece os tokens OAuth necessários para acesso seguro ao seu servidor implantado.

Implemente em AWS

Implante seu agente:

agentcore deploy

Esse comando irá:

Após a implantação, você receberá um ARN de tempo de execução do agente que se parece com:

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

Etapa 4: obter o cartão de agente

Os cartões de agente são documentos de metadados JSON que descrevem a identidade, os recursos, as habilidades, o endpoint de serviço e os requisitos de autenticação de um servidor A2A. Eles permitem a descoberta automática de agentes no ecossistema A2A.

Configurar variáveis de ambiente

Configurar variáveis de ambiente

Recupere o cartão do 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()

Depois de obter o URL do Agent Card, exporte AGENTCORE_RUNTIME_URL como uma variável de ambiente:

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

Etapa 5: invocar seu servidor A2A implantado

Crie um código de cliente para invocar seu servidor Amazon Bedrock AgentCore A2A implantado e envie mensagens para testar a funcionalidade.

Crie um novo arquivo my_a2a_client_remote.py para invocar seu servidor A2A implantado:

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 o grupo de usuários do Cognito para autenticação

Para obter instruções detalhadas de configuração do Cognito, consulte Configurar o grupo de usuários do Cognito para autenticação na documentação do MCP.

Teste remoto com inspetor A2A

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

Solução de problemas

A2A-specific Problemas comuns

A seguir estão os problemas comuns que você pode encontrar:

Tratamento de exceções

Especificações A2A para tratamento de erros: https://a2a-protocol.org/latest/specification/#81-standard-json-rpc-errors

Os servidores A2A retornam erros como respostas de JSON-RPC erro padrão com códigos de status HTTP 200. Os erros internos de tempo de execução são automaticamente traduzidos em erros JSON-RPC internos para manter a conformidade do protocolo.

O serviço agora fornece respostas de A2A-compliant erro adequadas com códigos de JSON-RPC erro padronizados: