Comece a usar 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 suporta 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 a autenticação SigV4 ou OAuth 2.0, tornando-o ideal para aplicativos como agentes de voz conversacionais em tempo real.

WebSocket Bibliotecas compatíveis

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 compatíveis (cabeçalhos SigV4, URL pré-assinado 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.

Essa flexibilidade permite que você use sua WebSocket implementação preferida em diferentes linguagens e estruturas de programação, garantindo 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 compatível com streaming bidirecional usando o SDK para Python bedrock-agentcore 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 o código do seu 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.

Teste a WebSocket conexão

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, digite Ctrl+C para interromper o agente.

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

Instalar ferramentas de implantação

Instale a AgentCore CLI:

npm install -g @aws/agentcore

Verifique a instalação:

agentcore --help

Crie um projeto e implante em AWS

Crie um novo projeto para seu agente de streaming bidirecional:

agentcore create

Implante seu agente:

agentcore deploy

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 um tempo de execução do 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 de token OAuth Bearer para conexões. WebSocket Para usar a autenticação OAuth, você precisa configurar o tempo de execução do agente com autorização JWT, conforme descrito na seção de amostra de autorização de entrada e acesso de saída do JWT 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 do portador seguindo a Etapa 4: Use o token do portador para invocar seu agente no guia do OAuth, você pode 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 handshake. Para oferecer suporte à autenticação OAuth a partir de 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 sentinel. 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) roteia 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. IDs de sessão diferentes 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 uma ID de sessão exclusiva para cada usuário ou conversa e transmita-a 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 mensagens entre o cliente e o agente. Isso inclui qualquer troca de WebSocket mensagens, como envio de dados de cliente para agente, recebimento de 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 interações contínuas.

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.

Parar sessão de tempo de execução

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

Observabilidade

O Amazon Bedrock AgentCore Observability ajuda você a rastrear, depurar e monitorar agentes que você hospeda 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ê passe 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 Transferir cabeçalhos personalizados para o Amazon Bedrock AgentCore Runtime.

Além disso, os 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 as conexões usam códigos de fechamento padrão para comunicação de erros. Os códigos de fechamento comuns incluem:

WebSocket versus outros protocolos

Quando usar WebSocket:

Quando usar HTTP:

Exemplos adicionais de introdução

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