Inizia con lo streaming bidirezionale utilizzando WebSocket

Amazon Bedrock AgentCore Runtime consente di distribuire agenti che supportano WebSocket lo streaming per comunicazioni bidirezionali in tempo reale. Questa guida ti 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 prevede che i container implementino gli WebSocket endpoint sulla porta /ws lungo il percorso, il che è 8080 in linea con le pratiche standard del server. WebSocket

AgentCore Il WebSocket supporto di Runtime fornisce le stesse funzionalità serverless, di isolamento delle sessioni, di identità e di 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 agenti vocali conversazionali in tempo reale.

Librerie supportate WebSocket

Lo streaming bidirezionale utilizzato WebSockets su AgentCore Runtime supporta le applicazioni che utilizzano qualsiasi libreria di WebSocket lingue. 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

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:

Passaggio 1: configurare il progetto e installare le 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 con nome in codice. 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 della websockets libreria python è inclusa

Comprendere il codice

Fase 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

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

Passaggio 4: Implementa 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 --help

Crea progetto e distribuiscilo su AWS

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

agentcore create

Implementa il tuo agente:

agentcore deploy

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 l'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 il client e l'agente. È possibile autenticare le WebSocket connessioni utilizzando i seguenti metodi:

Connect utilizzando header firmati SigV4

L'esempio seguente mostra come stabilire una WebSocket connessione e comunicare con un agente runtime 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

Operazione riuscita: dovresti vedere una risposta del tipo:

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

Connect tramite URL prefirmato (SIGv4 tramite parametri di query)

L'esempio seguente mostra come creare un WebSocket URL con i 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

Operazione riuscita: dovresti vedere una risposta del tipo:

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

Connect usando OAuth

AgentCore Runtime supporta l'autenticazione tramite 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 in uscita OAuth di Authenticate e autorizzate con autenticazione in entrata e autenticazione in uscita.

Dopo aver completato la configurazione OAuth e ottenuto un token al portatore seguendo il passaggio 4: Usa il token bearer per richiamare 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

Operazione riuscita: dovresti vedere una risposta del tipo:

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

JavaScript Client per 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 bearer incorporato nell'intestazione durante l'Sec-WebSocket-Protocolhandshake. WebSocket

Il token deve avere la codifica base64url e il prefisso, seguito 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

L'immissione di un session_id (X-Amzn-Bedrock-AgentCore-Runtime-Session-Id) sulla WebSocket connessione (come parametro di query URL o intestazione della richiesta) indirizza la connessione 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 l'isolamento completo tra utenti o conversazioni.

Per una gestione completa del ciclo di vita delle sessioni, tra cui tracciamento, pulizia e 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, garantisci che il contesto venga mantenuto durante la stessa conversazione, permettendo all'agente di fornire risposte coerenti basate su interazioni precedenti.

Ciclo di vita della sessione con connessioni WebSocket

Per le WebSocket connessioni, il timeout di inattività della sessione viene reimpostato ogni volta che si verifica un'attività di messaggistica tra il client e l'agente. Ciò include qualsiasi scambio di WebSocket messaggi, ad esempio l'invio di dati dal client all'agente, la ricezione di risposte da un agente all'altro o WebSocket ping/pong i frame. Ciò significa che WebSocket le conversazioni attive manterranno attiva la sessione finché i messaggi continueranno a fluire, evitando l'interruzione 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 AgentCore Bedrock. Per un controllo più diretto del ciclo di vita della sessione attraverso lo stato di salute dell'agente, consulta Gestione del ciclo di vita delle sessioni di Runtime.

Interrompi la sessione di runtime

Per interrompere una sessione in esecuzione prima di quella configurabile IdleRuntimeSessionTimeout (impostazione predefinita: 15 minuti), vedi Arrestare 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 Enabling Amazon Bedrock AgentCore runtime observability. 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 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 intestazioni personalizzate ad Amazon AgentCore Bedrock Runtime.

Inoltre, le intestazioni con il prefisso 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 dell'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 le connessioni utilizzano codici di chiusura standard per la comunicazione degli errori. 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