Server-to-server
Use this path when your own server talks to Kheelona for a device: a toy relay, a phone-line bridge, a test tool. Your server holds the API key. The device never sees it.
The relay receives the toy’s microphone audio, sends it to Kheelona as it arrives, and plays the reply audio back. Run the quickstart first to see all of this work in a few minutes. A web page uses the browser SDK instead; its SDK reference lists every call.
The flow
Section titled “The flow”- Your server creates a session with
POST /v1/sessionsand your API key. The answer holds asession_id, a one-timeclient_secretand theconnect.websocket_url. See Sessions. - Your server opens a WebSocket to that URL with the client secret in a header.
- Kheelona sends
session.ready, thenstate: listening. From now on you may send audio. - You send microphone audio as binary frames. Kheelona sends events as text frames and the voice’s audio as binary frames.
- You send
session.end, wait forsession.ended, and close.
Connect
Section titled “Connect”Open wss://api.voice.kheelona.com/v1/realtime?session_id=… with these three things:
- the header
Authorization: Bearer kvcs_…(the client secret from step 1); - the subprotocol
kheelona.v1; - no
Originheader.
A server client must not send Origin. A request that has one is treated as a browser and checked against your allowed
web origins. If your WebSocket library adds one by default, turn it off.
The client secret works once and expires about 60 seconds after it was issued. Connect straight away. If the connection fails for any reason, create a new session: a failed attempt needs a new secret.
/** * Opens the WebSocket the way a server does: the client secret in the Authorization header, the `kheelona.v1` * subprotocol, and no Origin header (a request with one is treated as a browser). Attach your listeners straight * away: the server's first event can arrive in the same moment the connection opens. */function openSocket(session: CreatedSession): WebSocket { return new WebSocket(session.websocketUrl, ['kheelona.v1'], { headers: { Authorization: `Bearer ${session.clientSecret}` }, handshakeTimeout: 20_000, });}
/** Before the upgrade, errors come back as HTTP errors with a JSON body: keep the code and the request ID. */function readRefusal(res: IncomingMessage): Promise<RelayError> { return new Promise((resolve) => { const chunks: Buffer[] = []; res.on('data', (chunk: Buffer) => chunks.push(chunk)); res.on('end', () => { let code = `http_${String(res.statusCode)}`; let requestId: string | undefined; try { const body = JSON.parse(Buffer.concat(chunks).toString('utf8').slice(0, 8192)) as { error?: { code?: string; request_id?: string } }; if (typeof body.error?.code === 'string' && /^[a-z0-9_]{1,64}$/.test(body.error.code)) code = body.error.code; if (typeof body.error?.request_id === 'string' && /^[A-Za-z0-9_]{1,64}$/.test(body.error.request_id)) requestId = body.error.request_id; } catch { // Not JSON: keep the HTTP status as the code. } resolve(new RelayError(code, `The connection was refused (${code})`, requestId)); }); });}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}))Before the upgrade, a refusal is a normal HTTP error with a JSON body, such as
{"error":{"code":"unauthenticated","message":"…","request_id":"req_…"}}. Keep the code and the request_id. The codes
are on Errors and retries.
Wait for session.ready
Section titled “Wait for session.ready”Send nothing until session.ready arrives. Audio sent before it is dropped. The event tells you what the session
runs with: the voice, the languages, the audio formats, the limits and whether interruptions is on.
Attach your message handler as soon as the socket opens. The first event can arrive in the same moment.
Send audio at real time
Section titled “Send audio at real time”Send the microphone as binary frames:
- PCM, signed 16-bit little-endian, mono, no header;
- 16 kHz by default (
pcm16_16k; 24 kHz and 48 kHz are also available); - 20 ms per frame is a good size (640 bytes at 16 kHz). Anything from 10 ms to 500 ms is accepted.
Pace the frames like a microphone: one 20 ms frame every 20 ms. Time each frame against a steady clock, not with a fixed sleep, so a late wake-up never makes you drift.
- Audio that arrives faster than twice real time is dropped, and you get one non-fatal
invalid_audioerror. - Quiet time does not build up an allowance: at most 1 second of audio can run ahead of real time.
- A frame with an odd number of bytes, or outside 10 to 500 ms, is dropped with
invalid_audio. - A WebSocket message over 64 KB closes the connection.
Keep sending while the session is live, silence included. Kheelona decides when the person has finished speaking
from the audio stream itself. A stream that stops leaves the voice waiting. Send {"type":"input.mute","muted":true}
when the microphone is off. Muting pauses the quiet-session check-in and goodbye counts, but the idle timeout keeps running.
const FRAME_MS = 20;const SAMPLES_PER_FRAME = 16 * FRAME_MS; // 16 kHz: 320 samples = 640 bytesconst SILENCE = Buffer.alloc(SAMPLES_PER_FRAME * 2);
/** Splits microphone audio (16 kHz PCM16) into 20 ms frames; the last one is padded with silence. */export function wavToFrames(samples: Int16Array): Buffer[] { const frames: Buffer[] = []; for (let at = 0; at < samples.length; at += SAMPLES_PER_FRAME) { const frame = Buffer.alloc(SAMPLES_PER_FRAME * 2); samples.subarray(at, at + SAMPLES_PER_FRAME).forEach((sample, i) => { frame.writeInt16LE(sample, i * 2); }); frames.push(frame); } return frames;}
/** * Sends one frame every 20 ms, in real time. Frame k goes out at start + k × 20 ms, measured against the clock, so a * late wake-up never makes the stream drift (the server drops audio that arrives faster than 2× real time). * Stops when `next` returns nothing or `stop` says so. */export async function sendPaced(opts: { next: () => Buffer | undefined; send: (frame: Buffer) => void; stop: () => boolean }): Promise<void> { const start = performance.now(); for (let k = 0; !opts.stop(); k += 1) { const frame = opts.next(); if (!frame) return; opts.send(frame); await new Promise((resolve) => setTimeout(resolve, Math.max(0, start + (k + 1) * FRAME_MS - performance.now()))); }}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 turnYou can also send a typed turn as {"type":"input.text","text":"…"} (up to 2,000 characters). It goes through the same safety
checks. On a child voice, what is said aloud is checked a little more strictly than typed text, because speech
recognition can change the words. A typed turn ends whatever is in progress. The quickstart uses it so you can test
without a microphone.
Pause support
Section titled “Pause support”A toy relay can make the voice stop faster and resume cleaner by pausing its own playback buffer when the person starts to speak. It is optional.
- Declare it once after
session.ready:{"type":"client.capabilities","playback_pause":true}. - Pause your buffer on
reply.pausedand resume it onreply.resumed. Keep the audio you hold while paused. - End the pause on whichever comes first:
reply.resumed,interrupted(flush the queue),reply.done, or the nextreply.started. Ignore a pause event or areply.donethat names a reply other than the current one.
/** * 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. */export class PausePlayback { private queue: Buffer[] = []; private current: string | undefined; private isPaused = false; /** Frames thrown away by `interrupted` or by the end of the session. */ dropped = 0;
/** `enabled = false` when the relay did not declare `playback_pause`: the server sends no pause events then. */ private readonly enabled: boolean; constructor(enabled = true) { this.enabled = enabled; }
get paused(): boolean { return this.isPaused; }
onEvent(event: { type: string; reply_id?: unknown }): void { switch (event.type) { case 'reply.started': this.current = typeof event.reply_id === 'string' ? event.reply_id : undefined; // a new reply ends any pause this.isPaused = false; break; case 'reply.paused': if (this.enabled && this.current !== undefined && event.reply_id === this.current) this.isPaused = true; break; case 'reply.resumed': if (event.reply_id === this.current) this.isPaused = false; break; case 'interrupted': // flush: what was queued will never play this.dropped += this.queue.length; this.queue = []; this.isPaused = false; this.current = undefined; // that reply is over: a pause naming it is stale break; case 'reply.done': if (event.reply_id === this.current) { this.isPaused = false; this.current = undefined; } break; case 'session.ended': this.dropped += this.queue.length; this.queue = []; this.isPaused = false; break; default: break; } }
push(frame: Buffer): void { this.queue.push(frame); }
/** The frames to play now: none while paused. */ drain(): Buffer[] { if (this.isPaused) return []; const frames = this.queue; this.queue = []; return frames; }}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 framesWithout the declaration you never get these events. See Interruptions and echo for the details.
Events
Section titled “Events”Every event is one JSON object in a text frame. It has a type. Server events also have an event_id and a ts.
Ignore event types and fields you don’t know. New ones can be added without a new version.
The ones a relay needs:
| Event | What to do |
|---|---|
session.ready |
Start sending audio |
state |
listening, thinking or speaking: drive a light or a face on the toy |
transcript.final |
What the person said, as text (live only; Kheelona does not keep it) |
reply.sentence |
The text of the sentence that is playing, as a caption |
reply.done |
The reply is finished; reason says why |
interrupted |
The person talked over the voice: flush queued audio now |
reply.paused, reply.resumed |
Only if you declared playback_pause: pause and resume your playback buffer |
safety |
A safety rule acted: see below |
session.warning |
The session is about to end: 60 s before the maximum length, 20 s before the idle timeout |
error |
A problem; if fatal is true the session ends next |
session.ended |
The last event; close the socket |
Every field of every event is in the server events reference and the client events reference.
Play the reply
Section titled “Play the reply”The reply audio comes as binary frames, each with an 8-byte header and then PCM samples. The samples are 16-bit
little-endian mono at 24 kHz by default (pcm16_24k).
| Byte | Size | Field | Meaning |
|---|---|---|---|
| 0 | 1 | kind |
0x01 is reply audio. Ignore other values. |
| 1 | 1 | flags |
bit 0: first frame of a sentence. bit 1: last frame of the reply. |
| 2 | 2 | sentence |
The sentence number within the reply, little-endian. It matches reply.sentence.index. |
| 4 | 4 | reply_seq |
A number that goes up by one for each reply in the session, little-endian. |
| 8 | … | samples | PCM16 audio |
- Frames are 40 ms long. The last frame of a sentence can be shorter. The last frame of a reply can be only the header, with no samples.
- Kheelona sends audio in real time, at most 300 ms ahead of playback. Play frames as they arrive, with a jitter buffer of about 100 ms.
- More about the formats: Audio formats.
When the person talks over the voice
Section titled “When the person talks over the voice”Kheelona sends an interrupted event. Do two things:
- Flush the audio you have queued for playback.
- Drop every later frame whose
reply_seqis the interrupted reply’s or lower. Frames of that reply can still be on the wire.
Optionally, send {"type":"playback.position","reply_id":"rep_…","played_ms":1840} within 500 ms. It tells Kheelona how
much the person actually heard, so the conversation continues from the right place.
/** Collects the reply audio, dropping what the server told us to drop. */export class ReplyAudio { private readonly chunks: Buffer[] = []; private readonly playback: PausePlayback; private newestSeq = -1; private droppedThrough = -1;
constructor(playback = new PausePlayback()) { this.playback = playback; }
/** A binary frame: an 8-byte header (kind, flags, sentence, reply_seq) then PCM16 24 kHz samples. */ push(frame: Buffer): void { if (frame.length < 8 || frame.readUInt8(0) !== 0x01) return; // not reply audio: ignore unknown kinds const seq = frame.readUInt32LE(4); this.newestSeq = Math.max(this.newestSeq, seq); if (seq <= this.droppedThrough) return; // a late frame of an interrupted reply this.playback.push(frame.subarray(8)); this.play(); }
/** Every event goes here first, so the playback buffer pauses, resumes and flushes with the voice. */ onEvent(event: { type: string; reply_id?: unknown }): void { this.playback.onEvent(event); this.play(); // a resume, a done or a new reply lets the held frames out }
/** `interrupted`: drop every later frame of the reply being played. (The playback buffer is flushed by `onEvent`.) */ interrupted(): void { this.droppedThrough = this.newestSeq; }
pcm(): Buffer { this.play(); return Buffer.concat(this.chunks); }
// A real player would hand these frames to the speaker; this example saves them. private play(): void { this.chunks.push(...this.playback.drain()); }}
/** The lines printed for an event: its type, plus the text of what was heard or said. */export function describeEvent(event: { type: string; text?: unknown }): string[] { const lines = [`event ${event.type}`]; if ((event.type === 'transcript.final' || event.type === 'reply.sentence') && typeof event.text === 'string') lines.push(` text: ${event.text}`); return lines;}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"))Echo cancellation is required
Section titled “Echo cancellation is required”Your device must cancel echo: a toy whose speaker is near its microphone must not send the voice’s own sound back as
if the child were talking. Kheelona checks for its own echo too, but only as a safety net. interruptions: false
exists for content that must play through without being cut off; it is not a substitute for echo cancellation.
Keep the connection alive
Section titled “Keep the connection alive”- Kheelona sends a WebSocket ping every 15 seconds and closes the connection after 45 seconds with no pong. Every WebSocket library answers pings by itself.
{"type":"ping"}gets{"type":"pong"}back, if you want a heartbeat of your own.- Keep reading. A client that stops reading what Kheelona sends is disconnected, and the session ends as
disconnected. That means about 20 seconds of unread output, or one write blocked for 10 seconds.
Safety
Section titled “Safety”Every turn is checked, on what the person says and on what the voice is about to say. You don’t switch this on.
When a rule acts, you receive a safety event with a category and the action taken. It never contains the words
themselves. For a child voice, when the input is about self-harm, sexual content or violence, the voice answers with a
gentle line that tells the child to talk to a grown-up they trust, in the child’s language. The reply ends with
reply.done and reason: safety.
Use the safety event for your own safeguarding process: log it with your session ID, alert the right person, or show a
note to a parent. Kheelona does not keep the conversation, so the event is your record.
Categories are short snake_case labels such as self_harm, sexual and violent. The list is stable, but it can
grow: treat a label you don’t know as “other”. The full list is in the
server events reference.
Ending
Section titled “Ending”- To end the session, send
{"type":"session.end"}. Kheelona answerssession.endedand closes the socket. - Your server can also end it with
POST /v1/sessions/{id}/end. The connection closes within about 10 seconds with the reasonpartner_ended. - A session also ends by itself after its maximum length (1,800 seconds unless Kheelona set another) or after the idle
timeout (120 seconds with nobody speaking and nothing playing). You get a
session.warningfirst. session.endedcarries areason:client_ended,partner_ended,max_duration,idle_timeout,disconnected,server_shutdown,provider_failure,suspended,service_stopped(the voice service was switched off; start a new session once it is back) orerror. It also hasduration_msandturns.
Every reply.started gets exactly one reply.done, even when the session ends in the middle of a reply. It arrives before
session.ended.
Reconnecting means a new session
Section titled “Reconnecting means a new session”A session is one conversation on one connection. If the connection drops, the session ends as disconnected. To carry
on, create a new session and connect again. The new voice starts fresh: it does not remember the earlier conversation.
Reconnect politely: wait for Retry-After when the API sends it, and don’t loop faster than that.
One refusal at connect is not a reason for a new session: if the WebSocket upgrade gets 503 capacity_exhausted with
Retry-After: 1, the server you reached is busy and the client secret is still unused. Wait a second and connect again
with the same secret, up to 3 times, while it is under 60 seconds old. With a longer Retry-After, create a new session
after the wait.
Errors
Section titled “Errors”During a session, a problem arrives as an error event with code, message, retryable and fatal. A non-fatal
error (a bad frame, a bad event) leaves the session running. After a fatal error the session ends. An error event
carries no request ID: when you ask for help, quote the session ID and the code (the example relays print both on
standard error), and the time. Every code, with what to do about it, is on
Errors and retries.
The whole relay
Section titled “The whole relay”The snippets above come from the example relays, which are tested against a running service. Read the whole file when you build your own.
Node: examples/node-server/src/relay.ts
// A server-to-server relay: creates a session, streams a WAV (and/or a typed turn) to the voice over the WebSocket at// real time, prints the events, and saves the reply audio to a WAV file. Run it with `pnpm relay --text "Hello"`.import { readFile, writeFile } from 'node:fs/promises';import type { IncomingMessage } from 'node:http';import { parseArgs } from 'node:util';import { pathToFileURL } from 'node:url';import { WebSocket, type RawData } from 'ws';import { createSession, KheelonaApiError, readEnv, type CreatedSession } from './kheelona.ts';import { readPcm16Wav, writeWav } from './wav.ts';
export interface RelayOptions { apiUrl: string; apiKey: string; voiceId: string; languages?: string[]; /** A typed turn, sent first. */ text?: string; /** A 16 kHz mono PCM16 WAV, sent in real time after the typed turn's reply. */ wavFile?: string; /** Where the reply audio (24 kHz mono WAV) is saved. */ outFile: string; /** `false` = the voice is never cut off. Default: the API's default (on). */ interruptions?: boolean; /** Declare `playback_pause` and pause the playback buffer on `reply.paused`. Default: on. `false` = the voice never pauses. */ pausePlayback?: boolean; /** Where event lines go (default: standard output). */ log?: (line: string) => void; /** Milliseconds to wait for a reply before giving up (default 20,000). */ replyTimeoutMs?: number;}
export interface RelaySummary { sessionId: string; endReason: string; replies: number; audioSeconds: number; outFile: string;}
/** A bad command line or input file: the CLI exits 2. */export class UsageError extends Error {}
/** * A failed session: the code, the request ID when an HTTP answer carried one, and the session ID once the session * existed (realtime `error` events carry no request ID: quote the session ID and the code). The CLI exits 1. */export class RelayError extends Error { readonly code: string; readonly requestId: string | undefined; sessionId: string | undefined; constructor(code: string, message: string, requestId?: string) { super(message); this.name = 'RelayError'; this.code = code; this.requestId = requestId; }}
/** * Opens the WebSocket the way a server does: the client secret in the Authorization header, the `kheelona.v1` * subprotocol, and no Origin header (a request with one is treated as a browser). Attach your listeners straight * away: the server's first event can arrive in the same moment the connection opens. */function openSocket(session: CreatedSession): WebSocket { return new WebSocket(session.websocketUrl, ['kheelona.v1'], { headers: { Authorization: `Bearer ${session.clientSecret}` }, handshakeTimeout: 20_000, });}
/** Before the upgrade, errors come back as HTTP errors with a JSON body: keep the code and the request ID. */function readRefusal(res: IncomingMessage): Promise<RelayError> { return new Promise((resolve) => { const chunks: Buffer[] = []; res.on('data', (chunk: Buffer) => chunks.push(chunk)); res.on('end', () => { let code = `http_${String(res.statusCode)}`; let requestId: string | undefined; try { const body = JSON.parse(Buffer.concat(chunks).toString('utf8').slice(0, 8192)) as { error?: { code?: string; request_id?: string } }; if (typeof body.error?.code === 'string' && /^[a-z0-9_]{1,64}$/.test(body.error.code)) code = body.error.code; if (typeof body.error?.request_id === 'string' && /^[A-Za-z0-9_]{1,64}$/.test(body.error.request_id)) requestId = body.error.request_id; } catch { // Not JSON: keep the HTTP status as the code. } resolve(new RelayError(code, `The connection was refused (${code})`, requestId)); }); });}
const FRAME_MS = 20;const SAMPLES_PER_FRAME = 16 * FRAME_MS; // 16 kHz: 320 samples = 640 bytesconst SILENCE = Buffer.alloc(SAMPLES_PER_FRAME * 2);
/** Splits microphone audio (16 kHz PCM16) into 20 ms frames; the last one is padded with silence. */export function wavToFrames(samples: Int16Array): Buffer[] { const frames: Buffer[] = []; for (let at = 0; at < samples.length; at += SAMPLES_PER_FRAME) { const frame = Buffer.alloc(SAMPLES_PER_FRAME * 2); samples.subarray(at, at + SAMPLES_PER_FRAME).forEach((sample, i) => { frame.writeInt16LE(sample, i * 2); }); frames.push(frame); } return frames;}
/** * Sends one frame every 20 ms, in real time. Frame k goes out at start + k × 20 ms, measured against the clock, so a * late wake-up never makes the stream drift (the server drops audio that arrives faster than 2× real time). * Stops when `next` returns nothing or `stop` says so. */export async function sendPaced(opts: { next: () => Buffer | undefined; send: (frame: Buffer) => void; stop: () => boolean }): Promise<void> { const start = performance.now(); for (let k = 0; !opts.stop(); k += 1) { const frame = opts.next(); if (!frame) return; opts.send(frame); await new Promise((resolve) => setTimeout(resolve, Math.max(0, start + (k + 1) * FRAME_MS - performance.now()))); }}
/** * 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. */export class PausePlayback { private queue: Buffer[] = []; private current: string | undefined; private isPaused = false; /** Frames thrown away by `interrupted` or by the end of the session. */ dropped = 0;
/** `enabled = false` when the relay did not declare `playback_pause`: the server sends no pause events then. */ private readonly enabled: boolean; constructor(enabled = true) { this.enabled = enabled; }
get paused(): boolean { return this.isPaused; }
onEvent(event: { type: string; reply_id?: unknown }): void { switch (event.type) { case 'reply.started': this.current = typeof event.reply_id === 'string' ? event.reply_id : undefined; // a new reply ends any pause this.isPaused = false; break; case 'reply.paused': if (this.enabled && this.current !== undefined && event.reply_id === this.current) this.isPaused = true; break; case 'reply.resumed': if (event.reply_id === this.current) this.isPaused = false; break; case 'interrupted': // flush: what was queued will never play this.dropped += this.queue.length; this.queue = []; this.isPaused = false; this.current = undefined; // that reply is over: a pause naming it is stale break; case 'reply.done': if (event.reply_id === this.current) { this.isPaused = false; this.current = undefined; } break; case 'session.ended': this.dropped += this.queue.length; this.queue = []; this.isPaused = false; break; default: break; } }
push(frame: Buffer): void { this.queue.push(frame); }
/** The frames to play now: none while paused. */ drain(): Buffer[] { if (this.isPaused) return []; const frames = this.queue; this.queue = []; return frames; }}
/** Collects the reply audio, dropping what the server told us to drop. */export class ReplyAudio { private readonly chunks: Buffer[] = []; private readonly playback: PausePlayback; private newestSeq = -1; private droppedThrough = -1;
constructor(playback = new PausePlayback()) { this.playback = playback; }
/** A binary frame: an 8-byte header (kind, flags, sentence, reply_seq) then PCM16 24 kHz samples. */ push(frame: Buffer): void { if (frame.length < 8 || frame.readUInt8(0) !== 0x01) return; // not reply audio: ignore unknown kinds const seq = frame.readUInt32LE(4); this.newestSeq = Math.max(this.newestSeq, seq); if (seq <= this.droppedThrough) return; // a late frame of an interrupted reply this.playback.push(frame.subarray(8)); this.play(); }
/** Every event goes here first, so the playback buffer pauses, resumes and flushes with the voice. */ onEvent(event: { type: string; reply_id?: unknown }): void { this.playback.onEvent(event); this.play(); // a resume, a done or a new reply lets the held frames out }
/** `interrupted`: drop every later frame of the reply being played. (The playback buffer is flushed by `onEvent`.) */ interrupted(): void { this.droppedThrough = this.newestSeq; }
pcm(): Buffer { this.play(); return Buffer.concat(this.chunks); }
// A real player would hand these frames to the speaker; this example saves them. private play(): void { this.chunks.push(...this.playback.drain()); }}
/** The lines printed for an event: its type, plus the text of what was heard or said. */export function describeEvent(event: { type: string; text?: unknown }): string[] { const lines = [`event ${event.type}`]; if ((event.type === 'transcript.final' || event.type === 'reply.sentence') && typeof event.text === 'string') lines.push(` text: ${event.text}`); return lines;}
type ServerEvent = { type: string; [key: string]: unknown };
/** Runs one session: see the file header. Rejects with `KheelonaApiError`, `RelayError` or `UsageError`. */export async function runRelay(opts: RelayOptions): Promise<RelaySummary> { const log = opts.log ?? ((line: string) => { console.log(line); }); const replyTimeoutMs = opts.replyTimeoutMs ?? 20_000; if (!opts.text && !opts.wavFile) throw new UsageError('Give --text and/or --wav.'); if (opts.text && Array.from(opts.text).length > 2000) throw new UsageError('--text is at most 2000 characters'); let wavFrames: Buffer[] | undefined; if (opts.wavFile) { try { wavFrames = wavToFrames(readPcm16Wav(await readFile(opts.wavFile)).samples); } catch (error) { throw new UsageError(error instanceof Error && 'code' in error ? `Cannot read ${opts.wavFile}.` : (error as Error).message); } }
const session = await createSession({ apiUrl: opts.apiUrl, apiKey: opts.apiKey, voiceId: opts.voiceId, ...(opts.languages ? { languages: opts.languages } : {}), ...(opts.interruptions === undefined ? {} : { interruptions: opts.interruptions }), }); const ws = openSocket(session); const pause = opts.pausePlayback ?? true;
const audio = new ReplyAudio(new PausePlayback(pause)); const queue: Buffer[] = []; // microphone frames waiting to go out; silence fills the gaps let wavQueued = false; let turnsSent = 0; let turnsAnswered = 0; let replies = 0; let stopPacer = false; let pacer: Promise<void> | undefined; let timer: NodeJS.Timeout | undefined; let finished = false;
const summary = await new Promise<RelaySummary>((resolve, reject) => { const finish = (error?: Error, endReason = 'unknown'): void => { if (finished) return; finished = true; stopPacer = true; clearTimeout(timer); if (error instanceof RelayError) error.sessionId ??= session.sessionId; if (error) reject(error); else resolve({ sessionId: session.sessionId, endReason, replies, audioSeconds: audio.pcm().length / 2 / 24_000, outFile: opts.outFile }); }; // Nothing from the server for a while (waiting for session.ready, a reply or session.ended): give up, so a copy // of this never hangs. Every message from the server restarts the wait. const armTimer = (): void => { clearTimeout(timer); timer = setTimeout(() => { finish(new RelayError('no_reply', `Nothing from the server for ${String(replyTimeoutMs / 1000)} s`)); }, replyTimeoutMs); }; const sendEvent = (event: Record<string, unknown>): void => { ws.send(JSON.stringify(event)); }; const queueWav = (): void => { if (!wavFrames || wavQueued) return; wavQueued = true; turnsSent += 1; queue.push(...wavFrames); }; // Microphone audio at real time for the whole session: your turn's frames, then silence. const startPacer = (): void => { pacer = sendPaced({ next: () => { const frame = queue.shift(); if (frame) return frame; return SILENCE; }, send: (frame) => { if (ws.readyState === WebSocket.OPEN) ws.send(frame); }, stop: () => stopPacer, }); };
ws.once('open', armTimer); ws.on('message', (data: RawData, isBinary: boolean) => { armTimer(); if (isBinary) { if (Buffer.isBuffer(data)) audio.push(data); return; } let event: ServerEvent; try { event = JSON.parse((data as Buffer).toString('utf8')) as ServerEvent; } catch { return; } if (typeof event.type !== 'string') return; for (const line of describeEvent(event)) log(line); audio.onEvent(event);
switch (event.type) { case 'session.ready': // Nothing is sent before this: audio sent earlier is dropped. if (pause) sendEvent({ type: 'client.capabilities', playback_pause: true }); if (opts.text) { sendEvent({ type: 'input.text', text: opts.text }); turnsSent += 1; } else { queueWav(); } startPacer(); break; case 'interrupted': audio.interrupted(); break; case 'reply.done': replies += 1; if (event['reason'] === 'interrupted') break; // the turn goes on with a new reply turnsAnswered += 1; if (turnsAnswered < turnsSent) break; if (!wavQueued && wavFrames) { queueWav(); // the typed reply is done: now the WAV } else { stopPacer = true; void (pacer ?? Promise.resolve()).then(() => { sendEvent({ type: 'session.end' }); }); } break; case 'error': if (event['fatal'] === true) { finish( new RelayError(typeof event['code'] === 'string' ? event['code'] : 'error', 'The session failed'), ); } break; case 'session.ended': finish(undefined, typeof event['reason'] === 'string' ? event['reason'] : 'unknown'); break; default: break; } }); ws.on('unexpected-response', (_req, res) => { void readRefusal(res).then((error) => { finish(error); }); }); ws.on('close', () => { finish(new RelayError('disconnected', 'The connection closed before the session ended')); }); ws.on('error', () => { finish(new RelayError('connection_failed', 'The WebSocket failed')); }); }).finally(() => { ws.terminate(); });
await writeFile(opts.outFile, writeWav(audio.pcm(), 24_000)); return summary;}
/** The command line. Returns the exit code: 0 ended normally, 1 API or session error, 2 usage error. */export async function runCli( argv: string[], env: Record<string, string | undefined>, out: (line: string) => void, err: (line: string) => void, overrides: { replyTimeoutMs?: number } = {},): Promise<number> { try { const { values } = parseArgs({ args: argv, options: { text: { type: 'string' }, wav: { type: 'string' }, out: { type: 'string', default: 'reply.wav' }, voice: { type: 'string' }, languages: { type: 'string' }, 'no-interruptions': { type: 'boolean', default: false }, 'no-pause': { type: 'boolean', default: false }, }, strict: true, }); const settings = readEnv(env); const summary = await runRelay({ apiUrl: settings.apiUrl, apiKey: settings.apiKey, voiceId: values.voice ?? settings.voiceId, ...(values.languages ? { languages: values.languages.split(',').map((l) => l.trim()) } : {}), ...(values.text ? { text: values.text } : {}), ...(values.wav ? { wavFile: values.wav } : {}), outFile: values.out, ...(values['no-interruptions'] ? { interruptions: false } : {}), ...(values['no-pause'] ? { pausePlayback: false } : {}), log: out, ...(overrides.replyTimeoutMs === undefined ? {} : { replyTimeoutMs: overrides.replyTimeoutMs }), }); out(`session ${summary.sessionId} ended ${summary.endReason} replies ${String(summary.replies)} audio ${summary.audioSeconds.toFixed(1)}s saved ${summary.outFile}`); return 0; } catch (error) { if (error instanceof KheelonaApiError) { // No answer at all (status 0): name the failure, like the Python example. const code = error.status === 0 ? `api_unreachable_${error.causeName ?? 'Error'}` : error.code; err(`error ${code} request_id ${error.requestId ?? '-'}`); return 1; } if (error instanceof RelayError) { const session = error.sessionId ? ` session ${error.sessionId}` : ''; err(`error ${error.code}${session} request_id ${error.requestId ?? '-'}`); return 1; } // A usage mistake (bad flag, missing key, unreadable WAV): one line, exit 2. err(error instanceof Error ? error.message : 'Invalid usage.'); return 2; }}
if (process.argv[1] && pathToFileURL(process.argv[1]).href === import.meta.url) { process.exitCode = await runCli( process.argv.slice(2), process.env, (line) => { console.log(line); }, (line) => { console.error(line); }, );}Python: examples/python-server/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 thisexample writes nothing but the reply audio file you asked for."""
import argparseimport asyncioimport jsonimport osimport structimport sysimport timefrom collections.abc import Callable, Mapping, Sequencefrom dataclasses import dataclass
import httpxfrom websockets.asyncio.client import connectfrom websockets.exceptions import InvalidStatus, WebSocketException
from kheelona import KheelonaApiError, create_sessionfrom wavio import read_pcm16_wav, write_wav
FRAME_S = 0.020 # one frame every 20 msFRAME_BYTES = 640 # 20 ms of 16 kHz, 16-bit mono audioSILENCE = 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
@dataclassclass RelaySummary: session_id: str end_reason: str replies: int audio_seconds: float out_file: str
@dataclassclass 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())The examples are Kheelona’s own and are given to partners privately. Contact the Kheelona team for licensing.