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

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

Wire into agent.py:

from subagents import create_finance_agent

def _create_agent() -> BidiAgent:
    model = BidiNovaSonicModel(...)
    finance = create_finance_agent(region=REGION)
    return BidiAgent(model=model, tools=[finance], system_prompt=SYSTEM_PROMPT)

Part 4 — Browser Client

See references/client-reference.md for the full browser client implementation including:

• AudioWorklet microphone capture at 16kHz
• Gapless audio playback with scheduled AudioBufferSourceNodes
• Speculative/final transcript handling
• Barge-in support

Available Voices

Audio Configuration

• Sample rate: 16000 Hz (Nova Sonic native)
• Bit depth: 16-bit signed integer PCM
• Channels: 1 (mono)
• Encoding: Base64 over JSON WebSocket

Common Pitfalls

• Sample rate mismatch: Nova Sonic requires 16kHz. Wrong rate produces garbled audio.
• Long system prompts: Keep under ~500 words. Move details into tool descriptions.
• JSON in tool results: Return natural-language strings, not raw JSON — the response is spoken aloud.
• Sub-agent verbosity: Tell sub-agents to be brief since output is spoken.
• Missing portaudio: Install portaudio system library before pip install pyaudio (only needed for CLI clients).

Nova Sonic Voice Agent Skill

A coding skill for building real-time voice agents from scratch using Amazon Nova Sonic and Strands Agents BidiAgent. Works with Kiro, Claude Code, and other AI coding assistants.

What This Skill Does

Guides AI coding assistants to scaffold and build a complete voice agent project with:

• WebSocket server — FastAPI + BidiAgent orchestrator with Amazon Nova Sonic
• Custom tools — @tool decorated functions invoked mid-conversation
• Sub-agents — Full Strands Agent instances used as tools for complex reasoning
• Browser client — Web Audio API mic capture and audio playback

When To Use This Skill

• Building a voice agent from scratch
• Working with Amazon Nova Sonic or BidiAgent
• Adding speech-to-speech, real-time voice, or audio streaming to a project
• Adding tools or sub-agents to a voice agent

When NOT To Use This Skill

• Migrating an existing text agent → use text-agent-to-strands-voice-agent skill instead
• TTS/STT without a live agent loop
• Deployment or infrastructure only

Skill Contents

nova-sonic-voice-agent/
├── SKILL.md                        # Main skill instructions and code patterns
├── README.md                       # This file
└── references/
    ├── server-reference.md         # Full server implementation details
    ├── client-reference.md         # Full browser client implementation
    └── sub-agent-patterns.md       # Sub-agent design patterns and examples

---

Register and Use in Kiro

Kiro uses a .kiro/skills/ directory in your workspace to discover skills.

Step 1 — Copy the skill into your project

mkdir -p .kiro/skills
cp -r skills/nova-sonic-voice-agent .kiro/skills/nova-sonic-voice-agent

Step 2 — Use it

Open Kiro and start a conversation. When you mention building a voice agent, Nova Sonic, or BidiAgent, Kiro automatically loads the skill and follows its instructions.

Example prompts:

• "Build me a voice agent with Amazon Nova Sonic"
• "Create a real-time speech-to-speech agent with tools"
• "Add a finance sub-agent to my voice agent"

How it works in Kiro

Kiro reads the SKILL.md front-matter (name and description fields) to decide when to activate the skill. When a user prompt matches the trigger conditions in the description, Kiro loads SKILL.md into context and follows its instructions to generate code. The references/ folder provides additional detail that Kiro pulls in as needed.

---

Register and Use in Claude Code

Claude Code uses a CLAUDE.md file in your project root for persistent instructions.

Option A — Reference the skill file (recommended)

Create or edit CLAUDE.md in your project root and add:

## Voice Agent Skill

When I ask about building voice agents with Amazon Nova Sonic or Strands BidiAgent,
read and follow the instructions in these files:

- skills/nova-sonic-voice-agent/SKILL.md (main instructions)
- skills/nova-sonic-voice-agent/references/server-reference.md (server details)
- skills/nova-sonic-voice-agent/references/client-reference.md (browser client details)
- skills/nova-sonic-voice-agent/references/sub-agent-patterns.md (sub-agent patterns)

Option B — Inline the skill content

Open skills/nova-sonic-voice-agent/SKILL.md from the file explorer, copy its contents, and paste it into your CLAUDE.md file.

Step 2 — Use it

Start Claude Code and ask it to build a voice agent:

> Build a real-time voice agent with Amazon Nova Sonic that can check the weather
> Create a voice agent with a finance sub-agent
> Scaffold a BidiAgent project with a browser client

Claude Code reads CLAUDE.md on every interaction, sees the skill reference, and follows the patterns in SKILL.md when generating code.

How it works in Claude Code

Claude Code loads CLAUDE.md at the start of every session. When you reference skill files, Claude Code reads them on demand when the topic is relevant. The skill provides project structure, code patterns, and implementation details so Claude Code generates consistent, working voice agent code.

---

Related

• Power: powers/voice-agent-nova-sonic-strands/ — The Kiro power this skill is derived from
• Skill: skills/text-agent-to-strands-voice-agent/ — For migrating existing text agents to voice

Browser Client Reference

Complete implementation for the browser-based voice agent client.

Overview

The client captures microphone audio via Web Audio API, streams it to the server over WebSocket, and plays back audio responses. It consists of a Python HTTP server (client.py) that serves a single-page HTML application (index.html).

client.py

#!/usr/bin/env python3
import argparse
import os
import sys
import webbrowser
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse


class VoiceClientHandler(BaseHTTPRequestHandler):
    """HTTP handler that serves the voice agent client."""

    websocket_url = None

    def log_message(self, format, *args):
        sys.stderr.write(f"[{self.log_date_time_string()}] {format % args}\n")

    def do_GET(self):
        parsed_path = urlparse(self.path)
        if parsed_path.path in ("/", "/index.html"):
            self.serve_client_page()
        else:
            self.send_error(404, "Not found")

    def serve_client_page(self):
        try:
            html_path = os.path.join(os.path.dirname(__file__), "index.html")
            with open(html_path, "r", encoding="utf-8") as f:
                html_content = f.read()

            if self.websocket_url:
                html_content = html_content.replace("{{WEBSOCKET_URL}}", self.websocket_url)
            else:
                html_content = html_content.replace("{{WEBSOCKET_URL}}", "")

            content = html_content.encode()
            self.send_response(200)
            self.send_header("Content-type", "text/html")
            self.send_header("Content-Length", len(content))
            self.end_headers()
            self.wfile.write(content)
        except FileNotFoundError:
            self.send_error(404, "index.html not found")


def main():
    parser = argparse.ArgumentParser(description="Voice Agent Browser Client")
    parser.add_argument("--ws-url", default="ws://localhost:8081/ws", help="WebSocket server URL")
    parser.add_argument("--port", type=int, default=8000, help="HTTP server port (default: 8000)")
    parser.add_argument("--no-browser", action="store_true", help="Don't open browser automatically")

    args = parser.parse_args()
    VoiceClientHandler.websocket_url = args.ws_url

    print("=" * 60)
    print("🎙️  Nova Sonic Voice Agent Client")
    print("=" * 60)
    print(f"🔗 WebSocket: {args.ws_url}")
    print(f"🌐 Server:    http://localhost:{args.port}")
    print(f"💡 Press Ctrl+C to stop")
    print("=" * 60)

    if not args.no_browser:
        webbrowser.open(f"http://localhost:{args.port}")

    httpd = HTTPServer(("", args.port), VoiceClientHandler)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n👋 Shutting down...")


if __name__ == "__main__":
    main()

Audio Implementation (JavaScript)

Global State

let ws = null;
let audioContext = null;
let audioPlaybackContext = null;
let isRecording = false;
let isBotSpeaking = false;
let nextPlayTime = 0;
let activeSources = 0;
const SAMPLE_RATE = 16000;

Microphone Capture

Uses ScriptProcessorNode with 4096-sample buffer. Downsamples from browser native rate to 16kHz. Suppresses mic input while bot is speaking (factor 0.15) to reduce echo while allowing barge-in.

async function startRecording() {
    const stream = await navigator.mediaDevices.getUserMedia({
        audio: { channelCount: 1, echoCancellation: true, noiseSuppression: true }
    });

    audioContext = new AudioContext();
    const source = audioContext.createMediaStreamSource(stream);
    const processor = audioContext.createScriptProcessor(4096, 1, 1);

    processor.onaudioprocess = (e) => {
        if (!ws || ws.readyState !== WebSocket.OPEN) return;

        const inputData = e.inputBuffer.getChannelData(0);
        const suppressionFactor = isBotSpeaking ? 0.15 : 1.0;
        const downsampleRatio = audioContext.sampleRate / SAMPLE_RATE;
        const outputLength = Math.floor(inputData.length / downsampleRatio);
        const int16Data = new Int16Array(outputLength);

        for (let i = 0; i < outputLength; i++) {
            const sourceIndex = Math.floor(i * downsampleRatio);
            int16Data[i] = Math.max(-32768, Math.min(32767,
                inputData[sourceIndex] * 32768 * suppressionFactor));
        }

        const bytes = new Uint8Array(int16Data.buffer);
        let binary = '';
        for (let i = 0; i < bytes.length; i++) {
            binary += String.fromCharCode(bytes[i]);
        }

        ws.send(JSON.stringify({
            type: "bidi_audio_input",
            audio: btoa(binary),
            format: "pcm",
            sample_rate: SAMPLE_RATE,
            channels: 1
        }));
    };

    source.connect(processor);
    processor.connect(audioContext.destination);
    isRecording = true;
}

function stopRecording() {
    if (audioContext) {
        audioContext.close();
        audioContext = null;
    }
    isRecording = false;
}

Audio Playback

Scheduled AudioBufferSourceNode playback for gapless output. Tracks active sources to know when bot finishes speaking.

async function playAudioOutput(base64Audio) {
    if (!audioPlaybackContext) {
        audioPlaybackContext = new AudioContext({ sampleRate: SAMPLE_RATE });
        nextPlayTime = 0;
        activeSources = 0;
    }

    if (audioPlaybackContext.state === 'suspended') {
        await audioPlaybackContext.resume();
    }

    isBotSpeaking = true;

    // Decode base64 → Int16 → Float32
    const binaryString = atob(base64Audio);
    const bytes = new Uint8Array(binaryString.length);
    for (let i = 0; i < binaryString.length; i++) {
        bytes[i] = binaryString.charCodeAt(i);
    }
    const int16Data = new Int16Array(bytes.buffer);
    const float32Data = new Float32Array(int16Data.length);
    for (let i = 0; i < int16Data.length; i++) {
        float32Data[i] = int16Data[i] / 32768.0;
    }

    const buffer = audioPlaybackContext.createBuffer(1, float32Data.length, SAMPLE_RATE);
    buffer.getChannelData(0).set(float32Data);

    const currentTime = audioPlaybackContext.currentTime;
    if (nextPlayTime < currentTime) {
        nextPlayTime = currentTime + 0.05;
    }

    const source = audioPlaybackContext.createBufferSource();
    source.buffer = buffer;
    source.connect(audioPlaybackContext.destination);
    source.start(nextPlayTime);
    nextPlayTime += buffer.duration;

    activeSources++;
    source.onended = () => {
        activeSources--;
        if (activeSources <= 0) {
            activeSources = 0;
            isBotSpeaking = false;
        }
    };
}

function stopBotAudio() {
    isBotSpeaking = false;
    nextPlayTime = 0;
    activeSources = 0;
    if (audioPlaybackContext) {
        audioPlaybackContext.close();
        audioPlaybackContext = null;
    }
}

WebSocket Connection

async function connect() {
    const wsUrl = "{{WEBSOCKET_URL}}";
    ws = new WebSocket(wsUrl);

    ws.onopen = async () => {
        ws.send(JSON.stringify({ type: "config" }));
        await startRecording();
    };

    ws.onmessage = async (event) => {
        const data = JSON.parse(event.data);

        switch (data.type) {
            case 'bidi_audio_stream':
                await playAudioOutput(data.audio);
                break;
            case 'bidi_transcript_stream':
                if (data.is_final === false) {
                    updateSpeculativeTranscript(data.role, data.text);
                } else {
                    finalizeTranscript(data.role, data.text);
                }
                break;
            case 'bidi_interruption':
                stopBotAudio();
                break;
            case 'tool_use_stream':
                showToolUse(data.current_tool_use.name);
                break;
            case 'tool_result':
                showToolResult(data.tool_result);
                break;
            case 'system':
                showSystemMessage(data.message);
                break;
            case 'error':
                showError(data.message);
                break;
        }
    };

    ws.onclose = () => {
        stopRecording();
        stopBotAudio();
    };
}

Transcript Display

Nova Sonic sends speculative transcripts (is_final: false) that get replaced by final ones:

function updateSpeculativeTranscript(role, text) {
    const prefix = role === 'user' ? '🎤' : '🔊';
    let el = document.getElementById('speculative-msg');
    if (el) {
        el.textContent = `${prefix} ${text}`;
    } else {
        el = addMessage(`${prefix} ${text}`, role);
        el.id = 'speculative-msg';
        el.style.opacity = '0.6';
    }
}

function finalizeTranscript(role, text) {
    const prefix = role === 'user' ? '🎤' : '🔊';
    const speculative = document.getElementById('speculative-msg');
    if (speculative) speculative.remove();
    addMessage(`${prefix} ${text}`, role);
}

Text Input

function sendTextMessage(text) {
    if (!text || !ws || ws.readyState !== WebSocket.OPEN) return;
    ws.send(JSON.stringify({ type: "text_input", text: text }));
}

Audio Format Details

requirements.txt

# No external dependencies — uses Python standard library only

Running

cd client
python client.py --ws-url ws://localhost:8081/ws

Server Reference

Complete implementation details for the WebSocket server orchestrator.

server.py — Full Implementation

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 a large event into smaller chunks by dividing the audio field.

    Ensures splits occur at base64 boundaries (4-char alignment) to avoid corruption.
    Returns a list of event dicts to send.
    """
    event_json = json.dumps(event_dict)
    event_size = len(event_json.encode("utf-8"))

    if event_size <= 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
    max_content_size = (max_content_size // 4) * 4  # Align to base64

    if max_content_size <= 0:
        return [event_dict]

    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)

    logger.info(f"Split audio event ({event_size} bytes) into {len(chunks)} chunks")
    return chunks


app = FastAPI(title="Nova Sonic Voice Agent")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.on_event("startup")
async def startup_event():
    logger.info("🚀 Starting Nova Sonic Voice Agent server...")
    logger.info(f"📍 Region: {os.getenv('AWS_DEFAULT_REGION', 'us-east-1')}")


@app.get("/ping")
async def ping():
    """Health check endpoint."""
    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):
        """Send output events, splitting large audio payloads."""
        chunks = split_large_event(event_dict)
        for chunk in chunks:
            await websocket.send_json(chunk)

    await handle_websocket_session(websocket, send_output=chunked_send_json)


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

agent.py — Full Implementation

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

# ---------------------------------------------------------------------------
# Agent configuration — edit these to change behavior
# ---------------------------------------------------------------------------
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):
    """Handle a single WebSocket voice session."""
    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":
                    text = message.get("text", "")
                    logger.info(f"Text input: {text}")
                    await agent.send(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" in str(e):
            logger.warning(f"Cleanup error (ignored): {e}")
        else:
            logger.error(f"Error: {e}")
            traceback.print_exc()
            try:
                await output_fn({"type": "error", "message": str(e)})
            except Exception:
                pass
    finally:
        logger.info("Session closed")


async def _wait_for_config(websocket: WebSocket):
    """Wait for the client to send a config event (readiness signal)."""
    while True:
        message = await websocket.receive_json()
        if message.get("type") == "config":
            logger.info("📥 Client ready (config event received)")
            return
        else:
            await websocket.send_json({
                "type": "system",
                "message": "Please send a config event first.",
            })


def _create_agent() -> BidiAgent:
    """Create a BidiAgent with hardcoded configuration."""
    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,
    )

tools.py — Custom Tools Pattern

from strands import tool
import json


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

    Args:
        city: The city name to get weather for.
    """
    # Replace with real API call
    return f"The weather in {city} is sunny and 72°F."


@tool
def get_time() -> str:
    """Get the current time."""
    from datetime import datetime
    return f"The current time is {datetime.now().strftime('%I:%M %p')}."

When tools are added, update _create_agent() in agent.py:

from tools import get_weather, get_time

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

requirements.txt

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

WebSocket Event Protocol

Client → Server

Server → Client

Key Behaviors

• One agent per connection: Each WebSocket gets its own BidiAgent instance
• Config-first protocol: Client must send {"type": "config"} before audio flows
• Barge-in: Nova Sonic handles VAD natively — no external silence detection needed
• Tool use mid-stream: Tools execute without pausing the audio flow
• Large event splitting: Audio events >10KB are split at base64 boundaries

Sub-Agent Patterns

Design patterns and examples for using Strands Agent instances as tools within the BidiAgent voice agent.

Concept

A sub-agent is a full Strands Agent instance (text-based, using a reasoning model) that gets passed as a tool to the BidiAgent. Nova Sonic invokes it mid-stream like any other tool — the sub-agent processes the request, returns text, and Nova Sonic speaks it back to the user.

Default Configuration

• Reasoning model: us.amazon.nova-2-lite-v1:0 (fast, cost-effective)
• Use Nova Pro (us.amazon.nova-2-pro-v1:0) only when stronger reasoning is needed
• Keep responses concise — output is spoken aloud

Basic Pattern

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


@tool
def lookup_order(order_id: str) -> str:
    """Look up an order by ID.

    Args:
        order_id: The order ID to look up.
    """
    # Replace with real API call
    return f"Order {order_id}: shipped on May 20, arriving May 23."


def create_order_agent(region: str = "us-east-1"):
    """Create an order tracking sub-agent."""
    model = BedrockModel(
        model_id="us.amazon.nova-2-lite-v1:0",
        region_name=region,
    )

    return Agent(
        model=model,
        tools=[lookup_order],
        system_prompt="You are an order tracking specialist. Look up orders and provide brief status updates.",
        name="order_tracker",
        description="Tracks and provides status updates for customer orders. Call this when the user asks about an order, shipment, or delivery.",
    )

Factory Pattern (Reusable)

def create_subagent_tool(
    name: str,
    description: str,
    system_prompt: str,
    tools: list = None,
    model_id: str = "us.amazon.nova-2-lite-v1:0",
    region: str = "us-east-1",
):
    """Create a Strands Agent configured as a tool for the BidiAgent.

    Args:
        name: Tool name exposed to BidiAgent (e.g., "finance_agent").
        description: What this sub-agent does — Nova Sonic uses this to decide when to call it.
        system_prompt: Instructions for the sub-agent.
        tools: Optional list of @tool functions the sub-agent can use.
        model_id: Bedrock model ID for reasoning.
        region: AWS region.

    Returns:
        Agent instance (pass directly to BidiAgent tools list).
    """
    model = BedrockModel(model_id=model_id, region_name=region)

    return Agent(
        model=model,
        tools=tools or [],
        system_prompt=system_prompt,
        name=name,
        description=description,
    )

Multiple Sub-Agents Example

# subagents.py
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"


@tool
def get_transactions(account_id: str, limit: int = 5) -> str:
    """Get recent transactions."""
    return f"Last {limit} transactions for {account_id}: grocery $45, gas $38, coffee $6"


@tool
def lookup_faq(question: str) -> str:
    """Search the FAQ knowledge base."""
    return f"FAQ answer for '{question}': Our business hours are 9am-5pm Monday through Friday."


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, get_transactions],
        system_prompt="You are a finance assistant. Be precise with numbers. Keep responses to one sentence.",
        name="finance_agent",
        description="Handles financial queries: account balances, transactions, transfers. Call for any banking topic.",
    )


def create_support_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=[lookup_faq],
        system_prompt="You are a customer support agent. Answer questions using the FAQ. Be brief and helpful.",
        name="support_agent",
        description="Answers general questions about policies, hours, and services using the FAQ knowledge base.",
    )

Wiring Sub-Agents into BidiAgent

# agent.py
from subagents import create_finance_agent, create_support_agent

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}},
    )

    finance = create_finance_agent(region=REGION)
    support = create_support_agent(region=REGION)

    return BidiAgent(
        model=model,
        tools=[finance, support],
        system_prompt=SYSTEM_PROMPT,
    )

How It Works at Runtime

1. User speaks → Nova Sonic transcribes and processes 2. Nova Sonic decides a sub-agent's expertise is needed (based on description) 3. Nova Sonic emits a toolUse event with the sub-agent's name 4. BidiAgent invokes the sub-agent synchronously (text-based reasoning with Nova Lite) 5. Sub-agent uses its own tools, returns a text response 6. Response is sent back to Nova Sonic as a tool result 7. Nova Sonic speaks the response to the user 8. Audio streaming continues uninterrupted

Design Guidelines

File Structure

websocket/
├── server.py
├── agent.py            # Imports and wires sub-agents
├── subagents.py        # Sub-agent definitions
├── tools.py            # Simple @tool functions (optional, separate from sub-agents)
└── requirements.txt

Mixing Tools and Sub-Agents

You can pass both simple @tool functions and sub-agents in the same tools list:

from tools import get_weather, get_time
from subagents import create_finance_agent

def _create_agent() -> BidiAgent:
    model = BidiNovaSonicModel(...)
    finance = create_finance_agent()

    return BidiAgent(
        model=model,
        tools=[get_weather, get_time, finance],  # Mix simple tools + sub-agents
        system_prompt=SYSTEM_PROMPT,
    )

Nova Sonic treats them identically — each appears as a tool with a name and description.