"""Khabeer (5abear) bridge for Vapi & Ultravox — custom Saudi voice + transcriber. Vapi lets you plug a custom voice ("custom-voice" provider) and a custom transcriber into any assistant. Both are plain HTTP endpoints that you host. This FastAPI app exposes them backed by Khabeer's Saudi voice API. pip install fastapi uvicorn httpx curl -O https://speak-saudi-ai.lovable.app/integrations/khabeer_vapi_server.py export KHABEER_API_KEY=5abear_sk_xxx export VAPI_SECRET=some-shared-secret uvicorn khabeer_vapi_server:app --host 0.0.0.0 --port 8000 Then in the Vapi assistant config: "voice": { "provider": "custom-voice", "server": { "url": "https://your-host/vapi/voice", "secret": "some-shared-secret" } }, "transcriber": { "provider": "custom-transcriber", "server": { "url": "https://your-host/vapi/transcriber", "secret": "some-shared-secret" } } Ultravox: point an external TTS/agent tool at POST /ultravox/tts (returns raw audio/wav) and at POST /ultravox/stt (returns {"text": ...}). """ from __future__ import annotations import base64 import json import os import struct from typing import Optional import httpx from fastapi import FastAPI, Header, HTTPException, Request, UploadFile, File from fastapi.responses import JSONResponse, Response, StreamingResponse API_BASE = os.environ.get("KHABEER_API_BASE", "https://speak-saudi-ai.lovable.app/api/public/v1") KHABEER_API_KEY = os.environ.get("KHABEER_API_KEY", "") VAPI_SECRET = os.environ.get("VAPI_SECRET", "") DEFAULT_VOICE = os.environ.get("KHABEER_VOICE", "noura") DEFAULT_ACCENT = os.environ.get("KHABEER_ACCENT", "aam") app = FastAPI(title="Khabeer bridge", version="1.0") def _auth_headers() -> dict: if not KHABEER_API_KEY: raise HTTPException(500, "KHABEER_API_KEY is not set") return {"Authorization": f"Bearer {KHABEER_API_KEY}"} def _check_secret(secret: Optional[str]) -> None: if VAPI_SECRET and secret != VAPI_SECRET: raise HTTPException(401, "invalid secret") def _wav_to_pcm(data: bytes) -> bytes: idx = data.find(b"data") return data[idx + 8 :] if idx != -1 else data def _pcm_to_wav(pcm: bytes, sample_rate: int = 24000) -> bytes: header = b"RIFF" + struct.pack(" bytes: """Returns raw 16-bit PCM at 24kHz.""" async with httpx.AsyncClient(timeout=60) as client: res = await client.post( f"{API_BASE}/tts", headers=_auth_headers(), json={"text": text, "voice": voice, "accent": accent, "speed": speed}, ) if res.status_code != 200: raise HTTPException(res.status_code, res.text) return _wav_to_pcm(base64.b64decode(res.json()["audio"])) async def _stream_synthesis(text: str, voice: str, accent: str, speed: float): """Yields raw 24 kHz PCM deltas as Khabeer produces them.""" async with httpx.AsyncClient(timeout=60) as client: async with client.stream( "POST", f"{API_BASE}/tts-stream", headers=_auth_headers(), json={"text": text, "voice": voice, "accent": accent, "speed": speed}, ) as res: if res.status_code != 200: detail = await res.aread() raise HTTPException(res.status_code, detail.decode(errors="ignore")) event = None async for line in res.aiter_lines(): if line.startswith("event: "): event = line[7:] elif line.startswith("data: ") and event == "audio_delta": payload = json.loads(line[6:]) if payload.get("audio"): yield base64.b64decode(payload["audio"]) async def _transcribe(audio_wav: bytes, accent: str = DEFAULT_ACCENT) -> dict: async with httpx.AsyncClient(timeout=120) as client: res = await client.post( f"{API_BASE}/stt", headers=_auth_headers(), files={"file": ("audio.wav", audio_wav, "audio/wav")}, data={"accent": accent}, ) if res.status_code != 200: raise HTTPException(res.status_code, res.text) return res.json() # ------------------------------- Vapi: voice ------------------------------- @app.post("/vapi/voice") async def vapi_voice(request: Request, x_vapi_secret: Optional[str] = Header(default=None)): """Vapi custom-voice endpoint: returns a raw PCM stream Vapi can play.""" _check_secret(x_vapi_secret) body = await request.json() message = body.get("message", body) text = message.get("text") or message.get("message") or "" if not text.strip(): raise HTTPException(400, "missing text") voice = (message.get("voiceId") or DEFAULT_VOICE).split("/")[-1] accent = message.get("accent") or DEFAULT_ACCENT speed = float(message.get("speed") or 1.0) return StreamingResponse( _stream_synthesis(text, voice, accent, speed), media_type="application/octet-stream", ) # ---------------------------- Vapi: transcriber ---------------------------- @app.post("/vapi/transcriber") async def vapi_transcriber(request: Request, x_vapi_secret: Optional[str] = Header(default=None)): """Vapi custom-transcriber endpoint (multipart: message JSON + raw PCM).""" _check_secret(x_vapi_secret) form = await request.form() raw = form.get("audio") meta = form.get("message") sample_rate = 16000 if meta is not None: try: payload = json.loads(meta if isinstance(meta, str) else await meta.read()) sample_rate = int(payload.get("sampleRate") or sample_rate) except Exception: pass pcm = await raw.read() if hasattr(raw, "read") else (raw or b"") if not pcm: return JSONResponse({"type": "transcriber-response", "transcription": "", "channel": "customer"}) result = await _transcribe(_pcm_to_wav(pcm, sample_rate)) return JSONResponse( { "type": "transcriber-response", "transcription": result.get("text", ""), "channel": "customer", } ) # --------------------------------- Ultravox -------------------------------- @app.post("/ultravox/tts") async def ultravox_tts(request: Request): """Returns a complete audio/wav body — usable by Ultravox or any HTTP TTS hook.""" body = await request.json() text = body.get("text") or body.get("input") or "" if not text.strip(): raise HTTPException(400, "missing text") voice = body.get("voice") or DEFAULT_VOICE accent = body.get("accent") or DEFAULT_ACCENT speed = float(body.get("speed") or 1.0) return StreamingResponse( _stream_synthesis(text, voice, accent, speed), media_type="application/octet-stream", ) @app.post("/ultravox/stt") async def ultravox_stt(file: UploadFile = File(...)): result = await _transcribe(await file.read()) return JSONResponse({"text": result.get("text", ""), "language": "ar-SA"}) @app.get("/health") async def health(): return {"ok": True, "voice": DEFAULT_VOICE, "accent": DEFAULT_ACCENT}