Comece com o streaming bidirecional usando WebSocket

O Amazon Bedrock AgentCore Runtime permite que você implante agentes que suportam WebSocket streaming para comunicação bidirecional em tempo real. Este guia explica como criar, testar e implantar seu primeiro agente de streaming bidirecional usando. WebSocket

Nesta seção, você aprende:

Para obter mais informações sobre o WebSocket protocolo, consulte WebSocket RFC 6455.

Como o AgentCore Runtime oferece suporte às WebSocket conexões

AgentCore O WebSocket suporte do Runtime permite conexões de streaming persistentes e bidirecionais entre clientes e agentes. AgentCore O Runtime espera que os contêineres implementem WebSocket endpoints 8080 na porta do /ws caminho, o que se alinha às práticas padrão WebSocket do servidor.

AgentCore O WebSocket suporte do Runtime fornece os mesmos recursos sem servidor, de isolamento de sessão, identidade e observabilidade do. InvokeAgentRuntime Além disso, ele permite o streaming bidirecional de mensagens em tempo real e de baixa latência por meio de WebSocket conexões usando autenticação SigV4 ou OAuth 2.0, tornando-o ideal para aplicativos como agentes de voz conversacionais em tempo real.

WebSocket Bibliotecas suportadas

O streaming bidirecional usado WebSockets no AgentCore Runtime oferece suporte a aplicativos que usam qualquer biblioteca de WebSocket idiomas. Os únicos requisitos são que os clientes se conectem ao endpoint do serviço com uma conexão de WebSocket protocolo:

wss://bedrock-agentcore.<region>.amazonaws.com/runtimes/<agentRuntimeArn>/ws

usando um dos métodos de autenticação suportados (cabeçalhos SigV4, URL pré-assinada SigV4 ou OAuth 2.0) e que o aplicativo do agente implemente o contrato de serviço conforme especificado no WebSocket contrato de protocolo HTTP. Contrato de protocolo HTTP

Essa flexibilidade permite que você use sua WebSocket implementação preferida em diferentes linguagens de programação e estruturas, garantindo a compatibilidade com bases de código e fluxos de trabalho de desenvolvimento existentes.

Usando WebSocket com AgentCore Runtime

Neste tutorial de introdução, você criará, testará e implantará um aplicativo de agente que suporta streaming bidirecional usando o SDK básico do Python e a CLI para implantação. AgentCore

Pré-requisitos

Antes de começar, verifique se você tem:

Etapa 1: configurar o projeto e instalar dependências

Crie uma pasta de projeto e instale os pacotes necessários:

mkdir agentcore-runtime-quickstart-websocket
cd agentcore-runtime-quickstart-websocket
python3 -m venv .venv
source .venv/bin/activate

Atualize o pip para a versão mais recente:

pip install --upgrade pip

Instale os seguintes pacotes necessários:

pip install bedrock-agentcore

Etapa 2: crie seu agente de streaming bidirecional

Crie um arquivo de origem para seu código de agente de streaming bidirecional chamado. websocket_echo_agent.py Adicione o seguinte código:

from bedrock_agentcore import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

@app.websocket
async def websocket_handler(websocket, context):
    """Simple echo WebSocket handler."""
    await websocket.accept()

    try:
        data = await websocket.receive_json()
        # Echo back
        await websocket.send_json({"echo": data})
    except Exception as e:
        print(f"Error: {e}")
    finally:
        await websocket.close()

if __name__ == "__main__":
    app.run(log_level="info")

Crie requirements.txt e adicione o seguinte:

bedrock-agentcore

A dependência da websockets biblioteca python está incluída

Entendendo o código

Etapa 3: teste seu agente de streaming bidirecional localmente

Inicie seu agente de streaming bidirecional

Abra uma janela de terminal e inicie seu agente de streaming bidirecional com o seguinte comando:

python websocket_echo_agent.py

Você deve ver uma saída indicando que o servidor está sendo executado na porta 8080.

WebSocket Conexão de teste

Crie um WebSocket cliente local chamadowebsocket_agent_client.py:

import asyncio
import websockets
import json

async def local_websocket():
    uri = "ws://localhost:8080/ws"

    try:
        async with websockets.connect(uri) as websocket:
            # Send a message
            await websocket.send(json.dumps({"inputText": "Hello WebSocket!"}))

            # Receive the echo response
            response = await websocket.recv()
            print(f"Received: {response}")
    except Exception as e:
        print(f"Connection failed: {e}")

if __name__ == "__main__":
    asyncio.run(local_websocket())

Teste seu agente de streaming bidirecional localmente abrindo outra janela do terminal e executando o cliente:

python websocket_agent_client.py

Sucesso: Você deve ver uma resposta comoReceived: {"echo":{"inputText":"Hello WebSocket!"}}. Na janela do terminal que está executando o agente, insira Ctrl+C para interromper o agente.

Etapa 4: implante seu agente de streaming bidirecional no Runtime AgentCore

Instale ferramentas de implantação

Instale a AgentCore CLI:

npm install -g @aws/agentcore

Verifique a instalação:

agentcore --version

Para obter os comandos e as opções disponíveis, consulte a referência da AgentCore CLI.

Crie um projeto e implante no AWS

Crie um novo projeto para seu agente de streaming bidirecional:

cd ..
agentcore create --project-name WebSocketProject --no-agent
cd WebSocketProject
agentcore add agent \
  --name WebSocketAgent \
  --type byo \
  --language Python \
  --framework Strands \
  --model-provider Bedrock \
  --memory none \
  --protocol HTTP \
  --code-location ../agentcore-runtime-quickstart-websocket \
  --entrypoint websocket_echo_agent.py

Implante seu agente:

agentcore deploy

O AgentCore projeto faz referência ao diretório agentcore-runtime-quickstart-websocket de origem existente.

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/websocket_echo_agent-xyz123

Salve esse ARN, pois você precisará dele para invocar seu agente implantado.

Etapa 5: invocar seu agente de streaming bidirecional implantado

Configurar variáveis de ambiente

Configure as variáveis de ambiente necessárias:

Métodos de autenticação

A ação InvokeAgentRuntimeWithWebSocketStream da API estabelece uma WebSocket conexão que suporta streaming bidirecional entre o cliente e o agente. Você pode autenticar WebSocket conexões usando os seguintes métodos:

Conecte-se usando cabeçalhos assinados SigV4

O exemplo a seguir mostra como estabelecer uma WebSocket conexão e se comunicar com o tempo de execução de um agente usando cabeçalhos assinados SigV4:

from bedrock_agentcore.runtime import AgentCoreRuntimeClient
import websockets
import asyncio
import json
import os

async def main():
    # Get runtime ARN from environment variable
    runtime_arn = os.getenv('AGENT_ARN')
    if not runtime_arn:
        raise ValueError("AGENT_ARN environment variable is required")

    # Initialize client
    client = AgentCoreRuntimeClient(region="us-west-2")

    # Generate WebSocket connection with authentication
    ws_url, headers = client.generate_ws_connection(
        runtime_arn=runtime_arn
    )

    try:
        async with websockets.connect(ws_url, additional_headers=headers) as ws:
            # Send message
            await ws.send(json.dumps({"inputText": "Hello!"}))

            # Receive response
            response = await ws.recv()
            print(f"Received: {response}")
    except websockets.exceptions.InvalidStatus as e:
        print(f"WebSocket handshake failed with status code: {e.response.status_code}")
        print(f"Response headers: {e.response.headers}")
        print(f"Response body: {e.response.body.decode()}")
    except Exception as e:
        print(f"Connection failed: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Execute o cliente para testar seu agente implantado:

python websocket_agent_client_sigv4_headers.py

Sucesso: Você deve ver uma resposta como:

Received: {"echo":{"inputText":"Hello!"}}

Conecte-se usando URL pré-assinado (SigV4 por meio de parâmetros de consulta)

O exemplo a seguir mostra como criar uma WebSocket URL com parâmetros de consulta SigV4 e estabelecer uma conexão:

from bedrock_agentcore.runtime import AgentCoreRuntimeClient
import websockets
import asyncio
import json
import os

async def main():
    runtime_arn = os.getenv('AGENT_ARN')
    if not runtime_arn:
        raise ValueError("AGENT_ARN environment variable is required")

    client = AgentCoreRuntimeClient(region="us-west-2")

    # Generate WebSocket pre-signed URL (with SigV4 via query parameters)
    # wss://...amazonaws.com/runtimes/.../ws?X-Amz-Algorithm=AWS4-HMAC-SHA256
    #   &X-Amz-Credential=...&X-Amz-Date=...&X-Amz-Expires=300
    #   &X-Amz-SignedHeaders=...&X-Amz-Signature=...
    sigv4_url = client.generate_presigned_url(
        runtime_arn=runtime_arn,
        expires=300  # 5 minutes
    )

    try:
        async with websockets.connect(sigv4_url) as ws:
            await ws.send(json.dumps({"inputText": "Hello!"}))
            response = await ws.recv()
            print(f"Received: {response}")
    except websockets.exceptions.InvalidStatus as e:
        print(f"WebSocket handshake failed with status code: {e.response.status_code}")
        print(f"Response headers: {e.response.headers}")
        print(f"Response body: {e.response.body.decode()}")
    except Exception as e:
        print(f"Connection failed: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Execute o cliente para testar seu agente implantado:

python websocket_agent_client_sigv4_query_parameters.py

Sucesso: Você deve ver uma resposta como:

Received: {"echo":{"inputText":"Hello!"}}

Conecte-se usando OAuth

AgentCore O Runtime oferece suporte à autenticação do token OAuth Bearer para conexões. WebSocket Para usar a autenticação OAuth, você precisa configurar o tempo de execução do agente com a autorização JWT, conforme descrito na seção de amostra de autorização de entrada do JWT e acesso de saída do OAuth em Autenticar e autorizar com autenticação de entrada e autenticação de saída.

Depois de concluir a configuração do OAuth e obter um token portador seguindo a Etapa 4: Use o token portador para invocar seu agente no guia do OAuth, você poderá usar esse token para estabelecer conexões. WebSocket

Cliente Python com OAuth

O exemplo a seguir mostra como estabelecer uma WebSocket conexão do Python usando o OAuth:

from bedrock_agentcore.runtime import AgentCoreRuntimeClient
import websockets
import asyncio
import json
import os

async def main():
    # Get runtime ARN from environment variable
    runtime_arn = os.getenv('AGENT_ARN')
    if not runtime_arn:
        raise ValueError("AGENT_ARN environment variable is required")

    # Get OAuth bearer token from environment variable
    bearer_token = os.getenv('BEARER_TOKEN')
    if not bearer_token:
        raise ValueError("BEARER_TOKEN environment variable required for OAuth")

    # Initialize client
    client = AgentCoreRuntimeClient(region="us-west-2")

    # Generate WebSocket connection with OAuth
    ws_url, headers = client.generate_ws_connection_oauth(
        runtime_arn=runtime_arn,
        bearer_token=bearer_token
    )

    try:
        async with websockets.connect(ws_url, additional_headers=headers) as ws:
            # Send message
            await ws.send(json.dumps({"inputText": "Hello!"}))

            # Receive response
            response = await ws.recv()
            print(f"Received: {response}")
    except websockets.exceptions.InvalidStatus as e:
        print(f"WebSocket handshake failed with status code: {e.response.status_code}")
        print(f"Response headers: {e.response.headers}")
        print(f"Response body: {e.response.body.decode()}")
    except Exception as e:
        print(f"Connection failed: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Execute o cliente para testar seu agente implantado:

python websocket_agent_client_oauth.py

Sucesso: Você deve ver uma resposta como:

Received: {"echo":{"inputText":"Hello!"}}

JavaScript Cliente de navegador com OAuth

A WebSocket API nativa do navegador não fornece um método para definir cabeçalhos personalizados durante o aperto de mão. Para oferecer suporte à autenticação OAuth dos navegadores, o AgentCore Runtime aceita o token do portador incorporado no Sec-WebSocket-Protocol cabeçalho durante o handshake. WebSocket

O token deve ser codificado em base64url e prefixado combase64UrlBearerAuthorization., seguido pelo subprotocolo sentinela. base64UrlBearerAuthorization

O exemplo a seguir mostra como estabelecer uma WebSocket conexão a partir do navegador JavaScript usando o OAuth:

<!DOCTYPE html>
<html>
<body>
    <button onclick="connect()">Connect</button>
    <div id="output"></div>

    <script>
        function connect() {
            const bearerToken = "your_oauth_token_here";
            const runtimeArn = "arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/agent-xyz123";

            // Base64url encode token
            const base64url = btoa(bearerToken)
                .replace(/\+/g, '-')
                .replace(/\//g, '_')
                .replace(/=/g, '');

            const ws = new WebSocket(
                `wss://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/${runtimeArn}/ws`,
                [`base64UrlBearerAuthorization.${base64url}`, "base64UrlBearerAuthorization"]
            );

            ws.onopen = () => ws.send(JSON.stringify({ inputText: "Hello!" }));
            ws.onmessage = (e) => document.getElementById("output").innerText = e.data;
        }
    </script>
</body>
</html>

Gerenciamento de sessões

Fornecer um session_id (X-Amzn-Bedrock-AgentCore-Runtime-Session-Id) na WebSocket conexão (como parâmetro de consulta de URL ou cabeçalho de solicitação) encaminha a conexão para uma sessão de tempo de execução isolada. O agente pode acessar o contexto da conversa armazenado nessa sessão para implementar a continuidade de uma conversa referenciando interações anteriores. Diferentes IDs de sessão acessam contextos isolados separados, garantindo isolamento completo entre usuários ou conversas.

Para um gerenciamento abrangente do ciclo de vida da sessão, incluindo rastreamento, limpeza e tratamento de erros, consulte Usar sessões isoladas para agentes.

Usando sessões com WebSocket conexões

Para usar sessões com WebSocket conexões, gere um ID de sessão exclusivo para cada usuário ou conversa e passe-o ao estabelecer a conexão:

Ao usar o mesmo ID de sessão para WebSocket conexões relacionadas, você garante que o contexto seja mantido na mesma conversa, permitindo que seu agente forneça respostas coerentes baseadas em interações anteriores.

Ciclo de vida da sessão com conexões WebSocket

Para WebSocket conexões, o tempo limite de inatividade da sessão é redefinido sempre que há atividade de mensagem entre o cliente e o agente. Isso inclui qualquer troca de WebSocket mensagens, como enviar dados de cliente para agente, receber respostas de agente para cliente ou WebSocket ping/pong frames. Isso significa que WebSocket as conversas ativas manterão a sessão ativa enquanto as mensagens continuarem fluindo, evitando o encerramento prematuro da sessão durante as interações em andamento.

Para obter mais informações sobre como definir as configurações do ciclo de vida, consulte Definir as configurações do ciclo de vida do Amazon Bedrock AgentCore . Para um controle mais direto do ciclo de vida da sessão por meio do status de integridade do agente, consulte Gerenciamento do ciclo de vida da sessão em tempo de execução.

Interromper a sessão de execução

Para interromper uma sessão em execução antes do configurável IdleRuntimeSessionTimeout (o padrão é 15 minutos), consulte Interromper uma sessão em execução.

Observabilidade

O Amazon Bedrock AgentCore Observability ajuda você a rastrear, depurar e monitorar agentes hospedados no Amazon Bedrock Runtime. AgentCore Primeiro, habilite a Pesquisa de CloudWatch Transações seguindo as instruções em Habilitando a observabilidade do tempo de AgentCore execução do Amazon Bedrock. Para observar seu agente, consulte Visualizar dados de observabilidade de seus agentes do Amazon Bedrock AgentCore .

Para WebSocket conexões, um rastreamento representa a sessão de conexão completa em vez de trocas de mensagens individuais.

Cabeçalhos personalizados

Os cabeçalhos personalizados permitem que você transmita informações contextuais do seu aplicativo diretamente para o código do agente na conexão inicial WebSocket . Para obter informações completas sobre suporte, configuração e limitações de cabeçalhos personalizados, consulte Passe cabeçalhos personalizados para o Amazon Bedrock AgentCore Runtime.

Além disso, cabeçalhos prefixados com X-Amzn-Bedrock-AgentCore-Runtime-Custom- podem ser passados como parâmetros de consulta de URL nas WebSocket conexões.

Por exemplo, você pode passar cabeçalhos personalizados como parâmetros de consulta no WebSocket URL:

wss://bedrock-agentcore.<region>.amazonaws.com/runtimes/<agentRuntimeArn>/ws?X-Amzn-Bedrock-AgentCore-Runtime-Custom-TestHeader=query-param-test-value

O contêiner do aplicativo do agente os receberá como cabeçalhos:

"headers": {
    "x-amzn-bedrock-agentcore-runtime-custom-testheader": "query-param-test-value"
  }

Apêndice

Considerações sobre segurança

Solução de problemas

WebSocket-specific Problemas comuns

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

Tratamento de erros

WebSocket os erros surgem em duas fases, dependendo de quando eles ocorrem.

Estabelecimento da conexão (antes da WebSocket atualização)

Abrir a conexão é uma solicitação HTTP padrão. O código de status HTTP reflete a exceção e o cabeçalho da x-amzn-ErrorType resposta carrega o nome da exceção. O serviço pode retornar qualquer um dos seguintes erros antes de estabelecer a WebSocket conexão.

Conexão ativa (após a WebSocket atualização)

Depois que o WebSocket é estabelecido, os erros são comunicados com códigos de WebSocket fechamento padrão em vez de códigos de status HTTP. Os códigos de fechamento comuns incluem:

WebSocket versus outros protocolos

Quando usar WebSocket:

Quando usar HTTP:

Exemplos adicionais de introdução

Para ver outros exemplos de uso de streaming WebSocket bidirecional com AgentCore Runtime, consulte os exemplos de streaming WebSocket bidirecional: GitHub