Get started with WebSocket streaming in AgentCore Runtime

Amazon Bedrock AgentCore Runtime lets you deploy agents that support WebSocket streaming for real-time bidirectional communication. This guide walks you through creating, testing, and deploying your first bidirectional streaming agent using WebSocket.

In this section, you learn:

For more information about the WebSocket protocol, see WebSocket RFC 6455.

How AgentCore Runtime supports WebSocket connections

AgentCore Runtime's WebSocket support enables persistent, bidirectional streaming connections between clients and agents. AgentCore Runtime expects containers to implement WebSocket endpoints on port 8080 at the /ws path, which aligns with standard WebSocket server practices.

AgentCore Runtime's WebSocket support provides the same serverless, session isolation, identity, and observability capabilities as InvokeAgentRuntime. Additionally, it enables low-latency, real-time bidirectional streaming of messages through WebSocket connections using SigV4 or OAuth 2.0 authentication, making it ideal for applications such as real-time conversational voice agents.

Supported WebSocket libraries

Bidirectional streaming using WebSockets on AgentCore Runtime supports applications using any WebSocket language library. The only requirements are that clients connect to the service endpoint with a WebSocket protocol connection:

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

using one of the supported authentication methods (SigV4 headers, SigV4 pre-signed URL, or OAuth 2.0) and that the agent application implements the WebSocket service contract as specified in HTTP protocol contract.

This flexibility allows you to use your preferred WebSocket implementation across different programming languages and frameworks, ensuring compatibility with existing codebases and development workflows.

Using WebSocket with AgentCore Runtime

In this getting started tutorial you will create, test, and deploy an agent application that supports bidirectional streaming using the bedrock-agentcore Python SDK and the Amazon Bedrock AgentCore starter toolkit for deployment.

Prerequisites

Before you start, make sure you have:

Step 1: Set up project and install dependencies

Create a project folder and install the required packages:

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

Upgrade pip to the latest version:

pip install --upgrade pip

Install the following required packages:

pip install bedrock-agentcore

Step 2: Create your bidirectional streaming agent

Create a source file for your bidirectional streaming agent code named websocket_echo_agent.py. Add the following code:

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

Create requirements.txt and add the following:

bedrock-agentcore

The python websockets library dependency is included

Understanding the code

Step 3: Test your bidirectional streaming agent locally

Start your bidirectional streaming agent

Open a terminal window and start your bidirectional streaming agent with the following command:

python websocket_echo_agent.py

You should see output indicating the server is running on port 8080.

Test WebSocket connection

Create a local WebSocket client named 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())

Test your bidirectional streaming agent locally by opening another terminal window and running the client:

python websocket_agent_client.py

Success: You should see a response like Received: {"echo":{"inputText":"Hello WebSocket!"}}. In the terminal window that's running the agent, enter Ctrl+C to stop the agent.

Step 4: Deploy your bidirectional streaming agent to AgentCore Runtime

Install deployment tools

Install the Amazon Bedrock AgentCore starter toolkit:

pip install bedrock-agentcore-starter-toolkit

Verify installation:

agentcore --help

Configure and deploy to AWS

Configure your bidirectional streaming agent for deployment:

agentcore configure -e websocket_echo_agent.py

Deploy your agent:

agentcore launch

After deployment, you'll receive an agent runtime ARN that looks like:

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

Save this ARN as you'll need it to invoke your deployed agent.

Step 5: Invoke your deployed bidirectional streaming agent

Set up environment variables

Set up the required environment variables:

Authentication methods

The InvokeAgentRuntimeWithWebSocketStream API action establishes a WebSocket connection that supports bidirectional streaming between the client and agent. You can authenticate WebSocket connections using the following methods:

Connect using SigV4 signed headers

The following example shows how to establish a WebSocket connection and communicate with an agent runtime using SigV4 signed headers:

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

Run the client to test your deployed agent:

python websocket_agent_client_sigv4_headers.py

Success: You should see a response like:

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

Connect using pre-signed URL (SigV4 via query parameters)

The following example shows how to create a WebSocket URL with SigV4 query parameters and establish a connection:

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

Run the client to test your deployed agent:

python websocket_agent_client_sigv4_query_parameters.py

Success: You should see a response like:

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

Connect using OAuth

AgentCore Runtime supports OAuth Bearer token authentication for WebSocket connections. To use OAuth authentication, you need to configure your agent runtime with JWT authorization as described in the JWT inbound authorization and OAuth outbound access sample section of Authenticate and authorize with Inbound Auth and Outbound Auth.

Once you have completed the OAuth setup and obtained a bearer token following Step 4: Use bearer token to invoke your agent in the OAuth guide, you can use that token to establish WebSocket connections as shown in the following example:

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

Run the client to test your deployed agent:

python websocket_agent_client_oauth.py

Success: You should see a response like:

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

Session management

Providing a session_id (X-Amzn-Bedrock-AgentCore-Runtime-Session-Id) on the WebSocket connection (as either a URL query parameter or request header) routes the connection to an isolated runtime session. The agent can access conversation context stored within that session, to implement continuity for a conversation by referencing previous interactions. Different session IDs access separate isolated contexts, ensuring complete isolation between users or conversations.

For comprehensive session lifecycle management including tracking, cleanup, and error handling, see Use isolated sessions for agents.

Using sessions with WebSocket connections

To use sessions with WebSocket connections, generate a unique session ID for each user or conversation and pass it when establishing the connection:

By using the same session ID for related WebSocket connections, you ensure that context is maintained across the same conversation, allowing your agent to provide coherent responses that build on previous interactions.

Appendix

Security considerations

Troubleshooting

Common WebSocket-specific issues

The following are common issues you might encounter:

Error handling

WebSocket connections use standard close codes for error communication. Common close codes include:

WebSocket vs other protocols

When to use WebSocket:

When to use HTTP:

Additional getting started examples

For additional examples using WebSocket bidirectional streaming with AgentCore Runtime, see the WebSocket bidirectional streaming GitHub samples: