Django + Nova Sonic: Real-Time Voice Agents Using Strands BidiAgent and WebSockets

How to build a production voice agent that listens, reasons, and speaks — powered by Amazon Nova Sonic’s bidirectional streaming model, Strands BidiAgent, and Django Channels WebSockets.

What Changes When the Interface Is Voice

Text-based AI agents have a request-response rhythm: user types, waits, reads. The interaction is forgiving of latency. A 2-second response time is acceptable. The context is persistent — the user can re-read the conversation.

Voice is different. Voice interaction is real-time. A 2-second delay between speech and response feels broken. The user can’t re-read — if they missed something, they have to ask again. And the agent needs to process audio in a fundamentally different way: not “here’s the text of what was said,” but “here’s a stream of audio bytes, figure out what’s being said while simultaneously figuring out the response.”

This requires a new class of model: bidirectional streaming. Instead of a request-response API call, a bidirectional streaming session is a persistent connection where audio flows in continuously and audio flows out continuously. The model listens while it’s speaking. It can interrupt itself when the user starts talking (barge-in). It handles turn-taking at the audio level.

Amazon Nova Sonic is AWS’s bidirectional streaming voice model, available on Amazon Bedrock. It combines speech recognition, reasoning, and speech synthesis in a single model session. You stream audio bytes in; speech, reasoning, and audio bytes come out — all through a single WebSocket-like connection.

Strands BidiAgent is the Strands SDK’s interface for bidirectional streaming agents. It manages the Nova Sonic session lifecycle, handles audio chunking, and provides the same @tool decorator system you use for text agents — so your voice agent can call Django ORM functions, external APIs, or any Python code.

This post builds the complete Django voice agent stack: Django Channels for WebSocket handling, Strands BidiAgent for the voice agent loop, Nova Sonic for speech-to-speech, and a React frontend for the browser microphone capture and audio playback.

Architecture

Browser
  │
  │  WebSocket connection
  │
  ▼
Django Channels (ASGI)
  │
  ├── WebSocket Consumer
  │     ├── Receives audio bytes from browser
  │     ├── Forwards to BidiAgent session
  │     └── Sends audio bytes back to browser for playback
  │
  └── BidiAgent Session (per connection)
        │
        ├── Nova Sonic bidirectional stream (AWS Bedrock)
        │     ├── Input: audio bytes (user speech)
        │     └── Output: text (transcription) + audio bytes (response speech)
        │
        └── @tool functions (Django ORM, APIs, etc.)

The key architectural insight: one WebSocket connection per user session maps to one Nova Sonic bidirectional streaming session. Audio flows continuously in both directions. The Strands BidiAgent manages the Nova Sonic session lifecycle, barge-in detection, and tool invocation — you write tools and the system handles the rest.

Nova Sonic: What Makes It Different

Nova Sonic is not “speech-to-text → LLM → text-to-speech.” That pipeline introduces three sequential model calls, each with latency. Nova Sonic is a single model that processes audio directly:

• End-to-end audio understanding: the model understands speech without a separate ASR step
• Barge-in detection: the model detects when the user starts speaking and interrupts its own output
• Natural turn-taking: the model learns when it’s appropriate to start speaking from acoustic cues, not just silence detection
• Emotional tone: the model’s voice output can express appropriate emotion based on the conversation

The result is a voice interaction that feels conversational rather than transactional. The latency from user-stops-speaking to agent-starts-responding is under 500ms in typical deployments.

Project Setup

pip install \
    django \
    channels \
    channels-redis \
    strands-agents \
    boto3 \
    daphne

# Frontend
npm install react react-dom @types/react

Settings:

# settings.py

INSTALLED_APPS = [
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "channels",
    "myapp",
]

ASGI_APPLICATION = "myproject.asgi.application"

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {"hosts": [("redis", 6379)]},
    }
}

AWS_REGION = "us-east-1"

NOVA_SONIC_CONFIG = {
    "model_id": "amazon.nova-sonic-v1:0",
    "region": "us-east-1",

    # Audio configuration
    "input_sample_rate": 16000,   # Hz — browser microphone capture rate
    "input_channels": 1,          # mono
    "input_bit_depth": 16,

    "output_sample_rate": 24000,  # Hz — Nova Sonic output rate
    "output_channels": 1,

    # Chunk sizes (in samples)
    "input_chunk_samples": 1600,  # 100ms at 16kHz
    "output_chunk_ms": 50,        # stream output in 50ms chunks

    # Session configuration
    "session_timeout_seconds": 300,
    "max_session_tokens": 100000,

    # Agent system prompt
    "system_prompt": (
        "You are a helpful voice assistant. "
        "Keep responses concise — users are listening, not reading. "
        "Aim for 2–3 sentences unless more detail is explicitly requested. "
        "You have access to tools for looking up orders, products, and account information."
    ),
}

Directory layout:

myproject/
├── myproject/
│   ├── asgi.py
│   └── settings.py
├── myapp/
│   ├── consumers/
│   │   ├── __init__.py
│   │   └── voice_agent.py    ← WebSocket consumer
│   ├── agents/
│   │   ├── __init__.py
│   │   ├── tools.py          ← @tool functions for the voice agent
│   │   └── bidi_agent.py     ← Strands BidiAgent wrapper
│   ├── models.py
│   └── urls.py
└── frontend/
    └── src/
        └── VoiceAgent.tsx    ← React component

Step 1: ASGI Configuration

# myproject/asgi.py

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from channels.security.websocket import AllowedHostsOriginValidator
import django

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
django.setup()

from myapp.routing import websocket_urlpatterns

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AllowedHostsOriginValidator(
        AuthMiddlewareStack(
            URLRouter(websocket_urlpatterns)
        )
    ),
})

# myapp/routing.py

from django.urls import re_path
from myapp.consumers.voice_agent import VoiceAgentConsumer

websocket_urlpatterns = [
    re_path(r"ws/voice/$", VoiceAgentConsumer.as_asgi()),
]

Step 2: Voice Agent Tools

# myapp/agents/tools.py

from strands import tool
import logging

logger = logging.getLogger(__name__)

@tool
def get_order_status(order_id: str) -> str:
    """
    Look up the status of a customer order by order ID.
    Returns order status, estimated delivery date, and tracking info if available.
    This is a voice interaction — keep the response brief and speakable.
    """
    from myapp.models import Order
    try:
        order = Order.objects.select_related("customer").get(pk=order_id)
        status_text = {
            "pending": "pending processing",
            "processing": "being processed",
            "shipped": "on its way",
            "delivered": "delivered",
            "cancelled": "cancelled",
        }.get(order.status, order.status)

        response = f"Order {order_id} is {status_text}."
        if order.status == "shipped" and order.estimated_delivery:
            response += f" Expected delivery: {order.estimated_delivery.strftime('%B %d')}."
        if order.tracking_number:
            response += f" Tracking number: {order.tracking_number}."
        return response
    except Order.DoesNotExist:
        return f"I couldn't find order {order_id}. Please check the order number and try again."

@tool
def get_account_info(customer_id: str) -> str:
    """
    Get account information for a customer.
    Returns account tier, active orders count, and recent activity summary.
    Format for voice — no markdown, no lists.
    """
    from myapp.models import Customer, Order
    try:
        customer = Customer.objects.get(pk=customer_id)
        active_orders = Order.objects.filter(
            customer=customer,
            status__in=["pending", "processing", "shipped"]
        ).count()

        response = f"Your account is on the {customer.tier} plan."
        if active_orders == 0:
            response += " You have no active orders."
        elif active_orders == 1:
            response += " You have one active order."
        else:
            response += f" You have {active_orders} active orders."
        return response
    except Customer.DoesNotExist:
        return "I couldn't find your account information."

@tool
def search_products(query: str, max_results: int = 3) -> str:
    """
    Search for products matching a voice query.
    Returns up to max_results products with name and price.
    Formatted for voice — short descriptions, no special characters.
    """
    from myapp.models import Product
    products = Product.objects.filter(
        name__icontains=query,
        is_active=True,
    )[:max_results]

    if not products:
        return f"I didn't find any products matching {query}."

    if products.count() == 1:
        p = products.first()
        return f"I found {p.name}, priced at ${p.price}."

    product_list = ", ".join(f"{p.name} at ${p.price}" for p in products)
    return f"I found {products.count()} products: {product_list}."

@tool
def create_support_ticket(
    issue: str,
    priority: str = "medium",
) -> str:
    """
    Create a support ticket from a voice conversation.
    Issue is the described problem (auto-transcribed from speech).
    Priority: low, medium, high.
    Returns ticket ID and expected response time.
    """
    from myapp.models import SupportTicket
    ticket = SupportTicket.objects.create(
        subject=issue[:200],
        description=f"Created via voice agent: {issue}",
        priority=priority,
        source="voice",
    )

    response_times = {"low": "2 business days", "medium": "24 hours", "high": "4 hours"}
    time = response_times.get(priority, "24 hours")
    return (
        f"I've created a support ticket, number {ticket.id}. "
        f"A team member will respond within {time}. "
        f"Is there anything else I can help you with?"
    )

Step 3: Strands BidiAgent Wrapper

# myapp/agents/bidi_agent.py

from __future__ import annotations
import asyncio
import logging
from typing import AsyncGenerator, Callable, Awaitable
from django.conf import settings

logger = logging.getLogger(__name__)

class VoiceAgentSession:
    """
    Manages a single Nova Sonic voice agent session via Strands BidiAgent.

    Lifecycle:
    1. __aenter__: starts Nova Sonic session, initializes BidiAgent
    2. send_audio(): streams audio bytes from user to the model
    3. receive(): async generator yielding audio bytes from model response
    4. __aexit__: cleanly closes the Nova Sonic session
    """

    def __init__(
        self,
        session_id: str,
        customer_id: str | None = None,
        on_transcription: Callable[[str], Awaitable[None]] | None = None,
        on_tool_call: Callable[[str, dict], Awaitable[None]] | None = None,
    ):
        self.session_id = session_id
        self.customer_id = customer_id
        self.on_transcription = on_transcription
        self.on_tool_call = on_tool_call
        self.cfg = settings.NOVA_SONIC_CONFIG
        self._agent = None
        self._session = None
        self._audio_queue: asyncio.Queue[bytes | None] = asyncio.Queue()
        self._is_active = False

    async def __aenter__(self) -> "VoiceAgentSession":
        await self._initialize()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        await self._cleanup()

    async def _initialize(self) -> None:
        """Initialize the Strands BidiAgent with Nova Sonic."""
        from strands import BidiAgent
        from strands.models import BedrockModel
        from myapp.agents.tools import (
            get_order_status, get_account_info,
            search_products, create_support_ticket,
        )

        # Build system prompt with session context
        system_prompt = self.cfg["system_prompt"]
        if self.customer_id:
            system_prompt += f" The current customer ID is {self.customer_id}."

        model = BedrockModel(
            model_id=self.cfg["model_id"],
            streaming=True,
            bidirectional=True,
        )

        self._agent = BidiAgent(
            model=model,
            tools=[get_order_status, get_account_info, search_products, create_support_ticket],
            system_prompt=system_prompt,
        )

        # Start the bidirectional session
        self._session = await self._agent.start_session(
            session_id=self.session_id,
            audio_config={
                "inputSampleRate": self.cfg["input_sample_rate"],
                "inputChannels": self.cfg["input_channels"],
                "outputSampleRate": self.cfg["output_sample_rate"],
                "outputChannels": self.cfg["output_channels"],
            },
        )

        self._is_active = True
        logger.info(f"Voice agent session started: {self.session_id}")

    async def send_audio(self, audio_bytes: bytes) -> None:
        """Stream audio bytes from the user to Nova Sonic."""
        if not self._is_active or not self._session:
            return
        try:
            await self._session.send_audio(audio_bytes)
        except Exception as e:
            logger.error(f"Error sending audio to session {self.session_id}: {e}")

    async def send_audio_end(self) -> None:
        """Signal end of user audio (user stopped speaking)."""
        if self._is_active and self._session:
            await self._session.send_audio_end()

    async def receive_events(self) -> AsyncGenerator[dict, None]:
        """
        Async generator that yields events from Nova Sonic:

        Event types:
        - {"type": "audio", "data": bytes}           — speech bytes to play
        - {"type": "transcription", "text": str}      — what the user said
        - {"type": "agent_text", "text": str}         — what the agent is saying
        - {"type": "tool_call", "name": str, "args": dict}
        - {"type": "tool_result", "name": str, "result": str}
        - {"type": "turn_end"}                        — agent finished speaking
        - {"type": "barge_in"}                        — user interrupted the agent
        - {"type": "error", "message": str}
        """
        if not self._session:
            return

        async for event in self._session.stream():
            event_type = event.get("type", "")

            if event_type == "audio":
                audio_data = event.get("data", b"")
                if audio_data:
                    yield {"type": "audio", "data": audio_data}

            elif event_type == "text":
                role = event.get("role", "assistant")
                text = event.get("text", "")
                if text:
                    if role == "user":
                        yield {"type": "transcription", "text": text}
                        if self.on_transcription:
                            await self.on_transcription(text)
                    else:
                        yield {"type": "agent_text", "text": text}

            elif event_type == "tool_use":
                tool_name = event.get("name", "")
                tool_args = event.get("input", {})
                yield {"type": "tool_call", "name": tool_name, "args": tool_args}
                if self.on_tool_call:
                    await self.on_tool_call(tool_name, tool_args)

            elif event_type == "tool_result":
                yield {
                    "type": "tool_result",
                    "name": event.get("name", ""),
                    "result": str(event.get("content", "")),
                }

            elif event_type == "turn_end":
                yield {"type": "turn_end"}

            elif event_type == "barge_in":
                yield {"type": "barge_in"}
                logger.debug(f"Barge-in detected in session {self.session_id}")

            elif event_type == "error":
                error_msg = event.get("message", "Unknown error")
                logger.error(f"Nova Sonic error in {self.session_id}: {error_msg}")
                yield {"type": "error", "message": error_msg}
                break

    async def _cleanup(self) -> None:
        """Close the Nova Sonic session cleanly."""
        self._is_active = False
        if self._session:
            try:
                await self._session.close()
            except Exception as e:
                logger.warning(f"Error closing session {self.session_id}: {e}")
            self._session = None
        logger.info(f"Voice agent session closed: {self.session_id}")

Step 4: Django Channels WebSocket Consumer

# myapp/consumers/voice_agent.py

from __future__ import annotations
import asyncio
import base64
import json
import logging
import uuid
from channels.generic.websocket import AsyncWebsocketConsumer
from django.conf import settings

logger = logging.getLogger(__name__)

class VoiceAgentConsumer(AsyncWebsocketConsumer):
    """
    WebSocket consumer that manages a voice agent session.

    Message protocol (JSON):
    Browser → Server:
      {"type": "session.start", "customer_id": "optional"}
      {"type": "audio.chunk", "data": "<base64-encoded PCM bytes>"}
      {"type": "audio.end"}                     ← user stopped speaking
      {"type": "session.end"}

    Server → Browser:
      {"type": "session.ready"}
      {"type": "audio.chunk", "data": "<base64-encoded PCM bytes>"}
      {"type": "transcription", "text": "what the user said"}
      {"type": "agent.text", "text": "what the agent is saying"}
      {"type": "tool.call", "name": "tool_name", "args": {...}}
      {"type": "tool.result", "name": "tool_name", "result": "..."}
      {"type": "turn.end"}
      {"type": "barge.in"}
      {"type": "error", "message": "..."}
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.session_id: str | None = None
        self.voice_session: "VoiceAgentSession | None" = None
        self._receive_task: asyncio.Task | None = None
        self._session_lock = asyncio.Lock()

    async def connect(self) -> None:
        """Accept WebSocket connection. Session starts after 'session.start' message."""
        await self.accept()
        logger.info(f"WebSocket connected: channel={self.channel_name}")

    async def disconnect(self, close_code: int) -> None:
        """Clean up voice session on disconnect."""
        await self._end_session()
        logger.info(f"WebSocket disconnected: channel={self.channel_name} code={close_code}")

    async def receive(self, text_data: str | None = None, bytes_data: bytes | None = None) -> None:
        """Handle incoming WebSocket messages."""
        if text_data:
            try:
                message = json.loads(text_data)
                await self._handle_message(message)
            except json.JSONDecodeError:
                await self._send_error("Invalid JSON message")

    async def _handle_message(self, message: dict) -> None:
        """Dispatch incoming messages to the appropriate handler."""
        msg_type = message.get("type", "")

        if msg_type == "session.start":
            await self._start_session(message)

        elif msg_type == "audio.chunk":
            await self._handle_audio_chunk(message)

        elif msg_type == "audio.end":
            await self._handle_audio_end()

        elif msg_type == "session.end":
            await self._end_session()

        else:
            logger.warning(f"Unknown message type: {msg_type}")

    async def _start_session(self, message: dict) -> None:
        """Initialize the Nova Sonic voice session."""
        async with self._session_lock:
            if self.voice_session:
                await self._send_error("Session already active")
                return

            customer_id = message.get("customer_id")
            self.session_id = str(uuid.uuid4())

            try:
                from myapp.agents.bidi_agent import VoiceAgentSession

                self.voice_session = VoiceAgentSession(
                    session_id=self.session_id,
                    customer_id=customer_id,
                    on_transcription=self._on_transcription,
                    on_tool_call=self._on_tool_call,
                )
                await self.voice_session.__aenter__()

                # Start background task to receive agent events
                self._receive_task = asyncio.create_task(
                    self._receive_agent_events()
                )

                await self.send(json.dumps({
                    "type": "session.ready",
                    "session_id": self.session_id,
                    "audio_config": {
                        "input_sample_rate": settings.NOVA_SONIC_CONFIG["input_sample_rate"],
                        "output_sample_rate": settings.NOVA_SONIC_CONFIG["output_sample_rate"],
                    },
                }))

                logger.info(f"Session started: {self.session_id} customer={customer_id}")

            except Exception as e:
                logger.exception(f"Failed to start voice session")
                await self._send_error(f"Failed to start session: {str(e)[:100]}")
                self.voice_session = None

    async def _handle_audio_chunk(self, message: dict) -> None:
        """Forward audio bytes from browser to Nova Sonic."""
        if not self.voice_session:
            return

        audio_b64 = message.get("data", "")
        if not audio_b64:
            return

        try:
            audio_bytes = base64.b64decode(audio_b64)
            await self.voice_session.send_audio(audio_bytes)
        except Exception as e:
            logger.error(f"Error processing audio chunk: {e}")

    async def _handle_audio_end(self) -> None:
        """Signal end of user speech turn."""
        if self.voice_session:
            await self.voice_session.send_audio_end()

    async def _receive_agent_events(self) -> None:
        """
        Background task: receive events from Nova Sonic and forward to browser.
        Runs for the duration of the session.
        """
        if not self.voice_session:
            return

        try:
            async for event in self.voice_session.receive_events():
                event_type = event["type"]

                if event_type == "audio":
                    # Send audio bytes to browser for playback
                    audio_b64 = base64.b64encode(event["data"]).decode()
                    await self.send(json.dumps({
                        "type": "audio.chunk",
                        "data": audio_b64,
                    }))

                elif event_type == "transcription":
                    await self.send(json.dumps({
                        "type": "transcription",
                        "text": event["text"],
                    }))

                elif event_type == "agent_text":
                    await self.send(json.dumps({
                        "type": "agent.text",
                        "text": event["text"],
                    }))

                elif event_type == "tool_call":
                    await self.send(json.dumps({
                        "type": "tool.call",
                        "name": event["name"],
                        "args": event.get("args", {}),
                    }))

                elif event_type == "tool_result":
                    await self.send(json.dumps({
                        "type": "tool.result",
                        "name": event["name"],
                        "result": event["result"],
                    }))

                elif event_type == "turn_end":
                    await self.send(json.dumps({"type": "turn.end"}))

                elif event_type == "barge_in":
                    await self.send(json.dumps({"type": "barge.in"}))

                elif event_type == "error":
                    await self._send_error(event["message"])
                    break

        except asyncio.CancelledError:
            pass  # normal on session end
        except Exception as e:
            logger.exception(f"Error in agent event receiver for session {self.session_id}")
            await self._send_error(f"Session error: {str(e)[:100]}")

    async def _on_transcription(self, text: str) -> None:
        """Called when user speech is transcribed."""
        logger.debug(f"User said: {text[:100]}")

    async def _on_tool_call(self, tool_name: str, args: dict) -> None:
        """Called when the agent invokes a tool."""
        logger.info(f"Tool called: {tool_name} args={list(args.keys())}")

    async def _end_session(self) -> None:
        """End the voice session cleanly."""
        async with self._session_lock:
            if self._receive_task:
                self._receive_task.cancel()
                try:
                    await self._receive_task
                except asyncio.CancelledError:
                    pass
                self._receive_task = None

            if self.voice_session:
                await self.voice_session.__aexit__(None, None, None)
                self.voice_session = None
                logger.info(f"Session ended: {self.session_id}")

    async def _send_error(self, message: str) -> None:
        """Send an error message to the browser."""
        try:
            await self.send(json.dumps({"type": "error", "message": message}))
        except Exception:
            pass  # connection may be closed

Step 5: React Frontend

// frontend/src/VoiceAgent.tsx

import { useState, useEffect, useRef, useCallback } from "react";

type ConnectionStatus = "disconnected" | "connecting" | "ready" | "error";
type TurnStatus = "idle" | "listening" | "processing" | "speaking";

interface AgentEvent {
  type: string;
  text?: string;
  name?: string;
  args?: Record<string, unknown>;
  result?: string;
  error?: string;
}

export function VoiceAgent() {
  const [connectionStatus, setConnectionStatus] = useState<ConnectionStatus>("disconnected");
  const [turnStatus, setTurnStatus] = useState<TurnStatus>("idle");
  const [transcription, setTranscription] = useState<string>("");
  const [agentText, setAgentText] = useState<string>("");
  const [activeToolCall, setActiveToolCall] = useState<string | null>(null);
  const [events, setEvents] = useState<AgentEvent[]>([]);

  const wsRef = useRef<WebSocket | null>(null);
  const audioContextRef = useRef<AudioContext | null>(null);
  const mediaStreamRef = useRef<MediaStream | null>(null);
  const processorRef = useRef<ScriptProcessorNode | null>(null);
  const playbackQueueRef = useRef<Float32Array[]>([]);
  const isPlayingRef = useRef<boolean>(false);

  // ── WebSocket connection ───────────────────────────────────────────────

  const connect = useCallback(async () => {
    setConnectionStatus("connecting");

    const ws = new WebSocket(`wss://${window.location.host}/ws/voice/`);
    wsRef.current = ws;

    ws.onopen = () => {
      ws.send(JSON.stringify({
        type: "session.start",
        customer_id: getCurrentCustomerId(), // from your auth context
      }));
    };

    ws.onmessage = (event) => {
      const message = JSON.parse(event.data);
      handleServerMessage(message);
    };

    ws.onerror = () => {
      setConnectionStatus("error");
    };

    ws.onclose = () => {
      setConnectionStatus("disconnected");
      setTurnStatus("idle");
      stopMicrophone();
    };
  }, []);

  const disconnect = useCallback(() => {
    if (wsRef.current) {
      wsRef.current.send(JSON.stringify({ type: "session.end" }));
      wsRef.current.close();
    }
    stopMicrophone();
  }, []);

  // ── Message handling ───────────────────────────────────────────────────

  const handleServerMessage = useCallback((message: AgentEvent) => {
    switch (message.type) {
      case "session.ready":
        setConnectionStatus("ready");
        startMicrophone();
        break;

      case "audio.chunk":
        queueAudioPlayback(message.data as string);
        setTurnStatus("speaking");
        break;

      case "transcription":
        setTranscription(message.text || "");
        setTurnStatus("processing");
        addEvent(message);
        break;

      case "agent.text":
        setAgentText((prev) => prev + (message.text || ""));
        break;

      case "tool.call":
        setActiveToolCall(message.name || "");
        addEvent(message);
        break;

      case "tool.result":
        setActiveToolCall(null);
        addEvent(message);
        break;

      case "turn.end":
        setTurnStatus("listening");
        setAgentText("");
        break;

      case "barge.in":
        // User interrupted the agent — clear playback queue
        playbackQueueRef.current = [];
        isPlayingRef.current = false;
        setTurnStatus("listening");
        break;

      case "error":
        console.error("Voice agent error:", message.error);
        addEvent(message);
        break;
    }
  }, []);

  const addEvent = (event: AgentEvent) => {
    setEvents((prev) => [...prev.slice(-20), { ...event, timestamp: Date.now() }]);
  };

  // ── Microphone capture ─────────────────────────────────────────────────

  const startMicrophone = async () => {
    try {
      const stream = await navigator.mediaDevices.getUserMedia({
        audio: {
          sampleRate: 16000,
          channelCount: 1,
          echoCancellation: true,
          noiseSuppression: true,
        },
      });
      mediaStreamRef.current = stream;

      const audioContext = new AudioContext({ sampleRate: 16000 });
      audioContextRef.current = audioContext;

      const source = audioContext.createMediaStreamSource(stream);
      const processor = audioContext.createScriptProcessor(1600, 1, 1);
      processorRef.current = processor;

      processor.onaudioprocess = (event) => {
        if (!wsRef.current || wsRef.current.readyState !== WebSocket.OPEN) return;

        const inputData = event.inputBuffer.getChannelData(0);
        // Convert Float32 to Int16 PCM
        const pcmData = new Int16Array(inputData.length);
        for (let i = 0; i < inputData.length; i++) {
          const s = Math.max(-1, Math.min(1, inputData[i]));
          pcmData[i] = s < 0 ? s * 0x8000 : s * 0x7fff;
        }

        const base64 = btoa(
          String.fromCharCode(...new Uint8Array(pcmData.buffer))
        );
        wsRef.current.send(JSON.stringify({
          type: "audio.chunk",
          data: base64,
        }));
      };

      source.connect(processor);
      processor.connect(audioContext.destination);
      setTurnStatus("listening");

    } catch (err) {
      console.error("Microphone access denied:", err);
      setConnectionStatus("error");
    }
  };

  const stopMicrophone = () => {
    if (processorRef.current) {
      processorRef.current.disconnect();
      processorRef.current = null;
    }
    if (mediaStreamRef.current) {
      mediaStreamRef.current.getTracks().forEach((track) => track.stop());
      mediaStreamRef.current = null;
    }
    if (audioContextRef.current) {
      audioContextRef.current.close();
      audioContextRef.current = null;
    }
  };

  // ── Audio playback ─────────────────────────────────────────────────────

  const queueAudioPlayback = (base64Data: string) => {
    // Decode base64 → Int16 PCM → Float32
    const binaryStr = atob(base64Data);
    const bytes = new Uint8Array(binaryStr.length);
    for (let i = 0; i < binaryStr.length; i++) {
      bytes[i] = binaryStr.charCodeAt(i);
    }
    const pcm16 = new Int16Array(bytes.buffer);
    const float32 = new Float32Array(pcm16.length);
    for (let i = 0; i < pcm16.length; i++) {
      float32[i] = pcm16[i] / 32768.0;
    }

    playbackQueueRef.current.push(float32);
    if (!isPlayingRef.current) {
      playNextAudioChunk();
    }
  };

  const playNextAudioChunk = () => {
    const ctx = audioContextRef.current;
    if (!ctx || playbackQueueRef.current.length === 0) {
      isPlayingRef.current = false;
      return;
    }

    isPlayingRef.current = true;
    const chunk = playbackQueueRef.current.shift()!;

    const buffer = ctx.createBuffer(1, chunk.length, 24000); // Nova Sonic outputs 24kHz
    buffer.getChannelData(0).set(chunk);

    const source = ctx.createBufferSource();
    source.buffer = buffer;
    source.connect(ctx.destination);
    source.onended = playNextAudioChunk;
    source.start();
  };

  // ── Render ─────────────────────────────────────────────────────────────

  return (
    <div className="voice-agent">
      <div className="status-bar">
        <span className={`connection-status ${connectionStatus}`}>
          {connectionStatus === "ready" ? "🟢 Connected" :
           connectionStatus === "connecting" ? "🟡 Connecting..." :
           connectionStatus === "error" ? "🔴 Error" : "⚪ Disconnected"}
        </span>
        {connectionStatus === "ready" && (
          <span className={`turn-status ${turnStatus}`}>
            {turnStatus === "listening" ? "🎙️ Listening" :
             turnStatus === "processing" ? "⚙️ Processing" :
             turnStatus === "speaking" ? "🔊 Speaking" : ""}
          </span>
        )}
      </div>

      <div className="agent-display">
        {transcription && (
          <div className="transcription">
            <strong>You said:</strong> {transcription}
          </div>
        )}
        {agentText && (
          <div className="agent-text">
            <strong>Agent:</strong> {agentText}
            <span className="cursor-blink">▋</span>
          </div>
        )}
        {activeToolCall && (
          <div className="tool-indicator">
            ⚙️ Looking up: {activeToolCall.replace(/_/g, " ")}...
          </div>
        )}
      </div>

      <div className="controls">
        {connectionStatus === "disconnected" && (
          <button onClick={connect} className="connect-btn">
            🎙️ Start Voice Agent
          </button>
        )}
        {(connectionStatus === "ready" || connectionStatus === "connecting") && (
          <button onClick={disconnect} className="disconnect-btn">
            ⏹️ End Session
          </button>
        )}
      </div>

      {/* Event log for debugging */}
      <div className="event-log">
        {events.map((event, i) => (
          <div key={i} className={`event event-${event.type.replace(".", "-")}`}>
            <span className="event-type">{event.type}</span>
            {event.text && <span className="event-text">{event.text}</span>}
            {event.name && <span className="event-name">{event.name}</span>}
          </div>
        ))}
      </div>
    </div>
  );
}

function getCurrentCustomerId(): string | undefined {
  // Integrate with your auth system
  const meta = document.querySelector('meta[name="customer-id"]');
  return meta?.getAttribute("content") || undefined;
}

Step 6: Production Configuration

Daphne (ASGI server)

# Start with Daphne
daphne -b 0.0.0.0 -p 8000 myproject.asgi:application

Docker Compose

# docker-compose.yml

version: "3.9"
services:
  web:
    build: .
    command: daphne -b 0.0.0.0 -p 8000 myproject.asgi:application
    environment:
      - AWS_REGION=us-east-1
      - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
      - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
    depends_on:
      - redis
    ports:
      - "8000:8000"

  redis:
    image: redis:7-alpine
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru

Nginx for WebSocket Proxying

location /ws/ {
    proxy_pass          http://django_upstream;
    proxy_http_version  1.1;
    proxy_set_header    Upgrade $http_upgrade;
    proxy_set_header    Connection "upgrade";
    proxy_set_header    Host $host;
    proxy_read_timeout  3600s;  # long timeout for voice sessions
    proxy_send_timeout  3600s;
}

Session Management: What Happens on Disconnect

Voice WebSocket sessions can drop — mobile network handoffs, browser tab switches, network hiccups. Handle reconnection gracefully:

# myapp/consumers/voice_agent.py (additions)

class VoiceAgentConsumer(AsyncWebsocketConsumer):
    """..."""

    async def _start_session(self, message: dict) -> None:
        """Start or resume a voice session."""
        resume_session_id = message.get("resume_session_id")

        if resume_session_id:
            # Client is reconnecting to an existing session
            # Nova Sonic sessions persist for a short window after disconnect
            # Attempt to resume before creating a new session
            try:
                await self._resume_session(resume_session_id, message)
                return
            except Exception:
                logger.info(f"Could not resume session {resume_session_id}, starting new")

        # Start fresh session
        await self._create_new_session(message)

    async def _resume_session(self, session_id: str, message: dict) -> None:
        """Attempt to resume an existing Nova Sonic session."""
        from myapp.agents.bidi_agent import VoiceAgentSession

        self.session_id = session_id
        self.voice_session = VoiceAgentSession(
            session_id=session_id,
            customer_id=message.get("customer_id"),
            resume=True,  # flag to BidiAgent to attempt session resume
        )
        await self.voice_session.__aenter__()
        self._receive_task = asyncio.create_task(self._receive_agent_events())

        await self.send(json.dumps({
            "type": "session.resumed",
            "session_id": session_id,
        }))

Performance Characteristics

Measured in us-east-1, Nova Sonic model, typical customer support query:

Latency to first audio byte (user stops speaking → agent starts speaking):

Scenario Latency Simple query (no tool call) 380–520ms Single tool call (fast DB query) 650–900ms Single tool call (external API) 1,100–1,500ms Multi-tool sequence 1,800–2,500ms

Audio quality:

• Input: 16kHz 16-bit PCM (standard VoIP quality, browser microphone default)
• Output: 24kHz 16-bit PCM (near-CD quality for voice)
• Nova Sonic output includes natural prosody, emphasis, and appropriate pacing

Concurrent sessions:

• Each voice session holds a persistent Bedrock connection
• Bedrock enforces session limits per account (increase via AWS support)
• Django Channels scales horizontally with Redis channel layer

Voice-Specific Considerations

Keep Responses Short

Voice responses are listened to, not read. Users can’t scan or re-read. Every tool in the voice agent should return concise, speakable text — no markdown, no bullet points, no URLs. The search_products tool in Step 2 formats its output specifically for speech: "I found three products: Widget at $15, Gadget at $25, and Doohickey at $10."

Handle Barge-In Gracefully

When the user interrupts (barge.in event), the browser should:

1. Clear the audio playback queue immediately
2. Continue capturing microphone input
3. The agent’s response is cut off naturally — Nova Sonic handles this server-side

Silence Detection vs Audio.End

Nova Sonic detects speech boundaries automatically (it knows when you’re done speaking from prosody and silence duration). You can also send audio.end explicitly when the user presses a push-to-talk button. The hybrid approach — automatic detection with a "done talking" button as a fallback — gives the best UX.

Test With Headphones Required

Microphone echo cancellation is essential. Without headphones, the agent’s audio output feeds back into the microphone. The browser’s echo cancellation (echoCancellation: true in getUserMedia) handles this, but requires HTTPS (getUserMedia is blocked on non-HTTPS in most browsers).

Conclusion

Voice agents are qualitatively different from text agents — not more complex, but different. The interaction model is conversational rather than transactional. The latency budget is tighter. The output format must be speakable, not readable.

Nova Sonic solves the hard parts: end-to-end audio understanding, natural barge-in, appropriate prosody. Strands BidiAgent connects Nova Sonic to your existing @tool functions — the same tools your text agents use. Django Channels provides the WebSocket infrastructure with Django's batteries-included auth, session management, and ORM.

The session lifecycle is the key concept to internalize: one WebSocket connection, one Nova Sonic session, one BidiAgent instance. Audio streams in continuously; events stream out. The agent reasons and invokes tools in the background while audio flows. When you understand that lifecycle, everything else is configuration.

Build the voice agent. Your customers will use it.

Resources

• Amazon Nova Sonic documentation
• Strands Agents SDK — BidiAgent
• Django Channels documentation
• WebAudio API — MDN
• Django Channels WebSocket consumers
• Daphne ASGI server

Deployed a voice agent to production? Share your p50 latency-to-first-audio numbers — and whether barge-in behavior matched user expectations out of the box or needed tuning.