WebSocket を使用して双方向ストリーミングを開始する

Amazon Bedrock AgentCore ランタイムを使用すると、WebSocket ストリーミングをサポートするエージェントをデプロイして、リアルタイムの双方向通信を実現できます。このガイドでは、WebSocket を使用して最初の双方向ストリーミングエージェントを作成、テスト、デプロイする方法について説明します。

このセクションでは、以下を行います。

WebSocket プロトコルの詳細については、WebSocket RFC 6455」を参照してください。

AgentCore Runtime が WebSocket 接続をサポートする方法

AgentCore Runtime の WebSocket サポートにより、クライアントとエージェント間の永続的な双方向ストリーミング接続が可能になります。AgentCore Runtime は、コンテナが/wsパスのポート8080に WebSocket エンドポイントを実装することを想定しています。これは、標準の WebSocket サーバープラクティスと一致しています。

AgentCore Runtime の WebSocket サポートは、 InvokeAgentRuntime と同じサーバーレス、セッション分離、アイデンティティ、オブザーバビリティ機能を提供します。さらに、SigV4 または OAuth 2.0 認証を使用した WebSocket 接続を介したメッセージの低レイテンシーでリアルタイムの双方向ストリーミングが可能になり、リアルタイムの会話音声エージェントなどのアプリケーションに最適です。

サポートされている WebSocket ライブラリ

AgentCore Runtime での WebSockets を使用した双方向ストリーミングは、WebSocket 言語ライブラリを使用するアプリケーションをサポートします。唯一の要件は、クライアントが WebSocket プロトコル接続を使用してサービスエンドポイントに接続することです。

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

サポートされている認証方法 (SigV4 ヘッダー、SigV4 署名付き URL、または OAuth 2.0) のいずれかを使用し、エージェントアプリケーションが HTTP プロトコル契約で指定されている WebSocket サービス契約を実装していること。

この柔軟性により、さまざまなプログラミング言語やフレームワークで任意の WebSocket 実装を使用し、既存のコードベースや開発ワークフローとの互換性を確保できます。

AgentCore ランタイムでの WebSocket の使用

この入門チュートリアルでは、bedrock-agentcore Python SDK とデプロイ用の AgentCore CLI を使用して、双方向ストリーミングをサポートするエージェントアプリケーションを作成、テスト、デプロイします。

前提条件

開始する前に、以下を確認してください。

ステップ 1: プロジェクトをセットアップし、依存関係をインストールする

プロジェクトフォルダを作成し、必要なパッケージをインストールします。

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

pip を最新バージョンにアップグレードします。

pip install --upgrade pip

以下の必須パッケージをインストールします。

pip install bedrock-agentcore

ステップ 2: 双方向ストリーミングエージェントを作成する

という名前の双方向ストリーミングエージェントコードのソースファイルを作成しますwebsocket_echo_agent.py。次のコードを追加します。

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

以下を作成してrequirements.txt追加します。

bedrock-agentcore

Python websocketsライブラリの依存関係が含まれています

コードについて

ステップ 3: 双方向ストリーミングエージェントをローカルでテストする

双方向ストリーミングエージェントを起動する

ターミナルウィンドウを開き、次のコマンドを使用して双方向ストリーミングエージェントを起動します。

python websocket_echo_agent.py

サーバーがポート 8080 で実行されていることを示す出力が表示されます。

WebSocket 接続をテストする

という名前のローカル WebSocket 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())

別のターミナルウィンドウを開き、クライアントを実行して、双方向ストリーミングエージェントをローカルでテストします。

python websocket_agent_client.py

成功: のようなレスポンスが表示されますReceived: {"echo":{"inputText":"Hello WebSocket!"}}。エージェントを実行しているターミナルウィンドウで、 と入力Ctrl+Cしてエージェントを停止します。

ステップ 4: 双方向ストリーミングエージェントを AgentCore ランタイムにデプロイする

デプロイツールをインストールする

AgentCore CLI をインストールします。

npm install -g @aws/agentcore

インストールを確認します。

agentcore --help

プロジェクトを作成して にデプロイする AWS

双方向ストリーミングエージェントの新しいプロジェクトを作成します。

agentcore create

エージェントをデプロイします。

agentcore deploy

デプロイ後、エージェントランタイム ARN は次のようになります。

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

この ARN は、デプロイされたエージェントを呼び出すために必要なときに保存します。

ステップ 5: デプロイされた双方向ストリーミングエージェントを呼び出す

環境変数をセットアップする

必要な環境変数を設定します。

認証方法

InvokeAgentRuntimeWithWebSocketStream API アクションは、クライアントとエージェント間の双方向ストリーミングをサポートする WebSocket 接続を確立します。WebSocket 接続は、次の方法で認証できます。

SigV4 署名付きヘッダーを使用して接続する

次の例は、WebSocket 接続を確立し、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())

クライアントを実行して、デプロイされたエージェントをテストします。

python websocket_agent_client_sigv4_headers.py

成功: 次のようなレスポンスが表示されます。

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

署名付き URL を使用して接続する (クエリパラメータを介して SigV4)

次の例は、SigV4 クエリパラメータを使用して WebSocket URL を作成し、接続を確立する方法を示しています。

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

クライアントを実行して、デプロイされたエージェントをテストします。

python websocket_agent_client_sigv4_query_parameters.py

成功: 次のようなレスポンスが表示されます。

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

OAuth を使用して接続する

AgentCore ランタイムは、WebSocket 接続の OAuth ベアラートークン認証をサポートしています。OAuth 認証を使用するには、「Authenticate and authorize with Inbound Auth and Outbound Auth」の「JWT インバウンド認可と OAuth アウトバウンドアクセスサンプル」セクションで説明されているように、JWT 認可を使用してエージェントのランタイムを設定する必要があります。 インバウンド認証とアウトバウンド認証による認証と認可

OAuth セットアップを完了し、ステップ 4: ベアラートークンを使用して OAuth ガイドのエージェントを呼び出すに従ってベアラートークンを取得したら、そのトークンを使用して WebSocket 接続を確立できます。 OAuth

OAuth を使用する Python クライアント

次の例は、OAuth を使用して Python から WebSocket 接続を確立する方法を示しています。

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

クライアントを実行して、デプロイされたエージェントをテストします。

python websocket_agent_client_oauth.py

成功: 次のようなレスポンスが表示されます。

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

OAuth を使用したブラウザ JavaScript クライアント

ブラウザのネイティブ WebSocket API は、ハンドシェイク中にカスタムヘッダーを設定するメソッドを提供しません。ブラウザからの OAuth 認証をサポートするために、AgentCore Runtime は WebSocket ハンドシェイク中に Sec-WebSocket-Protocolヘッダーに埋め込まれたベアラートークンを受け入れます。

トークンは base64url でエンコードされ、プレフィックスに base64UrlBearerAuthorization. が付き、その後に sentinel サブプロトコル が続く必要がありますbase64UrlBearerAuthorization。

次の例は、OAuth を使用してブラウザ JavaScript から WebSocket 接続を確立する方法を示しています。

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

セッション管理

WebSocket 接続に session_id ( X-Amzn-Bedrock-AgentCore-Runtime-Session-Id ) を指定すると (URL クエリパラメータまたはリクエストヘッダーとして）、接続が分離されたランタイムセッションにルーティングされます。エージェントは、そのセッションに保存されている会話コンテキストにアクセスして、以前のやり取りを参照することで会話の継続性を実装できます。セッション IDs、分離された個別のコンテキストにアクセスし、ユーザーまたは会話が完全に分離されます。

追跡、クリーンアップ、エラー処理などの包括的なセッションライフサイクル管理については、「エージェントの分離セッションを使用する」を参照してください。

WebSocket 接続でのセッションの使用

WebSocket 接続でセッションを使用するには、ユーザーまたは会話ごとに一意のセッション ID を生成し、接続を確立するときに渡します。

関連する WebSocket 接続に同じセッション ID を使用することで、コンテキストが同じ会話全体で維持され、エージェントが以前のやり取りに基づいて一貫したレスポンスを提供できるようになります。

WebSocket 接続によるセッションライフサイクル

WebSocket 接続の場合、クライアントとエージェントの間にメッセージアクティビティがあるたびに、セッションのアイドルタイムアウトがリセットされます。これには、クライアントからエージェントへのデータの送信、エージェントからクライアントへのレスポンスの受信、WebSocket ping/pong フレームなどの WebSocket メッセージ交換が含まれます。つまり、アクティブな WebSocket 会話は、メッセージが流れ続ける限りセッションを存続させ、進行中のインタラクション中のセッションの早期終了を防ぎます。

ライフサイクル設定の詳細については、「Amazon Bedrock AgentCore ライフサイクル設定の構成」を参照してください。エージェントのヘルスステータスによるセッションライフサイクルのより直接的な制御については、「ランタイムセッションライフサイクル管理」を参照してください。

ランタイムセッションを停止する

設定可能 IdleRuntimeSessionTimeout (デフォルトは 15 分) の前に実行中のセッションを停止するには、「実行中のセッションを停止する」を参照してください。

オブザーバビリティ

Amazon Bedrock AgentCore オブザーバビリティは、Amazon Bedrock AgentCore ランタイムでホストするエージェントをトレース、デバッグ、モニタリングするのに役立ちます。まず、「Amazon Bedrock AgentCore ランタイムオブザーバビリティを有効にする」の手順に従って CloudWatch トランザクション検索を有効にします。エージェントを監視するには、「Amazon Bedrock AgentCore エージェントのオブザーバビリティデータを表示する」を参照してください。

WebSocket 接続の場合、トレースは個々のメッセージ交換ではなく、完全な接続セッションを表します。

カスタムヘッダー

カスタムヘッダーを使用すると、アプリケーションから最初の WebSocket 接続のエージェントコードにコンテキスト情報を直接渡すことができます。カスタムヘッダーのサポート、設定、制限の詳細については、「Amazon Bedrock AgentCore ランタイムにカスタムヘッダーを渡す」を参照してください。

さらに、 というプレフィックスが付いたヘッダーは、WebSocket 接続で URL クエリパラメータとして渡すX-Amzn-Bedrock-AgentCore-Runtime-Custom-ことができます。

たとえば、WebSocket URL でクエリパラメータとしてカスタムヘッダーを渡すことができます。

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

エージェントアプリケーションコンテナは、これらをヘッダーとして受け取ります。

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

付録

セキュリティに関する考慮事項

トラブルシューティング

WebSocket 固有の一般的な問題

以下は、発生する可能性のある一般的な問題です。

エラー処理

WebSocket 接続では、エラー通信に標準のクローズコードを使用します。一般的なクローズコードは次のとおりです。

WebSocket と他のプロトコル

WebSocket を使用するタイミング:

HTTP を使用するタイミング:

その他の使用開始の例

AgentCore Runtime で WebSocket 双方向ストリーミングを使用するその他の例については、WebSocket 双方向ストリーミング GitHub サンプル を参照してください。