nova-sonic-voice-agent

Build a Nova Sonic Voice Agent

This skill creates a real-time voice agent from scratch using Strands BidiAgent with Amazon Nova Sonic. The agent runs as a WebSocket server that streams bidirectional audio with the user's browser.

Architecture

Browser (mic + speaker)  ←WebSocket→  FastAPI Server  ←BidiStream→  Amazon Nova Sonic
                                          ↕
                                    Tools / Sub-Agents

Prerequisites

Python 3.11+
AWS credentials with access to Amazon Bedrock (Amazon Nova Sonic model)
A modern browser with microphone access

Install Dependencies

pip install strands-agents strands-agents-builder aws-sdk-bedrock-runtime
pip install fastapi uvicorn[standard] websockets

Environment Variables

Variable	Required	Description
`AWS_DEFAULT_REGION`	Yes	AWS region (default: `us-east-1`)
`AWS_ACCESS_KEY_ID`	Yes*	AWS access key (*auto-detected from profile)
`AWS_SECRET_ACCESS_KEY`	Yes*	AWS secret key (*auto-detected from profile)

Project Structure

project/
├── README.md
├── websocket/
│   ├── server.py           # FastAPI WebSocket server with event splitting
│   ├── agent.py            # BidiAgent session handler (config: voice, model, prompt)
│   ├── tools.py            # Custom @tool functions (optional)
│   ├── subagents.py        # Sub-agents as tools (optional)
│   └── requirements.txt
└── client/
    ├── client.py           # Python HTTP server serving the web page
    ├── index.html          # Browser UI with mic capture + audio playback
    └── requirements.txt

Part 1 — WebSocket Server (Orchestrator)

requirements.txt

Always include aws-sdk-bedrock-runtime — it provides the Nova Sonic bidirectional streaming client:

strands-agents
strands-agents-builder
aws-sdk-bedrock-runtime
fastapi
uvicorn[standard]
websockets

server.py — FastAPI Application

The server exposes a /ws WebSocket endpoint and splits large audio events at base64 boundaries:

import logging
import uvicorn
import os
import json
from fastapi import FastAPI, WebSocket
from fastapi.middleware.cors import CORSMiddleware
from agent import handle_websocket_session

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

MAX_WS_MESSAGE_SIZE = 10000

def split_large_event(event_dict, max_size=MAX_WS_MESSAGE_SIZE):
    """Split large audio events into smaller chunks at base64 boundaries."""
    event_json = json.dumps(event_dict)
    if len(event_json.encode("utf-8")) <= max_size:
        return [event_dict]
    if "audio" not in event_dict or not isinstance(event_dict["audio"], str):
        return [event_dict]

    audio_content = event_dict["audio"]
    template = {k: v for k, v in event_dict.items() if k != "audio"}
    template["audio"] = ""
    overhead = len(json.dumps(template).encode("utf-8"))
    max_content_size = ((max_size - overhead - 100) // 4) * 4

    chunks = []
    for i in range(0, len(audio_content), max_content_size):
        chunk_event = {k: v for k, v in event_dict.items() if k != "audio"}
        chunk_event["audio"] = audio_content[i:i + max_content_size]
        chunks.append(chunk_event)
    return chunks

app = FastAPI(title="Nova Sonic Voice Agent")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])

@app.get("/ping")
async def ping():
    import time
    return {"status": "Healthy", "time_of_last_update": int(time.time())}

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    async def chunked_send_json(event_dict):
        for chunk in split_large_event(event_dict):
            await websocket.send_json(chunk)
    await handle_websocket_session(websocket, send_output=chunked_send_json)

if __name__ == "__main__":
    uvicorn.run(app, host=os.getenv("HOST", "0.0.0.0"), port=int(os.getenv("PORT", "8081")))

agent.py — BidiAgent Session Handler

import logging
import traceback
from fastapi import WebSocket, WebSocketDisconnect
from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.models.nova_sonic import BidiNovaSonicModel

logger = logging.getLogger(__name__)

MODEL_ID = "amazon.nova-2-sonic-v1:0"
REGION = "us-east-1"
VOICE = "tiffany"
INPUT_RATE = 16000
OUTPUT_RATE = 16000
SYSTEM_PROMPT = """You are a friendly voice assistant. Be warm, conversational, and concise."""

async def handle_websocket_session(websocket: WebSocket, send_output=None):
    output_fn = send_output or websocket.send_json
    try:
        await _wait_for_config(websocket)
        agent = _create_agent()
        logger.info(f"✅ Agent ready: model={MODEL_ID}, voice={VOICE}")
        await output_fn({"type": "system", "message": f"Ready: {MODEL_ID} with voice={VOICE}"})

        async def handle_input():
            while True:
                message = await websocket.receive_json()
                if message.get("type") == "text_input":
                    await agent.send(message.get("text", ""))
                    continue
                return message

        await agent.run(inputs=[handle_input], outputs=[output_fn])
    except WebSocketDisconnect:
        logger.info("Client disconnected")
    except Exception as e:
        if "CANCELLED" not in str(e):
            logger.error(f"Error: {e}")
            traceback.print_exc()
    finally:
        logger.info("Session closed")

async def _wait_for_config(websocket: WebSocket):
    while True:
        message = await websocket.receive_json()
        if message.get("type") == "config":
            logger.info("📥 Client ready")
            return
        await websocket.send_json({"type": "system", "message": "Send config event first."})

def _create_agent() -> BidiAgent:
    model = BidiNovaSonicModel(
        region=REGION, model_id=MODEL_ID,
        provider_config={"audio": {"input_rate": INPUT_RATE, "output_rate": OUTPUT_RATE, "voice": VOICE}},
    )
    return BidiAgent(model=model, tools=[], system_prompt=SYSTEM_PROMPT)

Part 2 — Adding Tools

Create tools.py with @tool decorated functions:

from strands import tool

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city.

    Args:
        city: The city name to get weather for.
    """
    return f"The weather in {city} is sunny and 72°F."

Wire into agent.py:

from tools import get_weather

def _create_agent() -> BidiAgent:
    model = BidiNovaSonicModel(...)
    return BidiAgent(model=model, tools=[get_weather], system_prompt=SYSTEM_PROMPT)

Part 3 — Adding Sub-Agents

Sub-agents are full Strands Agent instances (text-based) passed as tools to the BidiAgent. Nova Sonic invokes them mid-stream like any other tool.

from strands import Agent, tool
from strands.models.bedrock import BedrockModel

@tool
def check_balance(account_id: str) -> str:
    """Check account balance."""
    return f"Account {account_id} balance: $4,250.00"

def create_finance_agent(region: str = "us-east-1"):
    model = BedrockModel(model_id="us.amazon.nova-2-lite-v1:0", region_name=region)
    return Agent(
        model=model,
        tools=[check_balance],
        system_prompt="You are a finance assistant. Be precise with numbers and concise.",
        name="finance_agent",
        description="Handles financial queries including account balances and transactions.",
    )