Inizia con lo streaming bidirezionale utilizzando WebSocket

Amazon Bedrock AgentCore Runtime ti consente di distribuire agenti che supportano WebSocket lo streaming per comunicazioni bidirezionali in tempo reale. Questa guida illustra come creare, testare e distribuire il tuo primo agente di streaming bidirezionale utilizzando. WebSocket

In questa sezione, imparerai:

Per ulteriori informazioni sul WebSocket protocollo, vedere WebSocket RFC 6455.

In che modo AgentCore Runtime supporta le connessioni WebSocket

AgentCore Il WebSocket supporto di Runtime consente connessioni di streaming persistenti e bidirezionali tra client e agenti. AgentCore Runtime si aspetta che i container implementino gli WebSocket endpoint sulla porta 8080 in corrispondenza del /ws percorso, in linea con le pratiche standard del server. WebSocket

AgentCore Il WebSocket supporto di Runtime fornisce le stesse funzionalità serverless di isolamento della sessione, identità e osservabilità di. InvokeAgentRuntime Inoltre, consente lo streaming bidirezionale in tempo reale e a bassa latenza di messaggi tramite WebSocket connessioni che utilizzano l'autenticazione SIGv4 o OAuth 2.0, rendendolo ideale per applicazioni come gli agenti vocali conversazionali in tempo reale.

Librerie supportate WebSocket

Lo streaming bidirezionale con WebSockets on AgentCore Runtime supporta le applicazioni che utilizzano qualsiasi libreria WebSocket linguistica. Gli unici requisiti sono che i client si connettano all'endpoint del servizio con una WebSocket connessione di protocollo:

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

utilizzando uno dei metodi di autenticazione supportati (intestazioni SIGv4, URL prefirmato SIGv4 o OAuth 2.0) e che l'applicazione agente implementi il contratto di servizio come specificato nel contratto di protocollo HTTP. WebSocket Contratto di protocollo HTTP

Questa flessibilità consente di utilizzare l' WebSocket implementazione preferita in diversi linguaggi e framework di programmazione, garantendo la compatibilità con le basi di codice e i flussi di lavoro di sviluppo esistenti.

Utilizzo con Runtime WebSocket AgentCore

In questo tutorial introduttivo creerai, testerai e distribuirai un'applicazione agente che supporta lo streaming bidirezionale utilizzando l'SDK Python bedrock-agentcore e la CLI per la distribuzione. AgentCore

Prerequisiti

Prima di iniziare, assicurati di avere:

Fase 1: Configurazione del progetto e installazione delle dipendenze

Crea una cartella di progetto e installa i pacchetti richiesti:

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

Aggiorna pip alla versione più recente:

pip install --upgrade pip

Installa i seguenti pacchetti richiesti:

pip install bedrock-agentcore

Passaggio 2: crea il tuo agente di streaming bidirezionale

Crea un file sorgente per il tuo agente di streaming bidirezionale denominato. websocket_echo_agent.py Aggiungi il codice seguente:

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

Crea requirements.txt e aggiungi quanto segue:

bedrock-agentcore

La dipendenza dalla websockets libreria python è inclusa

Comprensione del codice

Passaggio 3: testa localmente il tuo agente di streaming bidirezionale

Avvia il tuo agente di streaming bidirezionale

Apri una finestra di terminale e avvia il tuo agente di streaming bidirezionale con il seguente comando:

python websocket_echo_agent.py

Dovresti vedere un output che indica che il server è in esecuzione sulla porta 8080.

Connessione di prova WebSocket

Crea un WebSocket client locale denominatowebsocket_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())

Testa il tuo agente di streaming bidirezionale localmente aprendo un'altra finestra di terminale ed eseguendo il client:

python websocket_agent_client.py

Successo: dovresti vedere una risposta del tipo. Received: {"echo":{"inputText":"Hello WebSocket!"}} Nella finestra del terminale in cui è in esecuzione l'agente, digitate Ctrl+C per arrestare l'agente.

Passaggio 4: distribuisci il tuo agente di streaming bidirezionale su Runtime AgentCore

Installa gli strumenti di distribuzione

Installa la AgentCore CLI:

npm install -g @aws/agentcore

Verifica l'installazione:

agentcore --version

Per i comandi e le opzioni disponibili, vedere il riferimento alla AgentCore CLI.

Crea un progetto e distribuiscilo su AWS

Crea un nuovo progetto per il tuo agente di streaming bidirezionale:

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

Implementa il tuo agente:

agentcore deploy

Il AgentCore progetto fa riferimento alla directory di agentcore-runtime-quickstart-websocket origine esistente.

Dopo la distribuzione, riceverai un ARN di runtime dell'agente simile a:

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

Salva questo ARN perché ti servirà per richiamare l'agente distribuito.

Passaggio 5: richiama il tuo agente di streaming bidirezionale distribuito

Impostazione delle variabili di ambiente

Imposta le variabili di ambiente richieste:

Metodi di autenticazione

L'azione InvokeAgentRuntimeWithWebSocketStream API stabilisce una WebSocket connessione che supporta lo streaming bidirezionale tra client e agente. È possibile autenticare le WebSocket connessioni utilizzando i seguenti metodi:

Connettiti utilizzando intestazioni firmate Sigv4

L'esempio seguente mostra come stabilire una WebSocket connessione e comunicare con il runtime di un agente utilizzando intestazioni firmate 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())

Esegui il client per testare l'agente distribuito:

python websocket_agent_client_sigv4_headers.py

Successo: dovresti vedere una risposta del tipo:

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

Connettiti utilizzando un URL prefirmato (SIGv4 tramite parametri di interrogazione)

L'esempio seguente mostra come creare un WebSocket URL con parametri di query SigV4 e stabilire una connessione:

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

Esegui il client per testare l'agente distribuito:

python websocket_agent_client_sigv4_query_parameters.py

Successo: dovresti vedere una risposta del tipo:

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

Connettiti usando OAuth

AgentCore Runtime supporta l'autenticazione con token OAuth Bearer per le connessioni. WebSocket Per utilizzare l'autenticazione OAuth, è necessario configurare il runtime dell'agente con l'autorizzazione JWT come descritto nella sezione di esempio di autorizzazione JWT in entrata e accesso OAuth in uscita di Autenticazione e autorizzazione con autenticazione in entrata e autenticazione in uscita. Autentica e autorizza con Inbound Auth e Outbound Auth

Una volta completata la configurazione OAuth e ottenuto un token al portatore seguendo il passaggio 4: Usa il token al portatore per invocare il tuo agente nella guida OAuth, puoi utilizzare quel token per stabilire WebSocket connessioni.

Client Python con OAuth

L'esempio seguente mostra come stabilire una WebSocket connessione da Python usando 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())

Esegui il client per testare l'agente distribuito:

python websocket_agent_client_oauth.py

Successo: dovresti vedere una risposta del tipo:

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

JavaScript Client browser con OAuth

L' WebSocket API nativa del browser non fornisce un metodo per impostare intestazioni personalizzate durante l'handshake. Per supportare l'autenticazione OAuth dai browser, AgentCore Runtime accetta il token al portatore incorporato nell'intestazione durante l'Sec-WebSocket-Protocolhandshake. WebSocket

Il token deve essere codificato in base64url e deve essere preceduto dal sottoprotocollo sentinel. base64UrlBearerAuthorization. base64UrlBearerAuthorization

L'esempio seguente mostra come stabilire una connessione dal browser utilizzando OAuth: WebSocket JavaScript

<!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>

Gestione della sessione

Fornendo un session_id (X-Amzn-Bedrock-AgentCore-Runtime-Session-Id) sulla WebSocket connessione (come parametro di query URL o intestazione della richiesta), la connessione viene indirizzata a una sessione di runtime isolata. L'agente può accedere al contesto di conversazione memorizzato all'interno di quella sessione, per implementare la continuità di una conversazione facendo riferimento alle interazioni precedenti. ID di sessione diversi accedono a contesti isolati separati, garantendo il completo isolamento tra utenti o conversazioni.

Per una gestione completa del ciclo di vita delle sessioni, che include il monitoraggio, la pulizia e la gestione degli errori, consulta Utilizzare sessioni isolate per gli agenti.

Utilizzo di sessioni con connessioni WebSocket

Per utilizzare le sessioni con WebSocket connessioni, genera un ID di sessione univoco per ogni utente o conversazione e passalo quando stabilisci la connessione:

Utilizzando lo stesso ID di sessione per WebSocket le connessioni correlate, ti assicuri che il contesto venga mantenuto nella stessa conversazione, consentendo al tuo agente di fornire risposte coerenti basate sulle interazioni precedenti.

Ciclo di vita della sessione con connessioni WebSocket

Per WebSocket le connessioni, il timeout di inattività della sessione viene reimpostato ogni volta che si verifica un'attività di messaggio tra il client e l'agente. Ciò include qualsiasi scambio di WebSocket messaggi, ad esempio l'invio di dati da client a agent, la ricezione di risposte da agente a client o WebSocket ping/pong frame. Ciò significa che WebSocket le conversazioni attive manterranno attiva la sessione finché i messaggi continuano a fluire, impedendo la chiusura prematura della sessione durante le interazioni in corso.

Per ulteriori informazioni sulla configurazione delle impostazioni del ciclo di vita, consulta Configurare le impostazioni del ciclo di vita di Amazon Bedrock. AgentCore Per un controllo più diretto del ciclo di vita della sessione tramite lo stato di integrità dell'agente, consulta Runtime session lifecycle management. https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-long-run.html#runtime-long-run-session-lifecycle

Interrompi la sessione di runtime

Per interrompere una sessione in esecuzione prima di quella configurabile IdleRuntimeSessionTimeout (impostazione predefinita a 15 minuti), vedi Interrompere una sessione in esecuzione.

Osservabilità

Amazon Bedrock AgentCore Observability ti aiuta a tracciare, eseguire il debug e monitorare gli agenti ospitati in Amazon Bedrock Runtime. AgentCore Per prima cosa abilita CloudWatch Transaction Search seguendo le istruzioni in Abilitazione dell'osservabilità del runtime di Amazon Bedrock. AgentCore Per osservare il tuo agente, consulta Visualizzare i dati di osservabilità per i tuoi agenti Amazon Bedrock. AgentCore

Per WebSocket le connessioni, una traccia rappresenta l'intera sessione di connessione anziché i singoli scambi di messaggi.

Intestazioni personalizzate

Le intestazioni personalizzate consentono di trasmettere le informazioni contestuali dall'applicazione direttamente al codice dell'agente durante la connessione iniziale WebSocket . Per informazioni complete sul supporto, la configurazione e le limitazioni delle intestazioni personalizzate, consulta Passare le intestazioni personalizzate ad Amazon Bedrock Runtime. AgentCore

Inoltre, le intestazioni precedute da X-Amzn-Bedrock-AgentCore-Runtime-Custom- possono essere passate come parametri di query URL nelle connessioni. WebSocket

Ad esempio, puoi passare intestazioni personalizzate come parametri di query nell'URL: WebSocket

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

Il contenitore dell'applicazione agente le riceverà come intestazioni:

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

Appendice

Considerazioni relative alla sicurezza

Risoluzione dei problemi

Problemi comuni WebSocket-specific

Di seguito sono riportati i problemi più comuni che potresti riscontrare:

Gestione degli errori

WebSocket gli errori si manifestano in due fasi, a seconda di quando si verificano.

Stabilimento della connessione (prima dell' WebSocket aggiornamento)

L'apertura della connessione è una richiesta HTTP standard. Il codice di stato HTTP riflette l'eccezione e l'intestazione della x-amzn-ErrorType risposta riporta il nome dell'eccezione. Il servizio può restituire uno dei seguenti errori prima di stabilire la WebSocket connessione.

Connessione attiva (dopo l' WebSocket aggiornamento)

Una volta stabilito, gli errori vengono comunicati con codici di WebSocket chiusura standard anziché codici di stato HTTP. WebSocket I codici di chiusura più comuni includono:

WebSocket rispetto ad altri protocolli

Quando usare WebSocket:

Quando usare HTTP:

Esempi introduttivi aggiuntivi

Per ulteriori esempi di utilizzo dello streaming WebSocket bidirezionale con AgentCore Runtime, consulta gli esempi di streaming WebSocket bidirezionale: GitHub