Telephony: Bandwidth

"""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"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
  <StartStream name="smallest" mode="bidirectional" tracks="inbound" destination="wss://{PUBLIC_HOST}/stream/{token}" />
  <StopStream name="smallest" wait="true" />
</Response>"""
    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

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"])