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 WebSocket streaming agent.

In this section, you learn:

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

How AgentCore Runtime supports WebSocket

AgentCore Runtime's WebSocket support enables persistent, bidirectional streaming connections between clients and agents by acting as a transparent proxy layer. When configured for WebSocket, AgentCore Runtime expects containers to implement WebSocket endpoints on port 8080 at the /ws path, which aligns with standard WebSocket server practices.

The service provides enterprise-grade session isolation while maintaining protocol transparency - WebSocket upgrade requests and messages are routed directly to your agent container without modification. This architecture preserves standard WebSocket features like real-time bidirectional communication, while adding enterprise authentication (SigV4/OAuth 2.0) and scalability.

Key WebSocket characteristics:

Using WebSocket with AgentCore Runtime

In this tutorial you create, test, and deploy a WebSocket streaming agent.

Prerequisites

Step 1: Create your WebSocket agent

Create your first WebSocket agent

Create a new file called 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:
        while True:
            data = await websocket.receive_json()

            # Echo back
            await websocket.send_json({"echo": data, "session": context.session_id})

            # Exit on close command
            if data.get("action") == "close":
                break
    except Exception as e:
        print(f"Error: {e}")
    finally:
        await websocket.close()

if __name__ == "__main__":
    app.run()

Understanding the code

Step 2: Test your WebSocket agent locally

Start your WebSocket agent

Run your WebSocket agent locally:

python websocket_echo_agent.py

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

Test WebSocket connection

You can test the WebSocket endpoint using a simple Python client:

import asyncio
import websockets

async def test_websocket():
    uri = "ws://localhost:8080/ws"
    
    async with websockets.connect(uri) as websocket:
        # Send a test message
        await websocket.send("Hello WebSocket!")
        
        # Receive the echo response
        response = await websocket.recv()
        print(f"Received: {response}")

if __name__ == "__main__":
    asyncio.run(test_websocket())

Step 3: Deploy your WebSocket agent to AgentCore Runtime

Install deployment tools

Install the Amazon Bedrock AgentCore starter toolkit:

pip install bedrock-agentcore-starter-toolkit

Create a project folder with the following structure:

your_project_directory/
├── websocket_echo_agent.py  # Your main WebSocket agent code
├── requirements.txt         # Dependencies for your agent

Create a new file called requirements.txt:

bedrock-agentcore
websockets

Set up authentication (optional)

If you need OAuth authentication for your WebSocket agent, set up a Cognito user pool as described in Authenticate and authorize with Inbound Auth and Outbound Auth. This provides OAuth tokens required for secure access to your deployed WebSocket agent.

Configure and deploy to AWS

Configure your WebSocket 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

Step 4: Invoke your deployed WebSocket agent

Set up environment variables

Set up the required environment variables:

Authentication methods

The InvokeAgentRuntimeWithWebSocketStream operation establishes a WebSocket connection that supports bidirectional communication and maintains persistent session state throughout the connection lifecycle. 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 AgentRuntimeClient
import websockets
import asyncio
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 = AgentRuntimeClient(region="us-west-2")

    # Generate WebSocket connection with authentication
    ws_url, headers = client.generate_ws_connection(
        runtime_arn=runtime_arn
    )

    # Connect using any WebSocket library
    async with websockets.connect(ws_url, extra_headers=headers) as ws:
        # Send message
        await ws.send('{"inputText": "Hello!"}')

        # Receive response
        response = await ws.recv()
        print(f"Received: {response}")

if __name__ == "__main__":
    asyncio.run(main())

Connect using SigV4 presigned URL

The following example shows how to create a presigned WebSocket URL and establish a connection:

from bedrock_agentcore.runtime import AgentRuntimeClient
import os

# Get runtime ARN from environment variable
runtime_arn = os.getenv('AGENT_ARN')
if not runtime_arn:
    raise ValueError("AGENT_ARN environment variable is required")

# Backend: Generate presigned URL
client = AgentRuntimeClient(region="us-west-2")

presigned_url = client.generate_presigned_url(
    runtime_arn=runtime_arn,
    expires=300  # 5 minutes
)

# Share presigned_url with frontend
# Frontend JavaScript: new WebSocket(presigned_url)

Connect using OAuth

The following example shows how to establish a WebSocket connection using OAuth Bearer token authentication:

from bedrock_agentcore.runtime import AgentRuntimeClient
import websockets
import asyncio
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 = AgentRuntimeClient(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
    )

    # Connect using OAuth authentication
    async with websockets.connect(ws_url, extra_headers=headers) as ws:
        # Send message
        await ws.send('{"inputText": "Hello!"}')

        # Receive response
        response = await ws.recv()
        print(f"Received: {response}")

if __name__ == "__main__":
    asyncio.run(main())

Session management

The InvokeAgentRuntimeWithWebSocketStream operation maintains session state throughout the WebSocket connection lifetime using the runtimeSessionId parameter. By providing the same session identifier when establishing the WebSocket connection, you can maintain conversation context, allowing the agent to reference previous interactions throughout the connection duration.

To start a new conversation, generate a unique session identifier when establishing the WebSocket connection. To continue an existing conversation, use the same session identifier. The session remains active for the entire duration of the WebSocket connection.

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:

AgentCore Runtime automatically handles authentication and authorization errors, translating them to appropriate WebSocket close codes while preserving error information for debugging.

WebSocket vs other protocols

When to use WebSocket:

When to use other protocols: