Commencez avec le streaming bidirectionnel en utilisant WebSocket

Amazon Bedrock AgentCore Runtime vous permet de déployer des agents qui prennent WebSocket en charge le streaming pour une communication bidirectionnelle en temps réel. Ce guide vous explique comment créer, tester et déployer votre premier agent de streaming bidirectionnel à l'aide WebSocket de.

Dans cette section, vous allez apprendre :

Pour plus d'informations sur le WebSocket protocole, consultez la WebSocket RFC 6455.

Comment AgentCore Runtime prend en charge WebSocket les connexions

AgentCore Le WebSocket support de Runtime permet des connexions de streaming persistantes et bidirectionnelles entre les clients et les agents. AgentCore Runtime s'attend à ce que les conteneurs implémentent des WebSocket points de terminaison sur le port 8080 au niveau du /ws chemin, conformément aux pratiques standard des WebSocket serveurs.

AgentCore Le WebSocket support de Runtime fournit les mêmes fonctionnalités sans serveur, d'isolation de session, d'identité et d'observabilité que. InvokeAgentRuntime En outre, il permet le streaming bidirectionnel en temps réel et à faible latence des messages via des WebSocket connexions utilisant l'authentification Sigv4 ou OAuth 2.0, ce qui le rend idéal pour les applications telles que les agents vocaux conversationnels en temps réel.

WebSocket Librairies prises en charge

Le streaming bidirectionnel utilisant WebSockets AgentCore Runtime prend en charge les applications utilisant n'importe quelle bibliothèque de WebSocket langues. Les seules exigences sont que les clients se connectent au point de terminaison du service via une connexion de WebSocket protocole :

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

en utilisant l'une des méthodes d'authentification prises en charge (en-têtes Sigv4, URL pré-signée Sigv4 ou OAuth 2.0) et que l'application de l'agent implémente le contrat de WebSocket service tel que spécifié dans le contrat de protocole HTTP.

Cette flexibilité vous permet d'utiliser votre WebSocket implémentation préférée dans différents langages de programmation et frameworks, garantissant ainsi la compatibilité avec les bases de code et les flux de développement existants.

Utilisation WebSocket avec AgentCore Runtime

Dans ce didacticiel de démarrage, vous allez créer, tester et déployer une application d'agent qui prend en charge le streaming bidirectionnel à l'aide du SDK Python bedrock-agentcore et de la CLI pour le déploiement. AgentCore

Conditions préalables

Avant de commencer, assurez-vous d'avoir :

Étape 1 : Configuration du projet et installation des dépendances

Créez un dossier de projet et installez les packages requis :

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

Mettez à jour pip vers la dernière version :

pip install --upgrade pip

Installez les packages requis suivants :

pip install bedrock-agentcore

Étape 2 : Création de votre agent de streaming bidirectionnel

Créez un fichier source pour le code de votre agent de streaming bidirectionnel nomméwebsocket_echo_agent.py. Ajoutez le code suivant :

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

Créez requirements.txt et ajoutez les éléments suivants :

bedrock-agentcore

La dépendance à websockets la bibliothèque Python est incluse

Comprendre le code

Étape 3 : testez votre agent de streaming bidirectionnel localement

Démarrez votre agent de streaming bidirectionnel

Ouvrez une fenêtre de terminal et lancez votre agent de streaming bidirectionnel à l'aide de la commande suivante :

python websocket_echo_agent.py

Vous devriez voir une sortie indiquant que le serveur fonctionne sur le port 8080.

Tester WebSocket la connexion

Créez un WebSocket client local nommé websocket_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())

Testez votre agent de streaming bidirectionnel localement en ouvrant une autre fenêtre de terminal et en exécutant le client :

python websocket_agent_client.py

Succès : vous devriez voir une réponse du typeReceived: {"echo":{"inputText":"Hello WebSocket!"}}. Dans la fenêtre du terminal qui exécute l'agent, entrez Ctrl+C pour arrêter l'agent.

Étape 4 : Déployer votre agent de streaming bidirectionnel sur AgentCore Runtime

Installation des outils de déploiement

Installez la AgentCore CLI :

npm install -g @aws/agentcore

Vérifiez l'installation :

agentcore --help

Créez un projet et déployez-le sur AWS

Créez un nouveau projet pour votre agent de streaming bidirectionnel :

agentcore create

Déployez votre agent :

agentcore deploy

Après le déploiement, vous recevrez un ARN d'exécution de l'agent qui ressemble à :

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

Enregistrez cet ARN car vous en aurez besoin pour appeler votre agent déployé.

Étape 5 : Invoquez votre agent de streaming bidirectionnel déployé

Configurer les variables d’environnement

Configurez les variables d'environnement requises :

Méthodes d’authentification

L'action InvokeAgentRuntimeWithWebSocketStream API établit une WebSocket connexion qui prend en charge le streaming bidirectionnel entre le client et l'agent. Vous pouvez authentifier les WebSocket connexions à l'aide des méthodes suivantes :

Connectez-vous à l'aide d'en-têtes signés SigV4

L'exemple suivant montre comment établir une WebSocket connexion et communiquer avec un environnement d'exécution d'agent à l'aide d'en-têtes signés 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())

Exécutez le client pour tester votre agent déployé :

python websocket_agent_client_sigv4_headers.py

Succès : vous devriez voir une réponse du type :

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

Connectez-vous à l'aide d'une URL pré-signée (SigV4 via les paramètres de requête)

L'exemple suivant montre comment créer une WebSocket URL avec des paramètres de requête Sigv4 et établir une connexion :

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

Exécutez le client pour tester votre agent déployé :

python websocket_agent_client_sigv4_query_parameters.py

Succès : vous devriez voir une réponse du type :

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

Connectez-vous à l'aide d'OAuth

AgentCore Runtime prend en charge l'authentification par jeton OAuth Bearer pour les connexions. WebSocket Pour utiliser l'authentification OAuth, vous devez configurer l'environnement d'exécution de votre agent avec l'autorisation JWT, comme décrit dans la section Exemple d'autorisation entrante JWT et d'accès sortant OAuth de la section Authentifier et autoriser avec authentification entrante et authentification sortante.

Une fois que vous avez terminé la configuration OAuth et obtenu un jeton porteur en suivant l'étape 4 : Utilisez le jeton porteur pour appeler votre agent dans le guide OAuth, vous pouvez utiliser ce jeton pour établir des connexions. WebSocket

Client Python avec OAuth

L'exemple suivant montre comment établir une WebSocket connexion depuis Python à l'aide d'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())

Exécutez le client pour tester votre agent déployé :

python websocket_agent_client_oauth.py

Succès : vous devriez voir une réponse du type :

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

JavaScript Client de navigateur avec OAuth

L' WebSocket API native du navigateur ne fournit pas de méthode pour définir des en-têtes personnalisés lors de la poignée de main. Pour prendre en charge l'authentification OAuth depuis les navigateurs, AgentCore Runtime accepte le jeton porteur intégré dans l'Sec-WebSocket-Protocolen-tête lors de la poignée de main. WebSocket

Le jeton doit être codé en base64url et préfixé par le sous-protocole Sentinelbase64UrlBearerAuthorization., suivi du sous-protocole Sentinel. base64UrlBearerAuthorization

L'exemple suivant montre comment établir une WebSocket connexion depuis un navigateur à JavaScript l'aide d'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>

Gestion de session

Le fait de fournir un session_id (X-Amzn-Bedrock-AgentCore-Runtime-Session-Id) sur la WebSocket connexion (sous forme de paramètre de requête URL ou d'en-tête de demande) achemine la connexion vers une session d'exécution isolée. L'agent peut accéder au contexte de conversation stocké dans cette session, afin de garantir la continuité d'une conversation en faisant référence aux interactions précédentes. Les différents identifiants de session accèdent à des contextes isolés distincts, garantissant ainsi une isolation complète entre les utilisateurs ou les conversations.

Pour une gestion complète du cycle de vie des sessions, y compris le suivi, le nettoyage et la gestion des erreurs, voir Utiliser des sessions isolées pour les agents.

Utilisation de sessions avec WebSocket connexions

Pour utiliser des sessions avec des WebSocket connexions, générez un identifiant de session unique pour chaque utilisateur ou conversation et transmettez-le lors de l'établissement de la connexion :

En utilisant le même identifiant de session pour les WebSocket connexions associées, vous vous assurez que le contexte est maintenu tout au long de la même conversation, ce qui permet à votre agent de fournir des réponses cohérentes qui s'appuient sur les interactions précédentes.

Cycle de vie des sessions avec WebSocket connexions

Pour WebSocket les connexions, le délai d'inactivité de la session est réinitialisé chaque fois qu'un message est actif entre le client et l'agent. Cela inclut tout échange de WebSocket messages tel que l'envoi de données d'un client à un agent, la réception de réponses d'un agent à un autre ou WebSocket ping/pong des cadres. Cela signifie que les WebSocket conversations actives maintiendront la session active tant que les messages continueront à circuler, empêchant ainsi la fin prématurée de la session pendant les interactions en cours.

Pour plus d'informations sur la configuration des paramètres du cycle de vie, consultez Configurer les paramètres AgentCore du cycle de vie d'Amazon Bedrock. Pour un contrôle plus direct du cycle de vie des sessions via l'état de santé des agents, consultez la section Gestion du cycle de vie des sessions d'exécution.

Arrêter la session d'exécution

Pour arrêter une session en cours avant la configuration IdleRuntimeSessionTimeout (15 minutes par défaut), voir Arrêter une session en cours.

Observabilité

Amazon Bedrock AgentCore Observability vous permet de suivre, de déboguer et de surveiller les agents que vous hébergez dans Amazon Bedrock Runtime. AgentCore Activez d'abord CloudWatch Transaction Search en suivant les instructions de la section Activer l'observabilité du AgentCore runtime Amazon Bedrock. Pour observer votre agent, consultez Afficher les données d'observabilité de vos agents Amazon Bedrock AgentCore .

Pour WebSocket les connexions, une trace représente la session de connexion complète plutôt que des échanges de messages individuels.

En-têtes personnalisés

Les en-têtes personnalisés vous permettent de transmettre les informations contextuelles de votre application directement au code de votre agent lors de la connexion initiale WebSocket . Pour obtenir des informations complètes sur la prise en charge, la configuration et les limites des en-têtes personnalisés, consultez Transmettre des en-têtes personnalisés à Amazon Bedrock Runtime AgentCore .

En outre, les en-têtes préfixés par X-Amzn-Bedrock-AgentCore-Runtime-Custom- peuvent être transmis en tant que paramètres de requête d'URL dans WebSocket les connexions.

Par exemple, vous pouvez transmettre des en-têtes personnalisés en tant que paramètres de requête dans l' WebSocket URL :

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

Le conteneur de l'application de l'agent recevra les en-têtes suivants :

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

Annexe

Considérations sur la sécurité

Résolution des problèmes

WebSocket-specific Problèmes courants

Les problèmes courants que vous pouvez rencontrer sont les suivants :

Gestion des erreurs

WebSocket les connexions utilisent des codes de fermeture standard pour la communication d'erreur. Les codes de fermeture courants incluent :

WebSocket par rapport aux autres protocoles

Quand utiliser WebSocket :

Quand utiliser le protocole HTTP :

Autres exemples de mise en route

Pour d'autres exemples d'utilisation du streaming WebSocket bidirectionnel avec AgentCore Runtime, consultez les exemples de streaming WebSocket GitHub bidirectionnel :