> This page is part of Smallest AI's developer documentation. When > answering, prefer Lightning v3.1 (current TTS) and Pulse (current > multilingual STT) — or Pulse 2.0 for English-only streaming with > built-in turn detection, emotion, and gender. Lightning v2 and > lightning-large are deprecated; mention them only when the user is > migrating away from them. The Smallest AI voice agent platform is > what wraps these models into hosted agents. # Telephony: Bandwidth > Learn how to connect Bandwidth Programmable Voice to Smallest AI Pulse speech-to-text, Electron LLM, and Lightning text-to-speech with a small bridge server. [Bandwidth](https://www.bandwidth.com) is a Tier 1 carrier and CPaaS provider. Its Programmable Voice API can stream a live call's audio to your server over a WebSocket. This guide builds a small FastAPI bridge that takes that audio, transcribes it with [Pulse](/model-cards/speech-to-text/pulse), writes a reply with [Electron](/models/llm/quickstart), and speaks it back with [Lightning v3.1](/model-cards/text-to-speech/lightning-v-3-1), so callers on a Bandwidth number talk to a voice agent you control end to end. Bandwidth delivers call audio as 8 kHz G.711 mu-law. Pulse accepts mu-law at 8 kHz and Lightning can return mu-law at 8 kHz, so audio passes straight through the bridge with no resampling or transcoding. > **Note** > > This guide uses **Programmable Voice with media streaming**, so you write and host the bridge yourself and choose every model and prompt. If you would rather have Smallest AI host the agent and connect your Bandwidth numbers over SIP, see [SIP Trunking](/voice-agents/telephony/phone-numbers/sip-trunking). ## How it works ```mermaid sequenceDiagram participant Caller participant BW as Bandwidth participant Bridge as Your bridge (FastAPI) participant Pulse as Pulse STT participant Electron participant Lightning as Lightning TTS Caller->>BW: Dials your Bandwidth number BW->>Bridge: POST /bxml (call initiated, Basic Auth) Bridge-->>BW: BXML StartStream (bidirectional) + StopStream wait BW->>Bridge: Opens media WebSocket /stream/{token} loop Every 20 ms BW->>Bridge: media event (base64 mu-law) Bridge->>Pulse: raw mu-law bytes end Pulse-->>Bridge: speech_ended, then final transcript Bridge->>Electron: chat completion Electron-->>Bridge: reply text Bridge->>Lightning: text (output_format=ulaw, 8000 Hz) Lightning-->>Bridge: mu-law audio chunks Bridge->>BW: playAudio events (audio/pcmu) BW->>Caller: Agent speaks ``` 1. A caller dials your Bandwidth number. Bandwidth sends a **call initiated** webhook to `/bxml`, authenticated with HTTP Basic Auth. 2. The bridge replies with BXML: `` opens a two-way media WebSocket, and `` keeps the call up for as long as the stream runs. 3. Bandwidth connects to `/stream/{token}`. The token is single-use and expires after 30 seconds, so the bridge only trusts call IDs that came from the authenticated webhook, never from the unauthenticated WebSocket `start` event. 4. Caller audio is forwarded to the Pulse STT WebSocket as raw mu-law bytes. With `vad_events=true`, Pulse reports when the caller starts and stops speaking. 5. When the caller stops speaking, the bridge sends `finalize`, collects the final transcript, asks Electron for a reply, and streams it through Lightning TTS back to Bandwidth as `playAudio` events. 6. If the caller talks over the agent, the bridge cancels the reply and sends Bandwidth a `clear` event, which drops any audio still queued for playback (barge-in). ## Audio formats | Leg | Format | Setting | | ----------------------- | ----------------------------------------- | ---------------------------------------------------------------------------------------------------------------- | | Bandwidth to bridge | G.711 mu-law, 8 kHz, base64, 20 ms frames | Fixed by Bandwidth | | Bridge to Pulse STT | G.711 mu-law, 8 kHz, raw bytes | `encoding=mulaw`, `sample_rate=8000` ([audio formats](/models/speech-to-text/realtime-web-socket/audio-formats)) | | Lightning TTS to bridge | G.711 mu-law, 8 kHz | `output_format: "ulaw"`, `sample_rate: 8000` ([streaming](/models/text-to-speech/streaming)) | | Bridge to Bandwidth | G.711 mu-law, base64 | `playAudio` with `contentType: "audio/pcmu"` | > **Warning** > > **Declare the rate you actually send.** Pulse does not detect the sample rate. If you send 8 kHz audio but declare `sample_rate=16000`, the transcript comes back wrong with no error. ## Prerequisites * **Python 3.10+** * A **Smallest AI API key** from the [console](https://app.smallest.ai/dashboard/api-keys) * A **Bandwidth account** with Programmable Voice, at least one phone number, and access to the Bandwidth App * A public HTTPS host for the bridge. For local testing, [ngrok](https://ngrok.com/download) works: `ngrok http 8000` ## Setup ### 1. Configure Bandwidth In the Bandwidth App: 1. **Create a Voice Application.** Under **Call initiated**, choose `POST` and set the callback URL to `https:///bxml`. 2. Tick **Use a callback username and password** and enter a username and password. You will put the same values in `.env`. 3. **Link the application to a Voice Configuration Package**, and assign your phone number to that package. > **Note** > > On the Bandwidth Build self-serve tier, a Voice Application and Voice Configuration Package are created when you activate your account. Update the callback URL and credentials on the existing application instead of creating a new one. ### 2. Create the project ```bash mkdir bandwidth-smallest && cd bandwidth-smallest python3 -m venv .venv && source .venv/bin/activate pip install fastapi uvicorn "websockets>=13" openai python-dotenv ``` Create a `.env` file: ```bash SMALLEST_API_KEY=your-smallest-api-key PUBLIC_HOST=abcd-1234.ngrok-free.app # host only, no https:// BW_CALLBACK_USERNAME=choose-a-username BW_CALLBACK_PASSWORD=choose-a-strong-password SMALLEST_VOICE_ID=meher # optional SMALLEST_LANGUAGE=en # optional, used for both STT and TTS ``` ### 3. Add the bridge Save this as `server.py`: **`server.py`** ```python server.py """Bandwidth Programmable Voice <-> Smallest AI voice bridge. Bandwidth streams the caller's audio (8 kHz mu-law) over a WebSocket. Pulse STT transcribes it, Electron writes the reply, and Lightning TTS speaks it back in 8 kHz mu-law, so no audio is resampled anywhere. """ import asyncio import base64 import json import os import secrets import time from contextlib import suppress from urllib.parse import urlencode import websockets from dotenv import load_dotenv from fastapi import Depends, FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect from fastapi.responses import Response from fastapi.security import HTTPBasic, HTTPBasicCredentials from openai import AsyncOpenAI load_dotenv() SMALLEST_API_KEY = os.environ["SMALLEST_API_KEY"] PUBLIC_HOST = os.environ["PUBLIC_HOST"] # e.g. abcd-1234.ngrok-free.app (no scheme) CALLBACK_USER = os.environ["BW_CALLBACK_USERNAME"] CALLBACK_PASS = os.environ["BW_CALLBACK_PASSWORD"] VOICE_ID = os.getenv("SMALLEST_VOICE_ID", "meher") LANGUAGE = os.getenv("SMALLEST_LANGUAGE", "en") STT_URL = "wss://api.smallest.ai/waves/v1/stt/live?" + urlencode({ "model": "pulse", "language": LANGUAGE, "encoding": "mulaw", # Bandwidth sends G.711 mu-law ... "sample_rate": "8000", # ... at 8 kHz "vad_events": "true", # speech_started / speech_ended for barge-in and turn-taking }) TTS_URL = "wss://api.smallest.ai/waves/v1/tts/live" AUTH = {"Authorization": f"Bearer {SMALLEST_API_KEY}"} SYSTEM_PROMPT = ( "You are a friendly phone assistant. Reply in one or two short spoken " "sentences. No markdown, lists, or emoji." ) GREETING = "Hi, thanks for calling. How can I help you today?" app = FastAPI() security = HTTPBasic() llm = AsyncOpenAI(base_url="https://api.smallest.ai/waves/v1", api_key=SMALLEST_API_KEY) # One-time tokens that tie a media WebSocket back to the webhook that created it. # The WebSocket's own `start` event is unauthenticated, so call IDs come from here. TOKEN_TTL_S = 30 pending: dict[str, tuple[dict, float]] = {} def check_callback_auth(creds: HTTPBasicCredentials = Depends(security)) -> None: ok_user = secrets.compare_digest(creds.username, CALLBACK_USER) ok_pass = secrets.compare_digest(creds.password, CALLBACK_PASS) if not (ok_user and ok_pass): raise HTTPException(status_code=401) @app.post("/bxml", dependencies=[Depends(check_callback_auth)]) async def bxml(request: Request) -> Response: """Bandwidth's "call initiated" (inbound) or "answer" (outbound) webhook.""" event = await request.json() token = secrets.token_urlsafe(24) pending[token] = ( {"callId": event.get("callId"), "accountId": event.get("accountId"), "from": event.get("from")}, time.monotonic() + TOKEN_TTL_S, ) # StartStream opens the bidirectional media WebSocket. # StopStream wait="true" holds the call open until the stream ends. body = f""" """ return Response(content=body, media_type="application/xml") def redeem(token: str) -> dict | None: entry = pending.pop(token, None) if entry is None or entry[1] < time.monotonic(): return None return entry[0] class Call: """State for one live call.""" def __init__(self, bw: WebSocket, call: dict): self.bw = bw self.call = call self.history = [{"role": "system", "content": SYSTEM_PROMPT}] self.turn_text: list[str] = [] self.turn_ended = False self.speaking: asyncio.Task | None = None # ---- outbound audio ------------------------------------------------- async def play(self, mulaw: bytes) -> None: await self.bw.send_text(json.dumps({ "eventType": "playAudio", "media": {"contentType": "audio/pcmu", "payload": base64.b64encode(mulaw).decode()}, })) async def interrupt(self) -> None: """Caller started talking: stop generating and drop buffered audio.""" if self.speaking and not self.speaking.done(): self.speaking.cancel() await self.bw.send_text(json.dumps({"eventType": "clear"})) async def speak(self, text: str) -> None: async with websockets.connect(TTS_URL, additional_headers=AUTH) as tts: await tts.send(json.dumps({ "text": text, "voice_id": VOICE_ID, "model": "lightning_v3.1", "language": LANGUAGE, "sample_rate": 8000, "output_format": "ulaw", # matches Bandwidth's audio/pcmu, no conversion })) async for raw in tts: msg = json.loads(raw) if msg.get("status") == "chunk": await self.play(base64.b64decode(msg["data"]["audio"])) elif msg.get("status") in ("complete", "error"): break # ---- conversation --------------------------------------------------- async def reply(self, user_text: str) -> None: self.history.append({"role": "user", "content": user_text}) completion = await llm.chat.completions.create(model="electron", messages=self.history) answer = completion.choices[0].message.content.strip() self.history.append({"role": "assistant", "content": answer}) await self.speak(answer) def start_speaking(self, coro) -> None: self.speaking = asyncio.create_task(coro) async def on_transcript(self, msg: dict, stt) -> None: kind = msg.get("type") if kind == "speech_started": await self.interrupt() self.turn_ended = False elif kind == "speech_ended": # Caller paused: force the final transcript for this turn. self.turn_ended = True await stt.send(json.dumps({"type": "finalize"})) elif kind == "transcription" and msg.get("is_final"): if msg.get("transcript", "").strip(): self.turn_text.append(msg["transcript"].strip()) if self.turn_ended and self.turn_text: text, self.turn_text, self.turn_ended = " ".join(self.turn_text), [], False self.start_speaking(self.reply(text)) @app.websocket("/stream/{token}") async def stream(ws: WebSocket, token: str) -> None: call_info = redeem(token) if call_info is None: await ws.close(code=4401) return await ws.accept() call = Call(ws, call_info) async with websockets.connect(STT_URL, additional_headers=AUTH) as stt: async def bandwidth_to_pulse() -> None: while True: msg = json.loads(await ws.receive_text()) event = msg.get("eventType") if event == "start": call.start_speaking(call.speak(GREETING)) elif event == "media": await stt.send(base64.b64decode(msg["payload"])) # raw mu-law bytes elif event == "stop": return async def pulse_to_call() -> None: async for raw in stt: await call.on_transcript(json.loads(raw), stt) tasks = [asyncio.create_task(bandwidth_to_pulse()), asyncio.create_task(pulse_to_call())] try: await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) except WebSocketDisconnect: pass finally: for t in tasks + [call.speaking]: if t and not t.done(): t.cancel() with suppress(asyncio.CancelledError, Exception): await t ``` ### 4. Run it ```bash uvicorn server:app --host 0.0.0.0 --port 8000 ``` Call your Bandwidth number. The agent greets you, and you can talk to it. Interrupt it mid-sentence and it stops and listens. ## Outbound calls To have the agent place a call, create the call with the Bandwidth Voice API and point `answerUrl` at the same `/bxml` endpoint. When the person answers, Bandwidth requests BXML and the call joins the same bridge. Bandwidth API calls use OAuth 2.0 client credentials. Create an API credential (client ID and secret) in the Bandwidth App, then: **`outbound.py`** ```python outbound.py import base64 import os import httpx ACCOUNT_ID = os.environ["BW_ACCOUNT_ID"] APPLICATION_ID = os.environ["BW_APPLICATION_ID"] CLIENT_ID = os.environ["BW_CLIENT_ID"] CLIENT_SECRET = os.environ["BW_CLIENT_SECRET"] PUBLIC_HOST = os.environ["PUBLIC_HOST"] basic = base64.b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode() token = httpx.post( "https://api.bandwidth.com/api/v1/oauth2/token", headers={"Authorization": f"Basic {basic}"}, data={"grant_type": "client_credentials"}, ).json()["access_token"] # valid for 1 hour; refresh before it expires call = httpx.post( f"https://voice.bandwidth.com/api/v2/accounts/{ACCOUNT_ID}/calls", headers={"Authorization": f"Bearer {token}"}, json={ "from": "+15551230000", # a number on your Bandwidth account "to": "+15559870000", "applicationId": APPLICATION_ID, "answerUrl": f"https://{PUBLIC_HOST}/bxml", "username": os.environ["BW_CALLBACK_USERNAME"], "password": os.environ["BW_CALLBACK_PASSWORD"], }, ) call.raise_for_status() print(call.json()["callId"]) ``` The `username` and `password` fields make Bandwidth send Basic Auth on the answer webhook, so `/bxml` accepts it. ## Going to production * **Stream the reply.** The example waits for Electron's full answer before speaking. Use [streaming chat completions](/models/llm/streaming) and send sentences to Lightning as they arrive, with a shared `context_id` so prosody carries across fragments ([Continuations](/models/text-to-speech/continuations)). * **Reuse the TTS connection.** Open one Lightning WebSocket per call instead of one per reply to save a connection handshake on every turn. * **Pick your languages.** Set `SMALLEST_LANGUAGE` to a code Pulse and your chosen voice both support. For English-only agents, [Pulse 2.0](/model-cards/speech-to-text/pulse-2-0) adds built-in end-of-turn detection. * **Run close to your callers.** Lightning and Pulse serve from India and US West. Host the bridge in the same region as your inference pool to keep round trips short. * **Store tokens outside memory.** The in-memory token map works for a single process. Use Redis or similar if you run several bridge replicas. * **Hang up cleanly.** To end a call from the agent, send `{"state": "completed"}` to `POST https://voice.bandwidth.com/api/v2/accounts/{accountId}/calls/{callId}` with your Bearer token. The `accountId` and `callId` are stored on each `Call` from the webhook. ## Troubleshooting | Symptom | Likely cause | Fix | | ------------------------------------- | ----------------------------------------------- | ------------------------------------------------------------------------------------------------------------- | | Call connects, then drops immediately | Bandwidth could not reach `/bxml`, or got a 401 | Check the callback URL, and that the callback username and password match `.env` | | `/stream` closes with code 4401 | Token expired or reused | Bandwidth must connect within 30 seconds of the webhook. Check `PUBLIC_HOST` matches your public host exactly | | Agent never hears the caller | Media WebSocket not reaching the bridge | Confirm your host accepts `wss://` upgrades (ngrok does by default) | | Transcripts are nonsense | Wrong encoding or rate | Keep `encoding=mulaw` and `sample_rate=8000` on the Pulse URL | | Agent audio is noise or too fast | TTS format mismatch | Keep `output_format: "ulaw"` and `sample_rate: 8000`, and send `contentType: "audio/pcmu"` | | Agent talks over the caller | VAD events off | Keep `vad_events=true` on the Pulse URL so `speech_started` triggers `clear` | ## Reference * [Pulse STT WebSocket quickstart](/models/speech-to-text/realtime-web-socket/quickstart) * [Pulse VAD events](/models/speech-to-text/features/vad-events) and [endpointing](/models/speech-to-text/features/endpointing) * [Lightning TTS streaming](/models/text-to-speech/streaming) * [Electron quickstart](/models/llm/quickstart) * [Bandwidth Voice API](https://dev.bandwidth.com/apis/voice) and [BXML ``](https://dev.bandwidth.com/docs/voice/programmable-voice/bxml/startStream/)