> ## Documentation Index > Fetch the complete documentation index at: https://docs.gradium.ai/llms.txt > Use this file to discover all available pages before exploring further. # Vapi > Use Gradium speech models in Vapi voice agents 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 WebSocket TTS setup and streaming behavior. Gradium WebSocket STT, VAD, and flushing. Tune semantic VAD horizons and thresholds. Sign up for Vapi and create an assistant. You need a [Vapi account](https://dashboard.vapi.ai) 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 ```bash theme={null} pip install gradium ``` Set your Gradium API key in the environment: ```bash theme={null} 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: ```json theme={null} { "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. ```python theme={null} 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](/guides/limits) for the full format list. Point the assistant's `voice` at the endpoint: ```json theme={null} { "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. ```python theme={null} 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](/guides/recipes/turn-taking) for tuning the horizon and threshold. Point the assistant's `transcriber` at the bridge over `wss`: ```json theme={null} { "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: ```python theme={null} 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_` using Vapi's `message.sampleRate`. | | `pronunciation_id` | TTS | Optional pronunciation dictionary ID. | | `input_format` | STT | Set to `pcm_` 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](/guides/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: ```json theme={null} { "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: ```python theme={null} @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](/guides/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](/guides/voices/overview). * **Domain vocabulary**: improve recognition of menu items, brand names, and product terms with [Keyword Boosting](/guides/recipes/keyword-boosting) and [Text Rewriting Rules](/guides/text-rewriting).