server.py

#!/usr/bin/env python3
"""Vapi custom-voice adapter for Gandr TTS (stdlib only, one file).

Vapi POSTs {"message": {"type": "voice-request", "text": ..., "sampleRate": ...}}.
The adapter renders the text through Gandr's OpenAI-compatible endpoint
(POST https://tts.gandr.ai/v1/audio/speech, response_format "pcm": 16-bit
little-endian mono at 24000 Hz), resamples to the requested rate, and replies
200 application/octet-stream with raw PCM, which is what Vapi expects.

Env: GANDR_API_KEY (required, keys at gandr.ai), VAPI_SECRET (recommended),
GANDR_VOICE (default gandr-ava), GANDR_PCM_RATE (default 24000), PORT (8000)."""
import array, json, os, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

API = os.environ.get("GANDR_API", "https://tts.gandr.ai/v1/audio/speech")
KEY = os.environ["GANDR_API_KEY"]
SECRET = os.environ.get("VAPI_SECRET", "")
VOICE = os.environ.get("GANDR_VOICE", "gandr-ava")
SRC = int(os.environ.get("GANDR_PCM_RATE", "24000"))

def gandr_pcm(text):
    body = {"model": "gandr-1", "voice": VOICE, "input": text[:2000],
            "response_format": "pcm"}
    req = urllib.request.Request(API, data=json.dumps(body).encode(), headers={
        "Authorization": "Bearer " + KEY, "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=60) as r:
        return r.read()

def resample(pcm, src, dst):
    if src == dst:
        return pcm
    s = array.array("h"); s.frombytes(pcm)
    out = array.array("h", bytes(2 * (len(s) * dst // src)))
    for i in range(len(out)):
        pos = i * src / dst
        j = int(pos)
        nxt = s[j + 1] if j + 1 < len(s) else s[j]
        out[i] = int(s[j] + (nxt - s[j]) * (pos - j))
    return out.tobytes()

class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        if SECRET and self.headers.get("x-vapi-secret") != SECRET:
            self.send_error(401)
            return
        raw = self.rfile.read(int(self.headers.get("Content-Length") or 0))
        msg = json.loads(raw or b"{}").get("message", {})
        if msg.get("type") != "voice-request" or not msg.get("text"):
            self.send_error(400)
            return
        pcm = resample(gandr_pcm(msg["text"]), SRC, int(msg.get("sampleRate") or SRC))
        self.send_response(200)
        self.send_header("Content-Type", "application/octet-stream")
        self.send_header("Content-Length", str(len(pcm)))
        self.end_headers()
        self.wfile.write(pcm)

if __name__ == "__main__":
    ThreadingHTTPServer(("", int(os.environ.get("PORT", "8000"))), Handler).serve_forever()