Vapi

pip install gradium

export GRADIUM_API_KEY=gd_your_api_key_here

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

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

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

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()

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

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

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

@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