Commencez à utiliser le streaming bidirectionnel à l'aide de 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 les WebSocket connexions

AgentCore La WebSocket prise en charge de Runtime permet des connexions de streaming bidirectionnelles persistantes 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 situé sur le /ws chemin, ce qui est conforme 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 un 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 des applications telles que les agents vocaux conversationnels en temps réel.

WebSocket Bibliothèques compatibles

Le streaming bidirectionnel à l'aide AgentCore d' WebSockets on Runtime prend en charge les applications utilisant n'importe quelle bibliothèque de WebSocket langue. Les seules exigences sont que les clients doivent se connecter au point de terminaison du service via une connexion WebSocket protocolaire :

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 agent implémente le contrat de WebSocket service tel que spécifié dans le contrat de protocole HTTP. 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 à niveau pip vers la dernière version :

pip install --upgrade pip

Installez les packages requis suivants :

pip install bedrock-agentcore

Étape 2 : Créez 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 de websockets la bibliothèque Python est incluse

Comprendre le code

Étape 3 : Testez localement votre agent de streaming bidirectionnel

Démarrez votre agent de streaming bidirectionnel

Ouvrez une fenêtre de terminal et démarrez 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.

WebSocket Connexion de test

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 localement votre agent de streaming bidirectionnel 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éployez 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 --version

Pour les commandes et options disponibles, consultez la référence AgentCore CLI.

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

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

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

Déployez votre agent :

agentcore deploy

Le AgentCore projet fait référence au répertoire agentcore-runtime-quickstart-websocket source existant.

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 d'InvokeAgentRuntimeWithWebSocketStreamAPI é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 moteur d'exécution d'un 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'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 Authentifier et autoriser avec l'authentification entrante et l'authentification sortante.

Une fois que vous avez terminé la configuration d'OAuth et obtenu un jeton porteur en suivant l'étape 4 : Utiliser 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 pendant 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é parbase64UrlBearerAuthorization., 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 d'assurer 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 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 les WebSocket connexions, le délai d'inactivité de la session est réinitialisé chaque fois qu'il y a une activité de message 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 au client ou WebSocket ping/pong des trames. Cela signifie que WebSocket les conversations actives maintiendront la session en vie tant que les messages continueront à circuler, évitant ainsi une interruption prématurée de la session pendant les interactions en cours.

Pour plus d'informations sur la configuration des paramètres de cycle de vie, voir Configurer les paramètres de AgentCore 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 la recherche de transactions en suivant les instructions de la section Activation de l'observabilité de l' AgentCore environnement d'exécution d'Amazon Bedrock. Pour observer votre agent, voir Afficher les données d'observabilité de vos agents Amazon Bedrock AgentCore .

Pour les WebSocket connexions, une trace représente la session de connexion complète plutôt que les é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 des informations complètes sur la prise en charge, la configuration et les limites des en-têtes personnalisés, voir 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 les recevra sous forme d'en-têtes :

"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 erreurs apparaissent en deux phases, selon le moment où elles se produisent.

Établissement de la connexion (avant la WebSocket mise à niveau)

L'ouverture de la connexion est une requête HTTP standard. Le code d'état HTTP reflète l'exception et l'en-tête de x-amzn-ErrorType réponse porte le nom de l'exception. Le service peut renvoyer l'une des erreurs suivantes avant d'établir la WebSocket connexion.

Connexion active (après la WebSocket mise à niveau)

Une fois que le WebSocket est établi, les erreurs sont communiquées à l'aide de codes de WebSocket fermeture standard plutôt que de codes d'état HTTP. Les codes de fermeture courants incluent :

WebSocket par rapport à d'autres protocoles

Quand utiliser WebSocket  :

Quand utiliser le protocole HTTP  :

Autres exemples de démarrage

Pour des exemples supplémentaires utilisant le streaming WebSocket bidirectionnel avec AgentCore Runtime, consultez les exemples de diffusion WebSocket GitHub bidirectionnelle :