Cartesia Integration

This guide will walk you through building a real-time voice AI agent on Bandwidth's Voice Network using Cartesia, transcribing callers with Ink 2 speech-to-text and replying with Sonic text-to-speech. You already have Bandwidth Voice; Cartesia is the add-on that gives your calls a voice.

Rather than a framework, this integration is a small FastAPI bridge you run yourself: it answers Bandwidth's voice webhook with BXML, accepts Bandwidth's bidirectional media WebSocket, and forwards audio to and from Cartesia's STT and TTS sockets. No vendor SDKs — just fastapi, uvicorn, httpx, websockets, and python-dotenv — and no resampling, since Bandwidth carries calls as 8 kHz μ-law and both Cartesia sockets speak pcm_mulaw at 8000 Hz natively.

info

This integration uses Bandwidth Programmable Voice with Media Streaming. Your application receives μ-law audio frames directly over a bidirectional WebSocket and feeds them into Cartesia's STT and TTS sockets — no SIP-to-SIP routing required.

What you'll need

• A Bandwidth Programmable Voice account with:
  • A purchased phone number assigned to a Voice Application
  • API credentials (OAuth 2.0 client ID + secret)
  • Don't have an account yet? Try Bandwidth Build for free — get a real phone number and 3000 credits to start building immediately.
  • If you have a full Bandwidth App account but haven't set it up yet, check out our Account Setup guide.
• A Cartesia API key (looks like sk_car_...) and a voice ID from the voice library
• Python 3.11+
• uv (or pip if you prefer)
• A publicly accessible URL for your application (e.g., using ngrok)

Call Flow

Before we dive in, let's walk through what an inbound call flow looks like with this integration.

This flow demonstrates a basic AI agent answering an inbound call and echoing back what it hears. Let's break it down:

1. A user calls your Bandwidth number.
2. Bandwidth POSTs a Basic-Auth-protected webhook to /bxml with the inbound call event, including callId and accountId.
3. Your application mints a one-time correlation token bound to those server-trusted IDs and responds with a <StartStream name="cartesia_agent"> BXML pointing at wss://<your-host>/stream/{token}, followed by a <StopStream name="cartesia_agent" wait="true"/> that keeps the call leg alive while the WebSocket session runs.
4. Bandwidth opens a WebSocket to your application. Your app validates the token and recovers the trusted callId/accountId before accepting the stream.
5. Caller audio arrives as base64 μ-law media events. Your app decodes them and forwards raw μ-law straight to Cartesia's Ink 2 STT socket — no resampling needed.
6. Each finalized transcript is turned into reply text (an echo in this demo — swap in your own LLM call here) and sent to Cartesia's Sonic TTS socket, which streams back μ-law audio chunks.
7. Each chunk is wrapped in a playAudio event and sent back over the same WebSocket. The caller hears the reply.
8. When the caller hangs up (or Bandwidth sends a stop event), your app ends the call via the Bandwidth Voice API using the trusted callId/accountId — never an ID read off the WebSocket itself.

Why not trust the WebSocket's own metadata?

Cartesia's published community sample reads callId and accountId straight out of the WebSocket's first start event and uses them to hang up the call. That event isn't authenticated — anyone who discovers your /stream endpoint could open a connection and feed it an arbitrary callId, triggering a hang-up (or worse) against a live call in your account.

This guide closes that gap the same way the Pipecat integration guide does: trust only the authenticated /bxml webhook body for callId and accountId, and bind them to the WebSocket via a short-lived, server-issued correlation token. The WebSocket's start event is used only to confirm the stream came up — never as a source of trusted IDs.

Let's Build It!

The sections below walk through a complete server.py, adapted from Cartesia's Bandwidth community guide and hardened with the Basic Auth + correlation-token pattern above. Cartesia's own sample skips authentication entirely for brevity — fine for a five-minute demo on your own machine, risky the moment your ngrok URL is reachable by anyone else. We add it back in here.

Set Up Your Environment

python -m venv .venv && source .venv/bin/activate
pip install fastapi 'uvicorn[standard]' httpx websockets python-dotenv

Create .env:

# Bandwidth OAuth 2.0 credentials, used to hang up the call via the Voice API
BANDWIDTH_CLIENT_ID=
BANDWIDTH_CLIENT_SECRET=

# Webhook Basic Auth. Set the same username/password on your Bandwidth
# Voice Application's inbound callback.
BANDWIDTH_WEBHOOK_USERNAME=
BANDWIDTH_WEBHOOK_PASSWORD=

# Cartesia
CARTESIA_API_KEY=
CARTESIA_VOICE_ID=

# Public hostname for the BXML StartStream destination (https://...)
PUBLIC_URL=
GREETING=Hi! Say something and I will read it back to you.

Note there's no BANDWIDTH_ACCOUNT_ID or BANDWIDTH_APPLICATION_ID to configure — the trusted callId/accountId for each call come from the authenticated /bxml webhook body, not from a static config value.

Imports and Cartesia Constants

import asyncio
import base64
import json
import os
import secrets
import time
import uuid
from contextlib import suppress

import httpx
import websockets
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect, status
from fastapi.responses import Response

load_dotenv()

PUBLIC_URL = os.environ["PUBLIC_URL"]
GREETING = os.environ["GREETING"]
CARTESIA_API_KEY = os.environ["CARTESIA_API_KEY"]
CARTESIA_VOICE_ID = os.environ["CARTESIA_VOICE_ID"]

BANDWIDTH_WEBHOOK_USERNAME = os.environ["BANDWIDTH_WEBHOOK_USERNAME"]
BANDWIDTH_WEBHOOK_PASSWORD = os.environ["BANDWIDTH_WEBHOOK_PASSWORD"]

# Cartesia pins its WebSocket protocol to a dated version string.
CARTESIA_VERSION = "2026-03-01"
TTS_URL = f"wss://api.cartesia.ai/tts/websocket?cartesia_version={CARTESIA_VERSION}"
# Ink 2 reads the call's native 8 kHz mu-law directly, so caller audio needs no resampling.
STT_URL = (
    "wss://api.cartesia.ai/stt/websocket"
    f"?model=ink-2&cartesia_version={CARTESIA_VERSION}"
    "&encoding=pcm_mulaw&sample_rate=8000&language=en"
)

BANDWIDTH_VOICE_BASE = "https://voice.bandwidth.com/api/v2"
BANDWIDTH_OAUTH_URL = "https://api.bandwidth.com/api/v1/oauth2/token"

app = FastAPI()

A few things to note:

1. CARTESIA_VERSION pins the Cartesia WebSocket protocol version (2026-03-01). Bump this deliberately when you upgrade, rather than floating on whatever is current.
2. Both STT_URL and TTS_URL request encoding=pcm_mulaw / sample_rate=8000 (via output_format for TTS) — matching Bandwidth's native μ-law format exactly, so audio crosses the bridge byte-for-byte with no resampling in either direction.
3. ink-2 is Cartesia's current streaming STT model; sonic-3.5 (used below in the TTS request) is the current Sonic snapshot. Pin a dated snapshot like sonic-3.5-2026-05-04 for production stability rather than floating on the alias.

Authentication and Correlation Tokens

These three helpers are the security spine of the bridge, so define them before the handlers that use them. _verify_webhook_auth rejects any /bxml request that doesn't present your Basic Auth credentials. _issue_token and _consume_token implement the one-time correlation token: /bxml issues a token bound to the webhook's trusted callId/accountId, and the WebSocket handler redeems it exactly once, so the IDs can never be spoofed by whatever connects to the socket.

# Correlation tokens: token -> (call_id, account_id, expires_at_monotonic).
# In-memory is fine for a single-process demo; use a shared store (e.g. Redis)
# if you run more than one worker.
_TOKENS: dict[str, tuple[str, str, float]] = {}
TOKEN_TTL_SECONDS = 30


def _verify_webhook_auth(request: Request) -> None:
    """Enforce HTTP Basic Auth on the inbound webhook; raise 401 on mismatch."""
    header = request.headers.get("Authorization", "")
    scheme, _, encoded = header.partition(" ")
    if scheme.lower() == "basic":
        with suppress(ValueError):
            username, _, password = base64.b64decode(encoded).decode().partition(":")
            user_ok = secrets.compare_digest(username, BANDWIDTH_WEBHOOK_USERNAME)
            pass_ok = secrets.compare_digest(password, BANDWIDTH_WEBHOOK_PASSWORD)
            if user_ok and pass_ok:
                return
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid or missing webhook credentials",
        headers={"WWW-Authenticate": "Basic"},
    )


async def _issue_token(call_id: str, account_id: str) -> str:
    """Mint a single-use, URL-safe token bound to server-trusted call IDs."""
    token = secrets.token_urlsafe(32)
    _TOKENS[token] = (call_id, account_id, time.monotonic() + TOKEN_TTL_SECONDS)
    return token


async def _consume_token(token: str) -> tuple[str, str] | None:
    """Redeem a token exactly once. Returns (call_id, account_id) or None."""
    entry = _TOKENS.pop(token, None)
    if entry is None:
        return None
    call_id, account_id, expires_at = entry
    if time.monotonic() > expires_at:
        return None
    return call_id, account_id

Handle the Inbound Voice Webhook

When a caller dials your Bandwidth number, Bandwidth POSTs to /bxml. The handler authenticates the request, extracts the trusted callId / accountId from the body, mints a one-time correlation token, and returns a <StartStream> BXML with a keepalive verb.

@app.post("/bxml")
async def bxml(request: Request) -> Response:
    """Bandwidth voice webhook. Basic Auth required; trust the body's IDs."""
    _verify_webhook_auth(request)

    body = await request.json()
    call_id = body.get("callId")
    account_id = body.get("accountId")
    if not call_id or not account_id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Webhook body missing callId or accountId",
        )

    token = await _issue_token(str(call_id), str(account_id))

    ws_url = PUBLIC_URL.replace("https://", "wss://") + f"/stream/{token}"
    bxml_body = (
        '<?xml version="1.0" encoding="UTF-8"?>'
        "<Response>"
        f'<StartStream name="cartesia_agent" destination="{ws_url}" mode="bidirectional" tracks="inbound"/>'
        '<StopStream name="cartesia_agent" wait="true"/>'
        "</Response>"
    )
    return Response(content=bxml_body, media_type="application/xml")

A few things to note:

1. _verify_webhook_auth enforces HTTP Basic Auth using BANDWIDTH_WEBHOOK_USERNAME / BANDWIDTH_WEBHOOK_PASSWORD, rejecting the request with 401 if the credentials don't match.
2. callId and accountId are read only from the authenticated webhook body — never from anything the WebSocket sends later.
3. _issue_token mints a short-lived, URL-safe token and stores (call_id, account_id, expires_at) server-side keyed by that token, the same pattern used by the Pipecat integration guide.
4. <StartStream> is non-blocking — BXML execution continues immediately after the stream starts, and the call would end (closing the WebSocket) if there were no verb left to run. <StopStream name="cartesia_agent" wait="true"/> holds the call open until your WebSocket handler closes it, with no artificial time cap. (A <Pause duration="600"/> works too, but caps the conversation at a fixed length — see the <Pause> and <StartStream> references for the trade-off. Note the attribute is duration, not length.)

Accept the WebSocket and Bridge to Cartesia

Bandwidth opens a WebSocket to /stream/{token}. The handler validates the token, recovers the trusted IDs, greets the caller, and spins up the two tasks that pump audio to and from Cartesia.

@app.websocket("/stream/{token}")
async def stream(ws: WebSocket, token: str) -> None:
    """Validate the correlation token, then bridge audio to/from Cartesia."""
    trusted = await _consume_token(token)
    if trusted is None:
        await ws.close(code=1008)  # 1008 = policy violation
        return
    call_id, account_id = trusted
    await ws.accept()

    # Bandwidth's first frame is a "start" event. We only use it to confirm the
    # stream came up cleanly -- callId/accountId come from the trusted token
    # mapping above, never from this (unauthenticated) WebSocket payload.
    start_event = json.loads(await ws.receive_text())
    if start_event.get("eventType") != "start":
        await ws.close(code=4400)
        return

    stt = await websockets.connect(STT_URL, additional_headers={"X-API-Key": CARTESIA_API_KEY})

    await speak(ws, GREETING)
    # One task pumps caller audio into Ink 2; the other speaks its transcripts back.
    pump = asyncio.create_task(_caller_audio_to_stt(ws, stt))
    replies = asyncio.create_task(_transcripts_to_replies(ws, stt))

    try:
        await pump  # returns on Bandwidth's "stop" event or a disconnect
    finally:
        replies.cancel()
        with suppress(asyncio.CancelledError):
            await replies
        await stt.close()
        await _hang_up(account_id, call_id)

Key points:

1. _consume_token pops the token from server-side state — single-use, short TTL. Invalid or expired tokens get rejected with WS close code 1008.
2. The start event's presence is checked, but its metadata.callId / metadata.accountId are deliberately ignored — the trusted values came from the token mapping instead.
3. Both Cartesia sockets authenticate with additional_headers={"X-API-Key": CARTESIA_API_KEY}.
4. speak(ws, GREETING) opens a fresh Sonic session before the caller has said anything, so the agent greets first.

Pump Caller Audio to Cartesia STT

async def _caller_audio_to_stt(ws: WebSocket, stt) -> None:
    # Bandwidth sends each media event as base64 mu-law in JSON; Ink 2 wants raw
    # binary, so decode before forwarding.
    try:
        async for raw in ws.iter_text():
            event = json.loads(raw)
            kind = event.get("eventType")
            if kind == "media":
                await stt.send(base64.b64decode(event["payload"]))
            elif kind == "stop":
                break
    except WebSocketDisconnect:
        pass
    finally:
        # Flush Ink 2's buffered audio and close its session cleanly.
        with suppress(Exception):
            await stt.send("finalize")
            await stt.send("close")

Every media event Bandwidth sends over the WebSocket carries base64-encoded μ-law audio for the inbound track. This decodes it and forwards the raw bytes straight to the open Ink 2 socket — no resampling, since both sides already agree on pcm_mulaw at 8000 Hz.

Turn Transcripts into Replies

async def _transcripts_to_replies(ws: WebSocket, stt) -> None:
    async for raw in stt:
        msg = json.loads(raw)
        if msg.get("type") == "transcript" and msg.get("is_final") and msg.get("text"):
            # Replace this echo with your own LLM / agent call to build a real bot.
            await speak(ws, f"You said: {msg['text']}")
        elif msg.get("type") == "error":
            raise RuntimeError(f"Cartesia STT error: {msg}")

This is the seam for your own logic. The demo above echoes back each finalized transcript; a real agent would route msg["text"] through an LLM (with conversation history, tools, whatever your use case needs) and pass the model's reply to speak() instead of the echoed string.

Synthesize the Reply with Cartesia Sonic

async def speak(ws: WebSocket, text: str) -> None:
    # Open a Sonic socket, request synthesis, and forward each chunk to Bandwidth.
    async with websockets.connect(TTS_URL, additional_headers={"X-API-Key": CARTESIA_API_KEY}) as tts:
        await tts.send(json.dumps({
            "context_id": str(uuid.uuid4()),  # required; groups one synthesis request
            "model_id": "sonic-3.5",
            "voice": {"mode": "id", "id": CARTESIA_VOICE_ID},
            "transcript": text,
            "output_format": {"container": "raw", "encoding": "pcm_mulaw", "sample_rate": 8000},
        }))
        async for raw in tts:
            msg = json.loads(raw)
            if msg.get("type") == "chunk":
                # Sonic's mu-law bytes are wire-compatible with Bandwidth's audio/pcmu.
                await ws.send_text(json.dumps({
                    "eventType": "playAudio",
                    "media": {"contentType": "audio/pcmu", "payload": msg["data"]},
                }))
            elif msg.get("type") == "done":
                return
            elif msg.get("type") == "error":
                raise RuntimeError(f"Cartesia TTS error: {msg}")

Each call opens its own Sonic session scoped to a fresh context_id. Sonic streams back base64 μ-law chunks, which are wrapped in a Bandwidth playAudio event (contentType: "audio/pcmu") and written straight back over the caller's WebSocket — again, byte-for-byte, no re-encoding.

Hang Up Cleanly

async def _bandwidth_token(client: httpx.AsyncClient) -> str:
    resp = await client.post(
        BANDWIDTH_OAUTH_URL,
        auth=(os.environ["BANDWIDTH_CLIENT_ID"], os.environ["BANDWIDTH_CLIENT_SECRET"]),
        data={"grant_type": "client_credentials"},
    )
    resp.raise_for_status()
    return resp.json()["access_token"]


async def _hang_up(account_id: str, call_id: str) -> None:
    async with httpx.AsyncClient(timeout=10.0) as client:
        token = await _bandwidth_token(client)
        resp = await client.post(
            f"{BANDWIDTH_VOICE_BASE}/accounts/{account_id}/calls/{call_id}",
            headers={"Authorization": f"Bearer {token}"},
            json={"state": "completed"},
        )
        if resp.status_code not in (200, 404):  # 404 = call already ended
            resp.raise_for_status()

_hang_up runs in the finally block of the WebSocket handler above, using the trusted account_id / call_id recovered from the correlation token — not anything read from the WebSocket. It fetches a fresh OAuth token via client_credentials and POSTs {"state": "completed"} to end the call. A 404 means the call already ended (e.g., the caller hung up first), which is treated as success.

Connect to a Public URL

Bandwidth needs a publicly reachable HTTPS host to deliver the inbound webhook and open the media-stream WebSocket. Start the server and expose it in a second terminal:

uvicorn server:app --host 0.0.0.0 --port 8000

ngrok http 8000

Copy ngrok's HTTPS URL into .env as PUBLIC_URL and restart the server so the BXML response references the correct destination.

Configure your Bandwidth Voice Application

Finally, point your Bandwidth Voice Application at your public webhook URL.

1. Log in to the Bandwidth App and open your Voice Application (or create a new one).
2. Under Call initiated, select POST as the callback method and set the Callback URL to https://<your-public-host>/bxml.
3. Tick Use a callback username and password and enter the same BANDWIDTH_WEBHOOK_USERNAME / BANDWIDTH_WEBHOOK_PASSWORD values you put in .env. Bandwidth will send these in a Basic Auth header on every inbound call event, and /bxml rejects the request without them.
4. Click Save.
5. Make sure the application is linked to a Voice Configuration Package and that the package is assigned to your Bandwidth phone number.

Test the Integration

Call your Bandwidth phone number. You should hear the agent greet you (GREETING), and when you speak, it reads your words back to you (You said: ...). When you hang up — or the app decides the conversation is over — the WebSocket handler tears down the Cartesia sockets and ends the call leg via the Bandwidth Voice API using the trusted callId.

Configuration

The Cartesia-facing knobs live in the generation request and the STT URL:

Parameter	Where	Value used	Notes
model_id	TTS request	sonic-3.5	Pin a dated Sonic snapshot (e.g. sonic-3.5-2026-05-04) for production stability
voice	TTS request	{"mode":"id","id":...}	Any voice ID from the voice library
output_format	TTS request	pcm_mulaw @ 8000	Matches Bandwidth's audio/pcmu; no conversion needed
model	STT URL	ink-2	Cartesia's latest streaming STT model
encoding / sample_rate	STT URL	pcm_mulaw / 8000	Matches the call's native format
cartesia_version	Both URLs	2026-03-01	Dated protocol version pin — bump deliberately on upgrade

The Bandwidth-facing knobs are the BANDWIDTH_WEBHOOK_USERNAME / BANDWIDTH_WEBHOOK_PASSWORD pair (must match the Voice Application's callback credentials) and the correlation token's TTL (short-lived by design — a token is only needed for the few seconds between the /bxml response and the WebSocket connecting).

What's next

• Plug in an LLM. The echo in _transcripts_to_replies is the seam — route each finalized transcript through your own agent and synthesize its reply with the same speak() call.
• Cleaner turn-taking. The manual STT socket emits incremental is_final segments, so this demo replies per fragment. Cartesia's turn-detection endpoint (/stt/turns/websocket) lets you reply once per completed utterance instead.
• Higher-fidelity TTS. Bandwidth's playAudio also accepts audio/pcm;rate=16000 and rate=24000 (mono, 16-bit, little-endian). Set Sonic's output_format to pcm_s16le at the matching rate; Bandwidth resamples to 8 kHz μ-law once at its edge instead of after a lossy round-trip.
• Barge-in. Send {"eventType": "clear"} on the media WebSocket to drop queued outbound audio when the caller talks over the agent.
• Use a framework. The Pipecat integration guide wraps this same protocol as a BandwidthFrameSerializer, with STT/LLM/TTS pipeline plumbing (VAD, interruption handling, tool calls) already built in — useful once you outgrow a hand-rolled echo bot.
• Harden further. Verify Bandwidth's webhook signature in addition to Basic Auth, IP-allowlist Bandwidth's egress ranges, and back the token store with Redis if you run more than one worker.

Resources

• Cartesia's Bandwidth community integration guide
• Bandwidth <StartStream> (BXML reference)
• Bandwidth <StopStream> (BXML reference)
• Cartesia STT WebSocket reference
• Cartesia TTS WebSocket reference