Vapi

Vapi is a platform for building and deploying voice agents, handling telephony, web transport, LLM calls, and turn-taking. Gradium is not a built-in Vapi provider, so you connect it through Vapi’s custom endpoints: a custom-voice webhook for speech output and a custom-transcriber WebSocket for speech input. You run a small bridge service, and Vapi keeps managing the rest of the call.

Gradium TTS guide

Gradium WebSocket TTS setup and streaming behavior.

Gradium STT guide

Gradium WebSocket STT, VAD, and flushing.

Turn-taking

Tune semantic VAD horizons and thresholds.

Vapi dashboard

You need a Vapi account and a Gradium API key. This page is the reference for the Gradium side of the integration: the endpoint contracts, the bridge code, and the tuning options.

How Gradium connects

Vapi endpoint	Transport	What your bridge does
`custom-voice`	HTTP POST	Vapi posts the text to speak; you stream raw PCM back from `tts_realtime`.
`custom-transcriber`	WebSocket	Vapi streams call audio; you return transcripts from `stt_realtime`.

You can use either side on its own. Keeping Vapi’s built-in transcriber while using Gradium for speech output is a valid setup, and vice versa.

Install

pip install gradium

Set your Gradium API key in the environment:

export GRADIUM_API_KEY=gd_your_api_key_here

Text to speech

Vapi posts one request per sentence, carrying the text and the sample rate it expects:

{
  "message": {
    "type": "voice-request",
    "text": "Your table is booked for seven o'clock.",
    "sampleRate": 24000
  }
}

Read message.sampleRate and pass it straight through as Gradium’s output_format. Vapi expects raw 16-bit little-endian mono PCM at exactly the rate it asked for, streamed as it is produced.

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from gradium.client import GradiumClient

app = FastAPI()
client = GradiumClient(api_key=GRADIUM_API_KEY)

@app.post("/vapi/tts")
async def vapi_tts(request: Request) -> StreamingResponse:
    message = (await request.json())["message"]
    text, rate = message["text"], message["sampleRate"]

    async def pcm():
        async with client.tts_realtime(
            voice_id="YTpq7expH9539ERJ",
            output_format=f"pcm_{rate}",
        ) as tts:
            await tts.send_text(text)
            await tts.send_eos()
            async for msg in tts:
                if msg["type"] == "audio":
                    yield msg["audio"]
                elif msg["type"] == "end_of_stream":
                    break

    return StreamingResponse(pcm(), media_type="application/octet-stream")

Every sample rate Vapi requests has a matching Gradium PCM format, so no resampling is needed. See Limits for the full format list.

Point the assistant’s voice at the endpoint:

{
  "voice": {
    "provider": "custom-voice",
    "server": { "url": "https://your-server.com/vapi/tts" }
  }
}

Speech to text

Vapi opens a WebSocket, sends a JSON start frame, then streams interleaved 16-bit PCM. When channels is 2, channel 0 carries the caller and channel 1 carries the assistant — forward only channel 0, or the agent transcribes its own speech.

import asyncio, json
from fastapi import FastAPI, WebSocket
from gradium.client import GradiumClient

app = FastAPI()
client = GradiumClient(api_key=GRADIUM_API_KEY)

def caller_channel(pcm: bytes, channels: int) -> bytes:
    """Keep channel 0 from interleaved 16-bit PCM."""
    if channels == 1:
        return pcm
    stride = channels * 2
    return b"".join(pcm[i:i + 2] for i in range(0, len(pcm) - stride + 1, stride))

@app.websocket("/vapi/transcriber")
async def vapi_transcriber(ws: WebSocket) -> None:
    await ws.accept()
    start = json.loads(await ws.receive_text())
    channels = int(start.get("channels", 2))
    transcript, high_vad_steps = [], 0

    async def send(text: str, transcript_type: str) -> None:
        await ws.send_text(json.dumps({
            "type": "transcriber-response",
            "transcription": text,
            "channel": "customer",
            "transcriptType": transcript_type,
        }))

    async with client.stt_realtime(
        input_format=f"pcm_{start['sampleRate']}",
        json_config={"language": "en", "delay_in_frames": 16},
    ) as stt:

        async def to_gradium() -> None:
            while True:
                pcm = await ws.receive_bytes()
                await stt.send_audio(caller_channel(pcm, channels))

        async def to_vapi() -> None:
            nonlocal transcript, high_vad_steps
            async for msg in stt:
                if msg["type"] == "text":
                    transcript.append(msg["text"])
                    await send(" ".join(transcript).strip(), "partial")

                elif msg["type"] == "step":
                    inactivity = msg["vad"][-1]["inactivity_prob"]
                    high_vad_steps = high_vad_steps + 1 if inactivity > 0.5 else 0
                    if high_vad_steps >= 3 and transcript:
                        await stt.send_flush(flush_id=1)

                elif msg["type"] == "flushed":
                    await send(" ".join(transcript).strip(), "final")
                    transcript, high_vad_steps = [], 0

        # Stop as soon as either side ends, so the Gradium session and the
        # audio pump are torn down when the caller hangs up.
        tasks = [asyncio.create_task(to_gradium()),
                 asyncio.create_task(to_vapi())]
        try:
            await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        finally:
            for task in tasks:
                task.cancel()

Send partial transcripts as well as final ones — Vapi needs to see speech while it is happening for barge-in to work. Treat flushed as the end of the turn rather than finalizing the moment the VAD threshold is crossed, because Gradium is still holding audio in its decoding window. See Turn-Taking with Semantic VAD for tuning the horizon and threshold. Point the assistant’s transcriber at the bridge over wss:

{
  "transcriber": {
    "provider": "custom-transcriber",
    "server": { "url": "wss://your-server.com/vapi/transcriber" }
  }
}

Secure the endpoints

Both endpoints are public HTTP surfaces. Set a secret on the server object in the assistant config, and Vapi sends it as an X-Vapi-Secret header on every request. Reject anything that does not match:

if request.headers.get("x-vapi-secret") != VAPI_WEBHOOK_SECRET:
    raise HTTPException(status_code=401, detail="bad secret")

Configuration

Setting	Applies to	Description
`api_key`	STT, TTS	Gradium API key. Defaults to `GRADIUM_API_KEY`.
`model_name`	STT, TTS	Gradium model name. Defaults to `default`.
`voice_id`	TTS	Gradium voice ID for synthesized replies.
`output_format`	TTS	Set to `pcm_<rate>` using Vapi’s `message.sampleRate`.
`pronunciation_id`	TTS	Optional pronunciation dictionary ID.
`input_format`	STT	Set to `pcm_<rate>` using the `start` frame’s `sampleRate`.
`json_config.language`	STT	Transcription language.
`json_config.delay_in_frames`	STT	Context before text is emitted. Each frame is 80 ms.
`json_config.keywords`	STT	Bias recognition toward names and domain vocabulary.

Reduce latency with multiplexing

Vapi posts one custom-voice request per sentence, so a fresh WebSocket per request pays connection setup several times per agent reply. Keep one connection open per sample rate and route concurrent requests over it with close_ws_on_eos: false and a per-request client_req_id. Recycle pooled connections before the session limit, and fall back to a single-use connection if a pooled socket closes before it produces audio. See Multiplexing for the full contract.

Let Gradium decide when the turn ends

By default Vapi runs its own endpointing on your partial transcripts, which can race Gradium’s VAD and make the assistant reply before the flushed final arrives. Hand the decision to Gradium with a custom endpointing model:

{
  "startSpeakingPlan": {
    "waitSeconds": 0.25,
    "smartEndpointingPlan": {
      "provider": "custom-endpointing-model",
      "server": { "url": "https://your-server.com/vapi/endpointing", "timeoutSeconds": 3 }
    }
  }
}

Vapi then posts a call.endpointing.request on every transcript update, and your server answers with how long to keep waiting based on the live Gradium session:

@app.post("/vapi/endpointing")
async def endpointing(body: dict) -> dict:
    if body.get("message", {}).get("type") != "call.endpointing.request":
        return {"ok": True}

    state = gradium_turn_state()          # your live STT session

    if not state["active"]:
        return {"timeoutSeconds": 1.0}    # no session, keep the call moving
    if state["seconds_since_final"] < 2.5:
        return {"timeoutSeconds": 0.05}   # Gradium flushed, the turn is over
    if state["inactivity_prob"] < 0.5:
        return {"timeoutSeconds": 5.0}    # caller is audibly mid-sentence
    return {"timeoutSeconds": 3.0}        # a pause; the final should arrive

Clear your reference to the Gradium session when a call ends. If a finished session stays registered, the first endpointing requests of the next call are answered from stale turn state.

Long calls

A single Gradium session lasts up to 300 seconds. Reconnect the Gradium session transparently when one ends while keeping Vapi’s socket open, so calls of any length keep transcribing. See Limits.

When to use Vapi with Gradium

Telephony agents: let Vapi manage phone numbers, SIP, and call routing while Gradium handles speech.
Swapping speech providers: keep an existing Vapi assistant, LLM, and tools, and change only the voice or transcriber layer.
Custom voices: use an instantly cloned Gradium voice in a Vapi agent. See Voices.
Domain vocabulary: improve recognition of menu items, brand names, and product terms with Keyword Boosting and Text Rewriting Rules.

Introduction

Agent Frameworks

Tools

Gradium TTS guide

Gradium STT guide

Turn-taking

Vapi dashboard

How Gradium connects

Install

Text to speech

Speech to text

Secure the endpoints

Configuration

Reduce latency with multiplexing

Let Gradium decide when the turn ends

Long calls

When to use Vapi with Gradium

Gradium TTS guide

Gradium STT guide

Turn-taking

Vapi dashboard

​How Gradium connects

​Install

​Text to speech

​Speech to text

​Secure the endpoints

​Configuration

​Reduce latency with multiplexing

​Let Gradium decide when the turn ends

​Long calls

​When to use Vapi with Gradium

How Gradium connects

Install

Text to speech

Speech to text

Secure the endpoints

Configuration

Reduce latency with multiplexing

Let Gradium decide when the turn ends

Long calls

When to use Vapi with Gradium