Skip to content
Kheelona Voice

Python server example

Two small programs for a Python backend:

  • app.py is a mint route for a web page. POST /api/voice-session creates a session with your API key and hands the page a client secret (the session ID, the client secret and the connection addresses), so the key never reaches a browser.
  • relay.py is a server-to-server relay. It sends a typed turn and/or a WAV file to a voice over the WebSocket, prints the events and saves the reply audio. A toy or a phone-line relay works the same way.
  • kheelona.py is the helper both use: create_session() with an idempotency key and retries on 429 and 503.
  • wavio.py reads and writes WAV files with the standard library.

You need Python 3.12 or newer, uv, and the example folder: the examples you received from the Kheelona team, with the SDK package.

Settings

Section titled “Settings”
Variable Default Meaning
KHEELONA_API_KEY none (required) Your API key. Server side only.
KHEELONA_API_URL https://api.voice.kheelona.com The API
KHEELONA_VOICE_ID lumi-dragon The voice
HOST / PORT 127.0.0.1 / 5182 Where the mint route listens
ALLOWED_ORIGIN http://localhost:5180 The one page origin allowed to call the mint route

Run

Section titled “Run”
Terminal window
cd examples/python-server
uv sync
export KHEELONA_API_KEY=kv_live_…
uv run python app.py                   # the mint route on port 5182


uv run python relay.py --text "Hello! Tell me a short story." --out reply.wav
uv run python relay.py --wav question.wav --out reply.wav --languages en,hi

The WAV must be 16 kHz, 16-bit, mono. Convert any recording with ffmpeg -i in.wav -ac 1 -ar 16000 -sample_fmt s16 out.wav. The relay prints events on standard output and errors on standard error as error <code> request_id <id>. See the server quickstart.

Before you copy it

Section titled “Before you copy it”
  • A request with no Origin, a different Origin, or a body that is not JSON is refused (403 or 415) before any session is created. The route is meant to be same-origin with your page, or behind your own proxy: CORS is not enabled.
  • Add your own sign-in check where app.py says Authenticate your user here, and pass that user’s opaque ID as end_user_ref. See Security.
  • Never print, log or commit the API key.

The files

Section titled “The files”
python-server/app.py
app.py
"""A mint route for browsers: POST /api/voice-session creates a session and hands the page a client secret.


Run it:  KHEELONA_API_KEY=... uv run python app.py   (it listens on 127.0.0.1:5182 unless HOST/PORT say otherwise)


The page never sees your API key. It gets a client secret that works once and expires after 60 seconds.
"""


import json
import logging
import os
import re
import sys
from collections.abc import Mapping
from contextlib import asynccontextmanager


import httpx
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


from kheelona import KheelonaApiError, create_session


DEFAULT_ORIGIN = "http://localhost:5180"
DEFAULT_API_URL = "https://api.voice.kheelona.com"
MAX_BODY_BYTES = 2048
log = logging.getLogger("kheelona.example")




def listen_address(env: Mapping[str, str]) -> tuple[str, int]:
    """Loopback by default. Binding to another address is your decision, and then you need real authentication."""
    return env.get("HOST", "127.0.0.1"), int(env.get("PORT", "5182"))




# The answers carry a client secret: never cache them, and say that they depend on the Origin header.
NO_STORE = {"Cache-Control": "no-store", "Vary": "Origin"}




def _error(status: int, code: str, headers: dict[str, str] | None = None) -> JSONResponse:
    return JSONResponse({"error": {"code": code}}, status_code=status, headers=headers)




def _languages(body: object) -> list[str] | None:
    """The `languages` of a request body, None if it has none, ValueError if the body is not what we accept."""
    if not isinstance(body, dict) or set(body) - {"languages"}:
        raise ValueError("body")
    languages = body.get("languages")
    if languages is None:
        return None
    if not isinstance(languages, list) or not 1 <= len(languages) <= 11:
        raise ValueError("languages")
    if not all(isinstance(code, str) and re.fullmatch(r"[a-z]{2,3}", code) for code in languages):
        raise ValueError("languages")
    return languages




def create_app(
    env: Mapping[str, str],
    *,
    transport: httpx.AsyncBaseTransport | None = None,
) -> FastAPI:
    """Build the app from environment settings. `transport` is only for tests."""
    api_key = env["KHEELONA_API_KEY"]  # KeyError if missing: main() turns it into a one-line hint
    api_url = env.get("KHEELONA_API_URL", DEFAULT_API_URL).rstrip("/")
    voice_id = env.get("KHEELONA_VOICE_ID", "lumi-dragon")
    allowed_origin = env.get("ALLOWED_ORIGIN", DEFAULT_ORIGIN).strip()
    if allowed_origin.lower() in ("", "null"):
        raise ValueError("ALLOWED_ORIGIN must be your page's origin, for example https://app.example.com")
    client = httpx.AsyncClient(transport=transport, timeout=15.0)


    @asynccontextmanager
    async def lifespan(_app: FastAPI):
        yield
        await client.aclose()


    app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)


    @app.post("/api/voice-session")
    async def mint(request: Request) -> JSONResponse:
        # Only your own page may call this: a missing or foreign Origin is refused before anything else.
        if request.headers.get("origin") != allowed_origin:
            return _error(403, "origin_not_allowed")
        # JSON only, so a cross-site HTML form cannot start a session.
        media_type, _, _ = request.headers.get("content-type", "").partition(";")
        if media_type.strip().lower() != "application/json":
            return _error(415, "unsupported_media_type", NO_STORE)
        # Read the body in pieces and stop as soon as it passes the limit, even with no Content-Length.
        raw = bytearray()
        async for chunk in request.stream():
            raw += chunk
            if len(raw) > MAX_BODY_BYTES:
                return _error(413, "payload_too_large", NO_STORE)
        try:
            languages = _languages(json.loads(raw))
        except ValueError:
            return _error(400, "invalid_request", NO_STORE)


        # Authenticate your user here: only signed-in users may start a session.
        # The end_user_ref is that user's opaque id (never a name or an email); this demo uses a fixed one.
        try:
            session = await create_session(
                client,
                api_url=api_url,
                api_key=api_key,
                voice_id=voice_id,
                languages=languages,
                end_user_ref="demo-user",
            )
        except KheelonaApiError as err:
            log.warning("create session failed: status=%s code=%s request_id=%s", err.status, err.code, err.request_id)
            body = {"code": err.code, **({"requestId": err.request_id} if err.request_id else {})}
            if err.status in (429, 503):
                headers = {**NO_STORE, **({"Retry-After": str(err.retry_after)} if err.retry_after else {})}
                return JSONResponse({"error": body}, status_code=err.status, headers=headers)
            return JSONResponse({"error": body}, status_code=502, headers=NO_STORE)
        except httpx.HTTPError as err:
            log.warning("create session failed: %s", type(err).__name__)
            return _error(502, "upstream_unreachable", NO_STORE)


        # Return only what the page needs, nothing else from the API's answer.
        return JSONResponse(
            {
                "sessionId": session["session_id"],
                "clientSecret": session["client_secret"]["value"],
                "endpoints": {
                    "webrtcUrl": session["connect"]["webrtc_url"],
                    "websocketUrl": session["connect"]["websocket_url"],
                },
            },
            status_code=201,
            headers=NO_STORE,
        )


    return app




def __getattr__(name: str) -> FastAPI:
    """`uvicorn app:app` works: the app is built on first use, from the environment."""
    if name == "app":
        return create_app(os.environ)
    raise AttributeError(name)




def main(env: Mapping[str, str] = os.environ) -> int:
    import uvicorn


    try:
        app = create_app(env)
    except KeyError:
        print("Set KHEELONA_API_KEY to your Kheelona Voice API key (server side only).", file=sys.stderr)
        return 2
    host, port = listen_address(env)
    uvicorn.run(app, host=host, port=port, log_level="warning")
    return 0




if __name__ == "__main__":
    sys.exit(main())
python-server/kheelona.py
kheelona.py
"""Create a Kheelona Voice session from your server (POST /v1/sessions).


Call this from your own backend only: it needs your API key, which must never reach a browser or a device.
"""


import asyncio
import re
import uuid
from collections.abc import Sequence


import httpx


MAX_ATTEMPTS = 3
MAX_WAIT_S = 30




class KheelonaApiError(Exception):
    """The API refused the request. Carries the status, error code and request id, never the key."""


    def __init__(self, status: int, code: str, request_id: str | None, retry_after: int | None = None):
        super().__init__(f"Kheelona API error {status} {code} (request_id {request_id or '-'})")
        self.status = status
        self.code = code
        self.request_id = request_id
        self.retry_after = retry_after




async def create_session(
    client: httpx.AsyncClient,
    *,
    api_url: str,
    api_key: str,
    voice_id: str,
    languages: Sequence[str] | None = None,
    end_user_ref: str | None = None,
    interruptions: bool = True,
) -> dict:
    """Create a session and return the API's answer: session_id, client_secret, connect, ...


    One Idempotency-Key covers every attempt, so a retry never creates a second session. A 429 or 503 is
    retried up to 3 times in all, after the seconds in its Retry-After header (1 if it has none).
    """
    body: dict = {"voice_id": voice_id, "languages": list(languages or ["en", "hi"])}
    if end_user_ref:
        body["end_user_ref"] = end_user_ref  # an opaque id of yours, never a name or an email
    if not interruptions:
        body["interruptions"] = False
    headers = {"Authorization": f"Bearer {api_key}", "Idempotency-Key": str(uuid.uuid4())}
    for attempt in range(1, MAX_ATTEMPTS + 1):
        response = await client.post(f"{api_url}/v1/sessions", json=body, headers=headers)
        if response.is_success:
            return response.json()
        error = _api_error(response)
        if response.status_code not in (429, 503) or attempt == MAX_ATTEMPTS:
            raise error
        await asyncio.sleep(min(error.retry_after or 1, MAX_WAIT_S))
    raise AssertionError("unreachable")




def _api_error(response: httpx.Response) -> KheelonaApiError:
    try:
        detail = response.json()["error"]
        code, request_id = _plain(detail["code"]) or "unknown_error", _plain(detail.get("request_id"))
    except (ValueError, KeyError, TypeError):
        code, request_id = "unknown_error", None
    retry_after = response.headers.get("Retry-After", "")
    return KheelonaApiError(response.status_code, code, request_id, int(retry_after) if retry_after.isdigit() else None)




def _plain(value: object) -> str | None:
    """A short identifier made of plain characters, or None: nothing else from an answer is repeated."""
    return value if isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9_.:-]{1,64}", value) else None
python-server/relay.py
relay.py
"""A server-to-server relay: send a typed turn and/or a WAV file to a Kheelona voice, print the events, save the reply.


    KHEELONA_API_KEY=... uv run python relay.py --text "Hello!" --wav question.wav --out reply.wav


Event text (what the voice heard and said) goes only to this terminal. Kheelona stores none of it, and this
example writes nothing but the reply audio file you asked for.
"""


import argparse
import asyncio
import json
import os
import struct
import sys
import time
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass


import httpx
from websockets.asyncio.client import connect
from websockets.exceptions import InvalidStatus, WebSocketException


from kheelona import KheelonaApiError, create_session
from wavio import read_pcm16_wav, write_wav


FRAME_S = 0.020  # one frame every 20 ms
FRAME_BYTES = 640  # 20 ms of 16 kHz, 16-bit mono audio
SILENCE = bytes(FRAME_BYTES)
OUTPUT_RATES = {"pcm16_16k": 16000, "pcm16_24k": 24000, "pcm16_48k": 48000}




class RelayError(Exception):
    """The session failed. `code`, `session_id` and `request_id` are safe to print.


    A realtime `error` event carries no request ID: quote the session ID and the code. An HTTP refusal carries a
    request ID.
    """


    def __init__(self, code: str, request_id: str | None = None, session_id: str | None = None):
        super().__init__(f"{code} (session {session_id or '-'}, request_id {request_id or '-'})")
        self.code = code
        self.request_id = request_id
        self.session_id = session_id




@dataclass
class RelaySummary:
    session_id: str
    end_reason: str
    replies: int
    audio_seconds: float
    out_file: str




@dataclass
class RelayOptions:
    api_url: str
    api_key: str
    voice_id: str = "lumi-dragon"
    languages: Sequence[str] = ("en", "hi")
    text: str | None = None
    wav_path: str | None = None
    out_path: str = "reply.wav"
    interruptions: bool = True
    pause_playback: bool = True  # declare playback_pause and pause the playback buffer on reply.paused
    out: Callable[[str], None] = print
    reply_timeout: float = 20.0  # seconds without anything from the server before giving up
    transport: httpx.AsyncBaseTransport | None = None  # a test hook




class FrameSource:
    """What to send next: the speech in 20 ms frames, then silence for as long as you keep asking."""


    def __init__(self) -> None:
        self._pending = bytearray()


    def load(self, pcm: bytes) -> None:
        self._pending += pcm


    def next(self) -> bytes:
        frame = bytes(self._pending[:FRAME_BYTES])
        del self._pending[:FRAME_BYTES]
        return frame.ljust(FRAME_BYTES, b"\0") if frame else SILENCE




async def pump(send, source: FrameSource, stop: asyncio.Event) -> None:
    """Send one frame every 20 ms, in real time, until `stop` is set.


    Frame k is due at start + k * 20 ms. Sleeping until that deadline (never a fixed 20 ms) keeps the pace
    exact: a late wake-up is made up on the next frame instead of piling up. Send silence while nobody
    speaks: the server's turn detection needs a steady stream to know when a person has finished.
    """
    start = time.monotonic()
    k = 0
    while not stop.is_set():
        delay = start + k * FRAME_S - time.monotonic()
        if delay > 0:
            await asyncio.sleep(delay)
        await send(source.next())
        k += 1
        await asyncio.sleep(0)  # give the receiving side a turn








class PausePlayback:
    """The playback buffer, with pause support.


    Frames wait in `push` until `drain` hands them to the speaker. While the voice has paused its own playback
    (`reply.paused`) `drain` returns nothing and the frames are kept, so nothing is lost. A pause ends at the first of
    `reply.resumed`, `interrupted` (the buffer is flushed), `reply.done` or the next `reply.started`. A pause, resume
    or done for any reply other than the one that is playing (or one that has already ended) is stale and ignored.
    """


    def __init__(self, enabled: bool = True) -> None:
        self.enabled = enabled  # False when the relay did not declare playback_pause
        self.paused = False
        self.dropped = 0  # frames thrown away by `interrupted` or by the end of the session
        self._queue: list[bytes] = []
        self._current: str | None = None


    def on_event(self, event: dict) -> None:
        kind = event.get("type")
        if kind == "reply.started":
            self._current = event.get("reply_id")  # a new reply ends any pause
            self.paused = False
        elif kind == "reply.paused":
            if self.enabled and self._current is not None and event.get("reply_id") == self._current:
                self.paused = True
        elif kind == "reply.resumed":
            if event.get("reply_id") == self._current:
                self.paused = False
        elif kind in ("interrupted", "session.ended"):  # flush: what was queued will never play
            self.dropped += len(self._queue)
            self._queue = []
            self.paused = False
            self._current = None  # that reply is over: a pause naming it is stale
        elif kind == "reply.done":
            if event.get("reply_id") == self._current:
                self.paused = False
                self._current = None


    def push(self, frame: bytes) -> None:
        self._queue.append(frame)


    def drain(self) -> list[bytes]:
        """The frames to play now: none while paused."""
        if self.paused:
            return []
        frames, self._queue = self._queue, []
        return frames








class Reply:
    """Collects the reply audio and counts finished replies."""


    def __init__(self, out: Callable[[str], None], pause: bool = True) -> None:
        self.out = out
        self.playback = PausePlayback(enabled=pause)
        self.pcm = bytearray()
        self.done = 0  # every reply.done, interrupted ones too
        self.answered = 0  # replies that finished a turn
        self.max_seq = 0  # highest reply_seq seen
        self.drop_upto = 0  # frames at or below this reply_seq belong to an interrupted reply
        self.output_rate = 24000
        self.session_id = ""
        self.ended: str | None = None


    def on_binary(self, data: bytes) -> None:
        """Reply audio: an 8-byte header (kind, flags, sentence, reply_seq) then PCM16 samples."""
        if len(data) < 8:
            return
        kind, _flags, _sentence, seq = struct.unpack_from("<BBHI", data)
        if kind != 1:  # 1 = reply audio
            return
        self.max_seq = max(self.max_seq, seq)
        if seq <= self.drop_upto:
            return  # a late frame of a reply that was interrupted
        self.playback.push(data[8:])
        self._play()


    def on_event(self, event: dict) -> None:
        kind = event.get("type")
        self.out(f"event {kind}")
        self.playback.on_event(event)
        self._play()  # a resume, a done or a new reply lets the held frames out
        if kind == "session.ready":
            self.session_id = event.get("session_id", self.session_id)
            self.output_rate = OUTPUT_RATES.get(event.get("audio", {}).get("output_format"), 24000)
        elif kind in ("transcript.final", "reply.sentence"):
            self.out(f"  text: {event.get('text', '')}")
        elif kind == "interrupted":
            self.drop_upto = self.max_seq  # flush what is queued, drop the rest of this reply
        elif kind == "reply.done":
            self.done += 1
            if event.get("reason") != "interrupted":  # an interrupted reply is followed by a new one
                self.answered += 1
        elif kind == "error" and event.get("fatal"):
            raise RelayError(str(event.get("code", "error")))
        elif kind == "session.ended":
            self.ended = str(event.get("reason", "unknown"))




    def _play(self) -> None:
        # A real player would hand these frames to the speaker; this example saves them.
        for frame in self.playback.drain():
            self.pcm += frame




async def _receive(ws, reply: Reply, timeout: float) -> None:
    """Handle one message from the server; give up after `timeout` seconds without one."""
    try:
        message = await asyncio.wait_for(ws.recv(), timeout)
    except TimeoutError:
        raise RelayError("no_reply") from None
    if isinstance(message, bytes):
        reply.on_binary(message)
    else:
        reply.on_event(json.loads(message))




def _refusal(err: InvalidStatus) -> RelayError:
    try:
        detail = json.loads(err.response.body or b"")["error"]
        return RelayError(str(detail["code"]), detail.get("request_id"))
    except (ValueError, KeyError, TypeError):
        return RelayError(f"http_{err.response.status_code}")




def _http(opts: RelayOptions) -> httpx.AsyncClient:
    """The HTTP client for creating the session, with a 15 second timeout."""
    return httpx.AsyncClient(transport=opts.transport, timeout=15.0)




async def run_relay(opts: RelayOptions) -> RelaySummary:
    speech = read_pcm16_wav(opts.wav_path)[1] if opts.wav_path else b""
    reply = Reply(opts.out, pause=opts.pause_playback)


    async with _http(opts) as http:
        session = await create_session(
            http,
            api_url=opts.api_url,
            api_key=opts.api_key,
            voice_id=opts.voice_id,
            languages=opts.languages,
            interruptions=opts.interruptions,
        )
    secret = session["client_secret"]["value"]
    try:
        # Servers send no Origin header (the websockets library adds none).
        async with connect(
            session["connect"]["websocket_url"],
            subprotocols=["kheelona.v1"],
            additional_headers={"Authorization": f"Bearer {secret}"},
        ) as ws:
            # Send nothing until the server says session.ready: audio sent before it is dropped.
            while not reply.session_id:
                await _receive(ws, reply, opts.reply_timeout)
            if opts.pause_playback:  # tell the voice this client can pause and resume its playback buffer
                await ws.send(json.dumps({"type": "client.capabilities", "playback_pause": True}))
            source = FrameSource()
            stop = asyncio.Event()
            turns = 0
            if opts.text:  # the typed turn goes first, and its reply is awaited before any audio is sent
                await ws.send(json.dumps({"type": "input.text", "text": opts.text}))
                turns += 1
            pumping = asyncio.create_task(pump(ws.send, source, stop))
            try:
                while reply.answered < turns:
                    await _receive(ws, reply, opts.reply_timeout)
                if speech:
                    source.load(speech)  # real time, 20 ms frames, then silence until the reply is done
                    turns += 1
                    while reply.answered < turns:
                        await _receive(ws, reply, opts.reply_timeout)
            finally:
                stop.set()
                await asyncio.gather(pumping, return_exceptions=True)  # the first error wins
            await ws.send(json.dumps({"type": "session.end"}))
            while reply.ended is None:
                await _receive(ws, reply, opts.reply_timeout)
    except InvalidStatus as err:
        raise _refusal(err) from None
    except RelayError as err:
        err.session_id = err.session_id or reply.session_id or None
        raise
    except (WebSocketException, OSError) as err:
        raise RelayError(f"connection_failed_{type(err).__name__}", session_id=reply.session_id or None) from None


    write_wav(opts.out_path, bytes(reply.pcm), reply.output_rate)
    seconds = len(reply.pcm) / 2 / reply.output_rate
    summary = RelaySummary(reply.session_id, reply.ended, reply.done, seconds, opts.out_path)
    opts.out(
        f"session {summary.session_id} ended {summary.end_reason} replies {summary.replies} "
        f"audio {seconds:.1f}s saved {summary.out_file}"
    )
    return summary




def main(argv: Sequence[str] | None = None, env: Mapping[str, str] = os.environ) -> int:
    parser = argparse.ArgumentParser(description="Talk to a Kheelona voice from a script and save the reply audio.")
    parser.add_argument("--text", help="a typed turn (the voice answers it first)")
    parser.add_argument("--wav", help="a 16-bit mono 16 kHz WAV file, sent as a spoken turn")
    parser.add_argument("--out", default="reply.wav", help="where to save the reply audio (default reply.wav)")
    parser.add_argument("--voice", default=env.get("KHEELONA_VOICE_ID", "lumi-dragon"))
    parser.add_argument("--languages", default="en,hi", help="comma-separated language codes (default en,hi)")
    parser.add_argument("--no-interruptions", action="store_true", help="the voice is never cut off")
    parser.add_argument(
        "--no-pause", action="store_true", help="do not declare playback_pause (the voice never pauses)"
    )
    args = parser.parse_args(argv)
    if not args.text and not args.wav:
        parser.error("give --text, --wav or both")
    if args.text and len(args.text) > 2000:
        parser.error("--text is at most 2000 characters")
    api_key = env.get("KHEELONA_API_KEY")
    if not api_key:
        print("Set KHEELONA_API_KEY to your Kheelona Voice API key (server side only).", file=sys.stderr)
        return 2
    if args.wav:
        try:
            read_pcm16_wav(args.wav)
        except (ValueError, OSError) as err:
            print(err, file=sys.stderr)
            return 2
    opts = RelayOptions(
        api_url=env.get("KHEELONA_API_URL", "https://api.voice.kheelona.com").rstrip("/"),
        api_key=api_key,
        voice_id=args.voice,
        languages=[code.strip() for code in args.languages.split(",") if code.strip()],
        text=args.text,
        wav_path=args.wav,
        out_path=args.out,
        interruptions=not args.no_interruptions,
        pause_playback=not args.no_pause,
    )
    try:
        asyncio.run(run_relay(opts))
    except KheelonaApiError as err:
        print(f"error {err.code} request_id {err.request_id or '-'}", file=sys.stderr)
        return 1
    except RelayError as err:
        session = f" session {err.session_id}" if err.session_id else ""
        print(f"error {err.code}{session} request_id {err.request_id or '-'}", file=sys.stderr)
        return 1
    except httpx.HTTPError as err:
        print(f"error api_unreachable_{type(err).__name__} request_id -", file=sys.stderr)
        return 1
    return 0




if __name__ == "__main__":
    sys.exit(main())
python-server/wavio.py
wavio.py
"""Read and write the WAV files the relay uses (standard library only)."""


import wave
from pathlib import Path


HINT = "Convert it with: ffmpeg -ac 1 -ar 16000 -sample_fmt s16 -i in.wav out.wav"




def read_pcm16_wav(path: str | Path) -> tuple[int, bytes]:
    """Return (sample_rate, raw PCM bytes) of a 16-bit, mono, 16 kHz WAV, or raise ValueError saying what is wrong."""
    try:
        with wave.open(str(path), "rb") as w:
            channels, width, rate = w.getnchannels(), w.getsampwidth(), w.getframerate()
            problems = []
            if channels != 1:
                problems.append("stereo" if channels == 2 else f"{channels} channels")
            if width != 2:
                problems.append(f"{width * 8}-bit samples")
            if rate != 16000:
                problems.append(f"a sample rate of {rate} Hz")
            if problems:
                found = " and ".join(problems)
                raise ValueError(f"{path}: the WAV must be mono, 16-bit, 16 kHz, but it has {found}. {HINT}")
            return rate, w.readframes(w.getnframes())
    except (wave.Error, EOFError) as err:
        raise ValueError(f"{path}: not a PCM WAV file ({err}). {HINT}") from err




def write_wav(path: str | Path, pcm: bytes, rate: int) -> None:
    """Write raw 16-bit mono PCM as a WAV file."""
    with wave.open(str(path), "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(rate)
        w.writeframes(pcm)
python-server/pyproject.toml
pyproject.toml
[project]
name = "kheelona-example-python-server"
version = "0.1.0"
description = "Kheelona Voice example: create a session, mint client secrets for a browser, and relay audio over the WebSocket."
readme = "README.md"
license = { text = "Proprietary" }
requires-python = ">=3.12"
dependencies = [
    "fastapi==0.143.0",
    "httpx==0.28.1",
    "uvicorn==0.54.0",
    "websockets==17.2",
]


[dependency-groups]
dev = ["pytest==9.1.1", "ruff==0.17.0"]


[tool.uv]
# A sample you copy from, not a package: nothing to build or install.
package = false
# Supply-chain guard: resolve only releases at least 14 days older than 2026-10-30, the day these examples were written.
exclude-newer = "2026-10-16T00:00:00Z"


[tool.ruff]
line-length = 120
target-version = "py312"


[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]


[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["."]

The examples are Kheelona’s own and are given to partners privately. Contact the Kheelona team for licensing.