Bidirectional streaming — the tier everyone else calls "realtime." Token-by-token text, streamed voice, barge-in, and tool calls over one persistent socket. On par with Gemini Live and GPT Realtime, on your own silicon.
The socket is wss://api.hawktalk.ai/v1/realtime — on a local node,
ws://HOST:8891/v1/realtime. It is an alias of /live/ouroboros; both
land on the same handler. If you are following an older page that says
/v1/stream, that route does not exist — it was never deployed under that name.
Three rules decide most of your porting work. JSON text frames only — binary frames are rejected by design, in both directions, so audio travels as base64 inside JSON. Input audio defaults to 24000 Hz PCM16 mono little-endian. Unknown frame types are not errors: log and ignore them, never crash a turn on one.
Auth has three carriages, because a browser cannot set a header on a WebSocket upgrade:
| form | how | use it when |
|---|---|---|
| header | Authorization: Bearer sk-... | server-side: Python, Node, Dart VM, Rust, Go. Preferred — the key never enters a URL or a log line. |
| query | ?api_key=sk-... | anything that cannot set headers. The key lands in proxy logs — treat it as short-lived. |
| subprotocol | openai-insecure-api-key.sk-... | browsers. Same name as the OpenAI convention so existing browser code ports unchanged. |
Pick the model at connect time with ?model=auto. Always start on
auto — it runs the router per utterance. Pinning a large tier for a whole
session is the single biggest source of overspend on this API. Do not hardcode registry ids
from any doc page, including this one: call GET /v1/models and read what the node
actually serves.
session.update also takes modalities (for example
["text","audio"]) and, next to tools, tool_choice.
Both are optional and both have defaults. Omit modalities and the node
answers in whatever the turn calls for; omit tool_choice and it behaves as
"auto" (see 02.7). The HawkTalkLive page spells them out to constrain a node —
that is the default made explicit, not a field that became required between the two pages.
# The shell tier. curl cannot hold a socket open; websocat can. websocat -H="Authorization: Bearer $HAWKTALK_API_KEY" \ 'wss://api.hawktalk.ai/v1/realtime?model=auto' # It prints session.created immediately, then prints whatever the server sends # in reply to the frames you paste. Paste this to re-point the live session # (one line, no newlines inside) — the server answers with session.updated: {"type":"session.update","session":{"model":"auto"}}
# arrives unprompted, immediately after the upgrade: {"type":"session.created","session":{"id":"sess_9f2c1a","model":"auto"}} # arrives only in response to YOUR session.update frame: {"type":"session.updated","session":{"id":"sess_9f2c1a","model":"auto"}} # Node builds differ in what else rides in `session`. Read `type` and the keys you # need; keep the rest as an opaque map. Never assert on the full object shape.
# pip install "websockets>=14" import asyncio, json, os, websockets KEY = os.environ["HAWKTALK_API_KEY"] URL = "wss://api.hawktalk.ai/v1/realtime?model=auto" async def main(): # websockets >= 14 calls this additional_headers. On 12.x and 13.x the # top-level websockets.connect still takes extra_headers — pin the version # or branch, because passing the wrong one raises TypeError at connect. # max_size matters: base64 audio deltas make frames far larger than text ones. async with websockets.connect( URL, additional_headers={"Authorization": f"Bearer {KEY}"}, max_size=16 * 1024 * 1024, ) as ws: hello = json.loads(await ws.recv()) assert hello["type"] == "session.created", hello await ws.send(json.dumps({"type": "session.update", "session": {"model": "auto"}})) print(json.loads(await ws.recv())) # session.updated asyncio.run(main())
// npm i ws import WebSocket from "ws"; const key = process.env.HAWKTALK_API_KEY!; const ws = new WebSocket("wss://api.hawktalk.ai/v1/realtime?model=auto", { headers: { Authorization: `Bearer ${key}` }, maxPayload: 16 * 1024 * 1024, }); ws.on("message", (buf: Buffer, isBinary: boolean) => { // The server never sends binary. Drop it rather than trusting it. if (isBinary) return; const e = JSON.parse(buf.toString("utf8")); if (e.type === "session.created") { ws.send(JSON.stringify({ type: "session.update", session: { model: "auto" } })); } if (e.type === "session.updated") console.log("model:", e.session?.model ?? "unknown"); if (e.type === "error") console.error(e.error); }); ws.on("error", (err) => console.error("socket", err));
// Fetch the key from YOUR backend. A key pasted into a browser bundle is a // published key. There is no ephemeral-token mint on this API today — proxy or // hand out a short-lived key you can revoke. // .trim() is not optional: a trailing newline makes the subprotocol token // invalid and the upgrade fails with an opaque browser error. const key = (await fetch("/api/hawktalk-key").then(r => r.text())).trim(); const ws = new WebSocket( "wss://api.hawktalk.ai/v1/realtime?model=auto", ["openai-insecure-api-key." + key], // subprotocol auth ); ws.onopen = () => console.log("open"); ws.onmessage = (ev) => { if (typeof ev.data !== "string") return; // text frames only const e = JSON.parse(ev.data); if (e.type === "session.created") ws.send(JSON.stringify({ type: "session.update", session: { model: "auto" } })); };
// pubspec: web_socket_channel: ^3.0.0 // NOTE: `io.dart` pulls in dart:io and will NOT compile for Flutter web. In a // cross-platform app, put the two branches behind a conditional import // (`import 'connect_io.dart' if (dart.library.html) 'connect_web.dart';`) // rather than a runtime flag like the one below. import 'dart:convert'; import 'dart:io'; // Platform.environment import 'package:web_socket_channel/web_socket_channel.dart'; import 'package:web_socket_channel/io.dart'; const base = 'wss://api.hawktalk.ai/v1/realtime'; WebSocketChannel connect(String key, {bool isWeb = false}) { final uri = Uri.parse('$base?model=auto'); // Flutter web cannot set headers on the upgrade -> subprotocol. // Mobile/desktop/server -> header, which keeps the key out of the URL. return isWeb ? WebSocketChannel.connect(uri, protocols: ['openai-insecure-api-key.$key']) : IOWebSocketChannel.connect(uri, headers: {'Authorization': 'Bearer $key'}); } void main() { final ch = connect(Platform.environment['HAWKTALK_API_KEY']!); ch.stream.listen((raw) { if (raw is! String) return; // binary is never ours final e = jsonDecode(raw) as Map<String, dynamic>; if (e['type'] == 'session.created') { ch.sink.add(jsonEncode({'type': 'session.update', 'session': {'model': 'auto'}})); } }, onError: (e) => print('socket: $e')); }
// Cargo.toml: tokio = { version = "1", features = ["full"] } // tokio-tungstenite = { version = "0.24", features = ["native-tls"] } // futures-util = "0.3" serde_json = "1" use futures_util::{SinkExt, StreamExt}; use tokio_tungstenite::{connect_async, tungstenite::{client::IntoClientRequest, Message}}; #[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> { let key = std::env::var("HAWKTALK_API_KEY")?; let mut req = "wss://api.hawktalk.ai/v1/realtime?model=auto".into_client_request()?; req.headers_mut().insert("Authorization", format!("Bearer {key}").parse()?); let (mut ws, _resp) = connect_async(req).await?; while let Some(msg) = ws.next().await { match msg? { Message::Text(t) => { let e: serde_json::Value = serde_json::from_str(&t)?; match e["type"].as_str().unwrap_or("unknown") { "session.created" => ws.send(Message::Text( r#"{"type":"session.update","session":{"model":"auto"}}"#.into())).await?, "session.updated" => println!("model: {}", e["session"]["model"]), _ => {} // unknown frames are not fatal } } Message::Binary(_) => continue, // rejected by design; ignore Message::Close(_) => break, _ => {} } } Ok(()) }
// go get github.com/gorilla/websocket package main import ( "encoding/json" "log" "net/http" "os" "github.com/gorilla/websocket" ) func main() { h := http.Header{} h.Set("Authorization", "Bearer "+os.Getenv("HAWKTALK_API_KEY")) c, resp, err := websocket.DefaultDialer.Dial( "wss://api.hawktalk.ai/v1/realtime?model=auto", h) if err != nil { if resp != nil { // 401 here means the key, not the network log.Fatalf("dial: %v (http %d)", err, resp.StatusCode) } log.Fatalf("dial: %v", err) } defer c.Close() for { typ, data, err := c.ReadMessage() if err != nil { log.Printf("read: %v", err); return } if typ != websocket.TextMessage { continue } var e map[string]any if err := json.Unmarshal(data, &e); err != nil { continue } if e["type"] == "session.created" { if err := c.WriteJSON(map[string]any{ "type": "session.update", "session": map[string]any{"model": "auto"}, }); err != nil { log.Fatalf("write: %v", err) } } } }
response.doneSend a message with conversation.item.create, ask for a turn with
response.create, then consume response.text.delta until
response.done. Here is a whole text turn on the wire, unedited:
// -> client sends {"type":"conversation.item.create","item":{"type":"message","role":"user", "content":[{"type":"input_text","text":"Name the highest tide on earth."}]} {"type":"response.create"} // <- server sends {"type":"conversation.item.created","item":{"id":"item_31","role":"user"}} {"type":"response.created","response":{"id":"resp_7c4"}} {"type":"response.output_item.added","item":{"id":"item_32","type":"message"}} {"type":"response.content_part.added","part":{"type":"text"}} {"type":"response.text.delta","delta":"The Bay of Fundy"} {"type":"ouroboros.telemetry", /* node-specific counters */} {"type":"response.text.delta","delta":", about 16 m."} {"type":"response.text.done","text":"The Bay of Fundy, about 16 m."} {"type":"response.content_part.done"} {"type":"response.output_item.done","item":{"id":"item_32","type":"message"}} {"type":"response.done","response":{"id":"resp_7c4","status":"completed"}, "x_ouroboros":{"stt_ms":null,"ttft_ms":184,"first_audio_out_ms":null, "end_to_end_ms":1042,"end_to_end_audio_ms":null}}
| x_ouroboros field | meaning | null when |
|---|---|---|
| stt_ms | speech-to-text time for the committed input buffer | the turn had no audio input, or the STT seam is not wired |
| ttft_ms | time to first text token | the turn produced no text |
| first_audio_out_ms | time to the first response.audio.delta | text-only turn, or the TTS seam is not wired |
| end_to_end_ms | commit/create to response.done | never, on a completed turn |
| end_to_end_audio_ms | commit to last audio byte out | no audio was produced |
null means unknown and nothing else. This API never fabricates a zero to fill a
dashboard. If you chart these, render null as a gap — averaging nulls-as-zero is how a p50
latency graph ends up lying to you. This is the socket's equivalent of the REST tier's
x_timing; the field sets differ, the null-means-unmeasured rule does not.
response.id is the join key here: it appears on
response.created, is echoed on response.done, and is what you tag your
own log lines with to stitch a turn together.
ouroboros.telemetry arrives every 8 tokens and as a 5 s idle heartbeat. Treat it
as your liveness signal (see 02.8); its payload varies by node, so log it, do not parse it blind.
rate_limits.updated can arrive at any point in a turn.
Other HawkTalk frames you will see and should ignore. On
/v1/realtime the extra frames to expect are ouroboros.telemetry,
ouroboros.endpoint and ouroboros.endpoint.check. None of them is
required to complete a turn and none of them is a malfunction in your parser — log and continue.
ouroboros.user_affect, ouroboros.presence, hawk.conduct
and hawk.brain.hello are preview-mux frames: they belong to the
HawkTalkLive lane mux and you may see them on a local node that runs it, not on the shipped
realtime socket. Either way the contract is the one you apply to any frame type you have never
heard of — log it and carry on.
import asyncio, json, os, websockets async def ask(ws, text): await ws.send(json.dumps({"type": "conversation.item.create", "item": { "type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}})) await ws.send(json.dumps({"type": "response.create"})) async def main(): url = "wss://api.hawktalk.ai/v1/realtime?model=auto" hdr = {"Authorization": f"Bearer {os.environ['HAWKTALK_API_KEY']}"} async with websockets.connect(url, additional_headers=hdr) as ws: await ask(ws, "Name the highest tide on earth.") async for raw in ws: e = json.loads(raw) t = e.get("type") if t == "response.text.delta": print(e.get("delta") or e.get("text", ""), end="", flush=True) elif t == "error": raise RuntimeError(e["error"]) # {message,type,code,param} elif t == "response.done": # x_ouroboros has been seen at the frame root and under `response`. # Read both; a migration that reads only one silently loses timings. x = e.get("x_ouroboros") or e.get("response", {}).get("x_ouroboros", {}) print(f"\nttft={x.get('ttft_ms')}ms e2e={x.get('end_to_end_ms')}ms") break asyncio.run(main())
import WebSocket from "ws"; const ws = new WebSocket("wss://api.hawktalk.ai/v1/realtime?model=auto", { headers: { Authorization: `Bearer ${process.env.HAWKTALK_API_KEY}` } }); const send = (o: unknown) => ws.send(JSON.stringify(o)); let answer = ""; ws.on("open", () => { send({ type: "conversation.item.create", item: { type: "message", role: "user", content: [{ type: "input_text", text: "Name the highest tide on earth." }] } }); send({ type: "response.create" }); }); ws.on("message", (buf: Buffer, isBinary: boolean) => { if (isBinary) return; const e = JSON.parse(buf.toString("utf8")); switch (e.type) { case "response.text.delta": answer += e.delta ?? e.text ?? ""; process.stdout.write(e.delta ?? ""); break; case "error": console.error("api error", e.error); ws.close(); break; case "response.done": { const x = e.x_ouroboros ?? e.response?.x_ouroboros ?? {}; // null is "not measured" — do not coerce with ?? 0 for a metric. console.log(`\nttft=${x.ttft_ms} e2e=${x.end_to_end_ms}`); ws.close(); break; } } });
import 'dart:convert'; import 'package:web_socket_channel/io.dart'; Future<void> ask(String key, String prompt) async { final ch = IOWebSocketChannel.connect( Uri.parse('wss://api.hawktalk.ai/v1/realtime?model=auto'), headers: {'Authorization': 'Bearer $key'}); ch.sink.add(jsonEncode({'type': 'conversation.item.create', 'item': { 'type': 'message', 'role': 'user', 'content': [{'type': 'input_text', 'text': prompt}]}})); ch.sink.add(jsonEncode({'type': 'response.create'})); final buf = StringBuffer(); await for (final raw in ch.stream) { if (raw is! String) continue; final e = jsonDecode(raw) as Map<String, dynamic>; switch (e['type']) { case 'response.text.delta': buf.write(e['delta'] ?? e['text'] ?? ''); case 'error': await ch.sink.close(); throw StateError('${e['error']}'); case 'response.done': final x = (e['x_ouroboros'] ?? (e['response'] as Map?)?['x_ouroboros'] ?? {}) as Map; print('$buf\nttft=${x['ttft_ms']} e2e=${x['end_to_end_ms']}'); await ch.sink.close(); return; // stop reading; the turn is over } } }
use futures_util::{SinkExt, StreamExt}; use serde_json::{json, Value}; use tokio_tungstenite::{connect_async, tungstenite::{client::IntoClientRequest, Message}}; #[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> { let key = std::env::var("HAWKTALK_API_KEY")?; let mut req = "wss://api.hawktalk.ai/v1/realtime?model=auto".into_client_request()?; req.headers_mut().insert("Authorization", format!("Bearer {key}").parse()?); let (mut ws, _) = connect_async(req).await?; ws.send(Message::Text(json!({ "type": "conversation.item.create", "item": {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "Name the highest tide on earth."}]} }).to_string())).await?; ws.send(Message::Text(r#"{"type":"response.create"}"#.into())).await?; let mut out = String::new(); while let Some(msg) = ws.next().await { let Message::Text(t) = msg? else { continue }; let e: Value = serde_json::from_str(&t)?; match e["type"].as_str().unwrap_or_default() { "response.text.delta" => out.push_str(e["delta"].as_str().unwrap_or_default()), "error" => return Err(format!("api: {}", e["error"]).into()), "response.done" => { let x = if e["x_ouroboros"].is_object() { &e["x_ouroboros"] } else { &e["response"]["x_ouroboros"] }; // Value::Null prints as `null` — that is the honest render. println!("{out}\nttft={} e2e={}", x["ttft_ms"], x["end_to_end_ms"]); break; } _ => {} } } Ok(()) }
// imports: encoding/json, fmt, strings, github.com/gorilla/websocket type ouroTimings struct { // Pointers, so a missing timing stays nil instead of becoming 0. TTFTMs *int `json:"ttft_ms"` E2EMs *int `json:"end_to_end_ms"` } func askOnce(c *websocket.Conn, prompt string) error { if err := c.WriteJSON(map[string]any{ "type": "conversation.item.create", "item": map[string]any{"type": "message", "role": "user", "content": []any{map[string]any{"type": "input_text", "text": prompt}}}, }); err != nil { return err } if err := c.WriteJSON(map[string]any{"type": "response.create"}); err != nil { return err } var sb strings.Builder for { typ, data, err := c.ReadMessage() if err != nil { return err } if typ != websocket.TextMessage { continue } var e struct { Type string `json:"type"` Delta string `json:"delta"` Error json.RawMessage `json:"error"` // Read BOTH placements. Reading only the root silently loses // every timing on nodes that nest it under `response`. X *ouroTimings `json:"x_ouroboros"` Response *struct { X *ouroTimings `json:"x_ouroboros"` } `json:"response"` } if err := json.Unmarshal(data, &e); err != nil { continue } switch e.Type { case "response.text.delta": sb.WriteString(e.Delta) case "error": return fmt.Errorf("api: %s", e.Error) case "response.done": fmt.Println(sb.String()) x := e.X if x == nil && e.Response != nil { x = e.Response.X } if x != nil && x.TTFTMs != nil { fmt.Println("ttft_ms", *x.TTFTMs) } else { fmt.Println("ttft_ms unknown") // not 0. never 0. } return nil } } }
Mic audio goes up as base64 PCM16 mono, 24000 Hz by default, in
input_audio_buffer.append frames. The buffer becomes a turn only when you send
input_audio_buffer.commit.
End-of-speech is the client's job. Full stop. This server does semantic endpointing
only and never emits speech_started or speech_stopped. If you
are porting from an API with server VAD, that listener is dead code here — delete it and run
VAD locally (energy gate + hangover, WebRTC VAD, Silero, whatever you already trust). Nothing
happens until you commit. Latency consequence: your hangover timer is the user's
perceived response time, so tune it (150–400 ms is the usual band) rather than blaming
ttft_ms.
Send 20–40 ms per append. Smaller frames burn JSON overhead — base64 already costs 33% —
and larger ones add a floor to stt_ms.
{"type":"input_audio_buffer.append","audio":"<base64 pcm16 @24000>"}
{"type":"input_audio_buffer.append","audio":"<base64 pcm16 @24000>"}
{"type":"input_audio_buffer.commit"}
{"type":"response.create"}
# then, when the STT seam is wired on this node:
{"type":"conversation.item.input_audio_transcription.completed",
"item_id":"item_44","transcript":"how high is the tide today"}
# and when it is NOT wired, you get this instead — never a guessed transcript:
{"type":"error","error":{"message":"stt seam not wired on this node",
"type":"server_error","code":"stt_not_wired","param":null}
Seam honesty. stt_not_wired (REST equivalent: HTTP 501) is a deliberate,
documented outcome on nodes where the seam is absent. It exists so your client falls back to
browser Web Speech or a local recogniser. You will never receive an invented transcript to
cover for it.
# pip install sounddevice websockets import asyncio, base64, json, os, queue, sounddevice as sd, websockets RATE, BLOCK = 24000, 480 # 480 frames = 20 ms at 24 kHz q: queue.Queue = queue.Queue() def on_block(indata, frames, t, status): q.put(bytes(indata)) # int16 mono, little-endian, already wire-ready async def talk(ws, speaking: asyncio.Event): with sd.RawInputStream(samplerate=RATE, blocksize=BLOCK, channels=1, dtype="int16", callback=on_block): while speaking.is_set(): pcm = await asyncio.get_running_loop().run_in_executor(None, q.get) await ws.send(json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(pcm).decode()})) # YOUR vad cleared `speaking`. The server will not do this for you. await ws.send(json.dumps({"type": "input_audio_buffer.commit"})) await ws.send(json.dumps({"type": "response.create"}))
// `ws` is the socket you opened in 02.1. // AudioContext({sampleRate: 24000}) makes the browser resample for you, so you // never ship a 48 kHz buffer the server will read as chipmunk speech. const ctx = new AudioContext({ sampleRate: 24000 }); const src = ctx.createMediaStreamSource( await navigator.mediaDevices.getUserMedia({ audio: { channelCount: 1, echoCancellation: true, noiseSuppression: true } })); const CAPTURE = ` class Cap extends AudioWorkletProcessor { process(inputs) { const ch = inputs[0][0]; if (!ch) return true; const pcm = new Int16Array(ch.length); for (let i = 0; i < ch.length; i++) { const s = Math.max(-1, Math.min(1, ch[i])); pcm[i] = s < 0 ? s * 0x8000 : s * 0x7fff; } this.port.postMessage(pcm.buffer, [pcm.buffer]); return true; } } registerProcessor('cap', Cap);`; await ctx.audioWorklet.addModule( URL.createObjectURL(new Blob([CAPTURE], { type: "application/javascript" }))); const node = new AudioWorkletNode(ctx, "cap"); src.connect(node); function b64(buf: ArrayBuffer): string { const b = new Uint8Array(buf); let s = ""; for (let i = 0; i < b.length; i += 0x8000) s += String.fromCharCode(...b.subarray(i, i + 0x8000)); return btoa(s); } let speaking = false; // driven by YOUR vad node.port.onmessage = (ev) => { if (!speaking || ws.readyState !== WebSocket.OPEN) return; ws.send(JSON.stringify({ type: "input_audio_buffer.append", audio: b64(ev.data) })); }; export function endOfSpeech() { // call this from your vad's hangover timer if (!speaking) return; speaking = false; ws.send(JSON.stringify({ type: "input_audio_buffer.commit" })); ws.send(JSON.stringify({ type: "response.create" })); }
// pubspec: record: ^5.1.0 (AudioRecorder streams raw PCM16 — no file, no codec) import 'dart:async'; import 'dart:convert'; import 'package:record/record.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; class MicUplink { MicUplink(this.ch); final WebSocketChannel ch; final _rec = AudioRecorder(); StreamSubscription<List<int>>? _sub; bool speaking = false; /// Start ONCE per call, not once per turn. Tearing the recorder down at /// every commit costs a device restart (~100-300 ms) on the next turn and /// makes barge-in impossible, because the mic is closed when the user cuts in. Future<void> start() async { if (!await _rec.hasPermission()) throw StateError('mic denied'); final stream = await _rec.startStream(const RecordConfig( encoder: AudioEncoder.pcm16bits, // raw PCM16 — what the wire wants sampleRate: 24000, // server input default numChannels: 1, )); _sub = stream.listen((chunk) { if (!speaking) return; // dropped locally, never committed ch.sink.add(jsonEncode({ 'type': 'input_audio_buffer.append', 'audio': base64Encode(chunk), })); }); } /// Call from YOUR vad. The server never tells you speech stopped. /// The mic keeps running; only the appending stops. void commit() { if (!speaking) return; speaking = false; ch.sink.add(jsonEncode({'type': 'input_audio_buffer.commit'})); ch.sink.add(jsonEncode({'type': 'response.create'})); } /// End of the whole call, not the end of a turn. Future<void> dispose() async { speaking = false; await _sub?.cancel(); await _rec.stop(); } }
// Feed raw PCM16 in on stdin so this example has no audio-backend dependency: // arecord -f S16_LE -c1 -r24000 -t raw | cargo run use base64::{engine::general_purpose::STANDARD, Engine}; use futures_util::SinkExt; use tokio::io::{stdin, AsyncReadExt}; use tokio_tungstenite::tungstenite::Message; async fn uplink<S>(ws: &mut S) -> Result<(), Box<dyn std::error::Error>> where S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin, { let mut input = stdin(); let mut buf = [0u8; 960]; // 480 samples = 20 ms @ 24 kHz loop { let n = input.read(&mut buf).await?; if n == 0 { break } // your vad decides this in a real app let b64 = STANDARD.encode(&buf[..n]); let frame = serde_json::json!({ "type": "input_audio_buffer.append", "audio": b64 }); ws.send(Message::Text(frame.to_string())).await?; } ws.send(Message::Text(r#"{"type":"input_audio_buffer.commit"}"#.into())).await?; ws.send(Message::Text(r#"{"type":"response.create"}"#.into())).await?; Ok(()) }
// arecord -f S16_LE -c1 -r24000 -t raw | go run ./uplink func uplink(c *websocket.Conn, r io.Reader, stop <-chan struct{}) error { buf := make([]byte, 960) // 20 ms @ 24 kHz mono pcm16 for { select { case <-stop: // YOUR vad fired end-of-speech; the server never will if err := c.WriteJSON(map[string]any{ "type": "input_audio_buffer.commit"}); err != nil { return err } return c.WriteJSON(map[string]any{"type": "response.create"}) default: } n, err := io.ReadFull(r, buf) if n > 0 { if werr := c.WriteJSON(map[string]any{ "type": "input_audio_buffer.append", "audio": base64.StdEncoding.EncodeToString(buf[:n]), }); werr != nil { return werr } } if err != nil { return err } // EOF / ErrUnexpectedEOF after the flush } }
Spoken output arrives as response.audio.delta. Read the sample rate off the
frame — output frames carry x_sample_rate_hz, and it is commonly 24000 while
your mic ran at some other rate. Hardcoding the playback rate is the classic chipmunk bug.
{"type":"response.audio.delta","delta":"<base64 pcm16>","x_sample_rate_hz":24000}
{"type":"response.viseme.delta","viseme":"AA"}
{"type":"response.audio.delta","delta":"<base64 pcm16>","x_sample_rate_hz":24000}
{"type":"response.audio.done"}
# On a node without the TTS seam you get this instead of audio — and you should
# fall back to browser speech synthesis, not to silence and not to fake audio:
{"type":"error","error":{"message":"tts seam not wired on this node",
"type":"server_error","code":"tts_not_wired","param":null}
Some builds carry the base64 under delta, some under audio. Read
delta ?? audio and you are correct on both.
response.viseme.delta is a HawkTalk extension for lip-sync — it rides the audio
channel, interleaved with the deltas, so drive your mouth shape from it in arrival order.
The event name is published; its payload keys are not. The nodes we have tested put the
mouth shape under viseme; code it so that a frame with no key you recognise is
skipped rather than thrown on, and your lip-sync degrades to "mouth closed" instead of a crash.
import base64, sounddevice as sd _stream = None _rate = None def _out(rate): # Reopen if the node changes rate mid-stream. One stream pinned to the # first frame's rate is the same chipmunk bug, one level down. global _stream, _rate if _stream is None or _rate != rate: if _stream is not None: _stream.close() _stream = sd.RawOutputStream(samplerate=rate, channels=1, dtype="int16") _stream.start() _rate = rate return _stream def on_frame(e, local_tts=None, last_text=""): """local_tts: your on-device speech function, e.g. pyttsx3.say.""" t = e.get("type") if t == "response.audio.delta": pcm = base64.b64decode(e.get("delta") or e.get("audio") or "") rate = e.get("x_sample_rate_hz") or 24000 # trust the frame first _out(rate).write(pcm) elif t == "response.viseme.delta": print("viseme", e.get("viseme")) elif t == "error" and e["error"].get("code") == "tts_not_wired": # Seam absent on this node. Speak locally; never synthesise a lie. if local_tts is not None: local_tts(last_text)
const out = new AudioContext(); let playhead = 0; let queued: AudioBufferSourceNode[] = []; let carry: Uint8Array = new Uint8Array(0); // odd-length chunk guard export function onAudioDelta(e: any) { const bin = atob(e.delta ?? e.audio ?? ""); let bytes = new Uint8Array(carry.length + bin.length); bytes.set(carry, 0); for (let i = 0; i < bin.length; i++) bytes[carry.length + i] = bin.charCodeAt(i); if (bytes.length % 2) { // keep the half sample for next time carry = bytes.slice(bytes.length - 1); bytes = bytes.slice(0, bytes.length - 1); } else carry = new Uint8Array(0); if (!bytes.length) return; const rate = e.x_sample_rate_hz ?? 24000; const pcm = new Int16Array(bytes.buffer, bytes.byteOffset, bytes.length / 2); const buf = out.createBuffer(1, pcm.length, rate); const ch = buf.getChannelData(0); for (let i = 0; i < pcm.length; i++) ch[i] = pcm[i] / 0x8000; const node = out.createBufferSource(); node.buffer = buf; node.connect(out.destination); const at = Math.max(out.currentTime, playhead); node.start(at); playhead = at + buf.duration; queued.push(node); // keep handles: barge-in needs them node.onended = () => { queued = queued.filter(n => n !== node); }; } export function onViseme(e: any) { if (typeof e.viseme !== "string") return; // unrecognised payload: skip document.querySelector("#mouth")?.setAttribute("data-viseme", e.viseme); }
// pubspec: audioplayers: ^6.0.0 import 'dart:async'; import 'dart:convert'; import 'dart:typed_data'; import 'package:audioplayers/audioplayers.dart'; Uint8List wavPcm16(Uint8List pcm, int rate) { final h = ByteData(44); h.setUint32(0, 0x52494646, Endian.big); // "RIFF" h.setUint32(4, 36 + pcm.length, Endian.little); h.setUint32(8, 0x57415645, Endian.big); // "WAVE" h.setUint32(12, 0x666d7420, Endian.big); // "fmt " h.setUint32(16, 16, Endian.little); h.setUint16(20, 1, Endian.little); // pcm h.setUint16(22, 1, Endian.little); // mono h.setUint32(24, rate, Endian.little); h.setUint32(28, rate * 2, Endian.little); // byte rate: mono, 2 bytes h.setUint16(32, 2, Endian.little); // block align h.setUint16(34, 16, Endian.little); // bits per sample h.setUint32(36, 0x64617461, Endian.big); // "data" h.setUint32(40, pcm.length, Endian.little); return Uint8List.fromList([...h.buffer.asUint8List(), ...pcm]); } class Playout { final _player = AudioPlayer(); final _pcm = BytesBuilder(); final visemes = StreamController<String>.broadcast(); int _rate = 24000; void onFrame(Map<String, dynamic> e) { switch (e['type']) { case 'response.audio.delta': _rate = (e['x_sample_rate_hz'] as int?) ?? _rate; final b64 = (e['delta'] ?? e['audio']) as String?; if (b64 != null) _pcm.add(base64Decode(b64)); case 'response.viseme.delta': final v = e['viseme']; if (v is String) visemes.add(v); // unrecognised payload: skip case 'response.audio.done': // Buffering to audio.done costs you the whole utterance in latency. // Ship a real PCM sink (platform channel / flutter_pcm_sound) for // chunked playout; this is the dependency-free version. final pcm = _pcm.takeBytes(); if (pcm.isNotEmpty) _player.play(BytesSource(wavPcm16(pcm, _rate))); } } Future<void> stop() async { _pcm.clear(); await _player.stop(); } }
// Raw PCM to stdout: cargo run | aplay -f S16_LE -c1 -r24000 -t raw use base64::{engine::general_purpose::STANDARD, Engine}; use std::io::Write; fn on_frame(e: &serde_json::Value, out: &mut impl Write) -> Result<(), Box<dyn std::error::Error>> { match e["type"].as_str().unwrap_or_default() { "response.audio.delta" => { let b64 = e["delta"].as_str().or_else(|| e["audio"].as_str()) .ok_or("audio delta with no payload")?; let rate = e["x_sample_rate_hz"].as_u64().unwrap_or(24000); eprintln!("pcm16 @ {rate} Hz"); // resample if your sink differs out.write_all(&STANDARD.decode(b64)?)?; out.flush()?; } "response.viseme.delta" => { if let Some(v) = e["viseme"].as_str() { eprintln!("viseme {v}") } } "error" if e["error"]["code"] == "tts_not_wired" => eprintln!("tts seam absent on this node — falling back to local speech"), _ => {} } Ok(()) }
// go run ./downlink | aplay -f S16_LE -c1 -r24000 -t raw type audioFrame struct { Type string `json:"type"` Delta string `json:"delta"` Audio string `json:"audio"` Rate *int `json:"x_sample_rate_hz"` Vis string `json:"viseme"` } func handleAudio(data []byte, w io.Writer) error { var f audioFrame if err := json.Unmarshal(data, &f); err != nil { return nil } // not fatal switch f.Type { case "response.audio.delta": b64 := f.Delta if b64 == "" { b64 = f.Audio } if b64 == "" { return nil } pcm, err := base64.StdEncoding.DecodeString(b64) if err != nil { return err } rate := 24000 if f.Rate != nil { rate = *f.Rate } // the frame is the authority log.Printf("pcm16 @ %d Hz, %d bytes", rate, len(pcm)) _, err = w.Write(pcm) return err case "response.viseme.delta": if f.Vis != "" { log.Printf("viseme %s", f.Vis) } } return nil }
Cancelling on the server is the easy half. What makes barge-in feel instant is dropping
the audio you have already queued locally — by the time the user starts talking, one or two
seconds of the model's speech is typically already in your playout buffer or scheduled on the
audio clock. The network round-trip for response.cancel is 30–200 ms; your local
flush is 0 ms. Do it in that order: stop the speaker, drop the queue, then send
response.cancel.
Do not stop capturing. Barge-in means the opposite: keep the microphone open and keep appending, because the user is mid-utterance and that audio is the next turn. Audio you decide not to use is simply never appended, and an uncommitted buffer is not a turn — so there is nothing to undo.
response.cancel rolls conversation history back to the turn boundary.
This is a real difference from the API you are migrating from: you do not have to send a
truncate frame with a measured audio_end_ms to keep the model's memory honest.
After a cancel, the model does not believe it said the words the user never heard.
{"type":"response.cancel"}
export function bargeIn(ws: WebSocket) { // 1. Local flush — this is the half the user actually perceives. for (const n of queued) { try { n.stop(); } catch { /* already ended */ } } queued = []; playhead = 0; carry = new Uint8Array(0); // 2. Server-side cancel. History rolls back to the turn boundary. ws.send(JSON.stringify({ type: "response.cancel" })); // 3. Keep the mic hot — the interrupting speech IS the next turn. speaking = true; }
Future<void> bargeIn(WebSocketChannel ch, Playout playout, MicUplink mic) async { await playout.stop(); // speaker silent immediately ch.sink.add(jsonEncode({'type': 'response.cancel'})); // The recorder is still running (MicUplink.commit never stopped it), so // flipping this flag resumes appending on the very next chunk. mic.speaking = true; }
async def barge_in(ws, playout): """playout: the sd.RawOutputStream you opened on the first audio delta. Returns None so the caller can drop its handle; the next audio delta reopens a stream at whatever rate that frame declares. """ if playout is not None: playout.abort() # abort(), not stop(): stop() drains the buffer first playout.close() await ws.send(json.dumps({"type": "response.cancel"})) return None
// PlayoutQueue::flush takes &self and drops its chunks behind a Mutex — // barge-in is called from the read task, not the audio task. async fn barge_in<S>(ws: &mut S, playout: &PlayoutQueue) -> Result<(), tokio_tungstenite::tungstenite::Error> where S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin, { playout.flush(); // drop every queued PCM chunk before the network call ws.send(Message::Text(r#"{"type":"response.cancel"}"#.into())).await }
func bargeIn(c *websocket.Conn, playout *Playout) error { playout.Flush() // local first: this is what makes it feel instant return c.WriteJSON(map[string]any{"type": "response.cancel"}) }
Every conglomerate realtime API is one model per session: to change model you tear down the socket and rebuild the conversation. Here you re-point a live socket with one frame, mid-call, and the conversation continues.
Three levels of selection, in increasing precedence — each accepts a registry id, a tier pin,
or "auto":
| level | frame | scope |
|---|---|---|
| connection | ?model=auto | the session's default |
| session | session.update → session.model | every turn from now on |
| turn | response.create → response.model | this turn only; session default untouched |
Tier vocabulary: ouro · quick · dank ·
think · cloud (aliases self, route,
ouromega, live, specialist, t0–t3).
Pin with "tier:dank". "auto" runs the router per utterance and is
what you should ship; escalate deliberately and drop back immediately, because a session
left pinned to a big tier bills at that tier for every "mm-hm" the user says. Registry ids
differ per node — read them from GET /v1/models, never from a doc page.
# paste these one at a time into a live socket {"type":"session.update","session":{"model":"tier:quick"}} # cheap chit-chat {"type":"session.update","session":{"model":"tier:think"}} # hard question {"type":"session.update","session":{"model":"auto"}} # hand it back # one turn only, session default untouched: {"type":"response.create","response":{"model":"tier:dank"}}
async def set_model(ws, model: str): """Re-point the live session. The socket, the audio and the history all survive.""" await ws.send(json.dumps({"type": "session.update", "session": {"model": model}})) async def escalate_for_one_turn(ws, prompt: str): # Per-turn override beats the session default and expires with the turn — # the safest way to spend big money exactly once. await ws.send(json.dumps({"type": "conversation.item.create", "item": { "type": "message", "role": "user", "content": [{"type": "input_text", "text": prompt}]}})) await ws.send(json.dumps({"type": "response.create", "response": {"model": "tier:think"}})) async def arc(ws): # typical arc: start cheap, escalate on demand, drop straight back await set_model(ws, "tier:quick") await escalate_for_one_turn(ws, "Work the tide table; tell me when to launch.") await set_model(ws, "auto")
const setModel = (m: string) => ws.send(JSON.stringify({ type: "session.update", session: { model: m } })); const turnWith = (m: string) => ws.send(JSON.stringify({ type: "response.create", response: { model: m } })); setModel("tier:quick"); // greeting + small talk on the cheap tier turnWith("tier:think"); // this one question gets the expensive brain setModel("auto"); // back to the router; the call never dropped ws.on("message", (b: Buffer, isBinary: boolean) => { if (isBinary) return; const e = JSON.parse(b.toString("utf8")); // session.updated echoes the effective model — log it, don't assume it. if (e.type === "session.updated") console.log("now:", e.session?.model ?? "unknown"); // A bad id is 404-class: model_not_found. Fall back to "auto", don't retry. if (e.type === "error" && e.error?.code === "model_not_found") setModel("auto"); });
extension ModelSwitch on WebSocketChannel { /// Re-points the live session. No teardown, no lost history, no dropped audio. void setModel(String model) => sink.add(jsonEncode({'type': 'session.update', 'session': {'model': model}})); /// One-turn override; the session default is untouched. void respondWith(String model) => sink.add(jsonEncode({'type': 'response.create', 'response': {'model': model}})); } // In a Flutter call screen: the user taps "think harder" mid-conversation. void onThinkHarder(WebSocketChannel ch) { ch.respondWith('tier:think'); // spends once // no ch.setModel here — leaving the session pinned is how bills happen }
async fn set_model<S>(ws: &mut S, model: &str) -> Result<(), tokio_tungstenite::tungstenite::Error> where S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin, { let f = serde_json::json!({"type": "session.update", "session": {"model": model}}); ws.send(Message::Text(f.to_string())).await } // start cheap, escalate for one turn, drop back to the router async fn arc<S>(ws: &mut S) -> Result<(), tokio_tungstenite::tungstenite::Error> where S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin, { set_model(ws, "tier:quick").await?; ws.send(Message::Text(serde_json::json!({ "type": "response.create", "response": {"model": "tier:think"} }).to_string())).await?; set_model(ws, "auto").await }
func setModel(c *websocket.Conn, model string) error { return c.WriteJSON(map[string]any{ "type": "session.update", "session": map[string]any{"model": model}, }) } func turnWith(c *websocket.Conn, model string) error { // Per-turn precedence beats the session default and expires with the turn. return c.WriteJSON(map[string]any{ "type": "response.create", "response": map[string]any{"model": model}, }) } func arc(c *websocket.Conn) error { if err := setModel(c, "tier:quick"); err != nil { return err } if err := turnWith(c, "tier:think"); err != nil { return err } return setModel(c, "auto") }
Scope note, before you build on this. Function calling here follows the OpenAI
Realtime convention: tools declared on session.update, a
function_call item delivered on response.output_item.done and/or
response.function_call_arguments.done, a
function_call_output item sent back with conversation.item.create.
Every frame type in that flow is in the published machine-readable spec; the tool
fields inside those frames are not separately enumerated there. Verify on your node
before you depend on it — declare one tool, ask a question that needs it, and see whether a
function_call item comes back. If a node ignores tools you simply
never receive one; unknown fields are not an error and nothing else in your client breaks.
Declare tools once, in session.update, right after
session.created. Do not attach the schema to every
response.create: the schema is re-sent and re-billed as input tokens on every
single turn, which on a chatty voice session is the largest avoidable line on the invoice.
Declared at session level, it is established once for the socket's life.
tool_choice rides beside tools in that same
session.update, and it is optional: the default is "auto" — the
model decides whether to call anything. Send "none" to suppress calls for a stretch
of a call, or name a function to force one. A page that spells it out is showing the default
made explicit, not a field you are now obliged to send.
// -> once, at session start {"type":"session.update","session":{"model":"auto","tools":[ {"type":"function","name":"tide_height", "description":"Current tide height in metres for a port.", "parameters":{"type":"object", "properties":{"port":{"type":"string"}}, "required":["port"]}}]}} // <- the model wants the call (arguments arrive as a JSON *string*). // TWO frames can carry the SAME finished call. Same call_id — dedupe on it. {"type":"response.output_item.done","item":{"id":"item_58","type":"function_call", "name":"tide_height","call_id":"call_a19","arguments":"{\"port\":\"Saint John\"}"}} {"type":"response.function_call_arguments.done","call_id":"call_a19", "name":"tide_height","arguments":"{\"port\":\"Saint John\"}"} {"type":"response.done","response":{"id":"resp_9d1","status":"completed"}} // -> you answer every call the turn produced, then ask for the spoken // turn ONCE — after response.done, not once per call. {"type":"conversation.item.create","item":{"type":"function_call_output", "call_id":"call_a19","output":"{\"metres\":11.7,\"rising\":true}"}} {"type":"response.create"}
One finished call can arrive twice. Two frame types carry a completed tool call:
response.output_item.done with item.type == "function_call", and
response.function_call_arguments.done, which carries the same
call_id and the same argument string at the frame root. A node may send either or
both, in either order. Dedupe on call_id: keep a set of the ids you have
already executed for this turn and drop the second frame, or you will run the tool twice. Both
frames are complete and carry the whole argument string, so you never have to reassemble partial
JSON. If a node also streams argument fragments under a frame type you do not recognise, ignore
it: unknown frames are non-fatal by contract.
Send exactly one response.create per turn, after
response.done — however many tool calls that turn produced. Firing one after
each function_call_output is the classic port bug: on a two-call turn it asks for
two spoken replies and the second talks over the first. Queue the outputs as the calls arrive,
send them all with conversation.item.create, and ask for the reply once, when the
turn that requested them has closed.
A returned call is a request, not an authorisation. Run it through whatever consent gate your app has before it touches an actuator.
TOOLS = [{"type": "function", "name": "tide_height",
"description": "Current tide height in metres for a port.",
"parameters": {"type": "object",
"properties": {"port": {"type": "string"}},
"required": ["port"]}}]
def tide_height(port: str) -> dict:
return {"metres": 11.7, "rising": True, "port": port}
async def run(ws):
# ONCE. Not per response.create — that re-bills the schema every turn.
await ws.send(json.dumps({"type": "session.update",
"session": {"model": "auto", "tools": TOOLS}}))
seen: set = set() # call_ids already executed this turn
pending: list = [] # function_call_output items waiting to go back
async for raw in ws:
e = json.loads(raw)
t = e.get("type")
if t == "response.done":
# ONE response.create for the whole turn, after it closes —
# never one per call, which asks for overlapping replies.
if pending:
for item in pending:
await ws.send(json.dumps({"type": "conversation.item.create",
"item": item}))
pending.clear()
await ws.send(json.dumps({"type": "response.create"}))
seen.clear()
continue
# Two frame types carry ONE finished call: the item frame, and the
# arguments frame with the fields at the root. Take whichever came.
if t == "response.output_item.done" and \
e.get("item", {}).get("type") == "function_call":
call = e["item"]
elif t == "response.function_call_arguments.done":
call = e
else:
continue
cid = call.get("call_id")
if not cid or cid in seen:
continue # the duplicate frame; drop it
seen.add(cid)
args = json.loads(call.get("arguments") or "{}")
result = tide_height(**args) # gate this on consent in real code
pending.append({"type": "function_call_output",
"call_id": cid,
"output": json.dumps(result)}) # output is a STRING
const tools = [{ type: "function", name: "tide_height", description: "Current tide height in metres for a port.", parameters: { type: "object", properties: { port: { type: "string" } }, required: ["port"] }, }]; const impls: Record<string, (a: any) => Promise<unknown>> = { tide_height: async ({ port }) => ({ metres: 11.7, rising: true, port }), }; const seen = new Set<string>(); // call_ids already executed this turn let pending: unknown[] = []; // function_call_output items waiting to go back ws.on("message", async (b: Buffer, isBinary: boolean) => { if (isBinary) return; const e = JSON.parse(b.toString("utf8")); if (e.type === "session.created") { // Once per socket. Per-turn declaration re-bills the schema every turn. ws.send(JSON.stringify({ type: "session.update", session: { model: "auto", tools } })); } // Two frame types carry ONE finished call — the item frame, and the // arguments frame with the fields at the root. Accept either. const call = e.type === "response.output_item.done" && e.item?.type === "function_call" ? e.item : e.type === "response.function_call_arguments.done" ? e : null; if (call?.call_id && !seen.has(call.call_id)) { // dedupe on call_id seen.add(call.call_id); const fn = impls[call.name]; const out = fn ? await fn(JSON.parse(call.arguments ?? "{}")) : { error: `no such tool: ${call.name}` }; // answer, never stall pending.push({ type: "function_call_output", call_id: call.call_id, output: JSON.stringify(out) }); } if (e.type === "response.done") { // ONE response.create for the whole turn, after it closes — not one // per call, which would ask for two overlapping spoken replies. if (pending.length) { for (const item of pending) ws.send(JSON.stringify({ type: "conversation.item.create", item })); pending = []; ws.send(JSON.stringify({ type: "response.create" })); } seen.clear(); } });
const tools = [{ 'type': 'function', 'name': 'tide_height', 'description': 'Current tide height in metres for a port.', 'parameters': { 'type': 'object', 'properties': {'port': {'type': 'string'}}, 'required': ['port'], }, }]; void wireTools(WebSocketChannel ch) { final seen = <String>{}; // call_ids done this turn final pending = <Map<String, dynamic>>[]; // outputs waiting to go back ch.stream.listen((raw) async { if (raw is! String) return; final e = jsonDecode(raw) as Map<String, dynamic>; if (e['type'] == 'session.created') { // Declared once for the socket's life — not per turn. ch.sink.add(jsonEncode({'type': 'session.update', 'session': {'model': 'auto', 'tools': tools}})); return; } if (e['type'] == 'response.done') { // ONE response.create for the whole turn, after it closes. if (pending.isNotEmpty) { for (final item in pending) { ch.sink.add(jsonEncode( {'type': 'conversation.item.create', 'item': item})); } pending.clear(); ch.sink.add(jsonEncode({'type': 'response.create'})); } seen.clear(); return; } // Two frame types carry ONE finished call: the item frame, and the // arguments frame with the fields at the root. Take whichever came. final item = e['item'] as Map<String, dynamic>?; final Map<String, dynamic>? call = (e['type'] == 'response.output_item.done' && item?['type'] == 'function_call') ? item : (e['type'] == 'response.function_call_arguments.done') ? e : null; if (call == null) return; final cid = call['call_id'] as String?; if (cid == null || !seen.add(cid)) return; // duplicate frame: drop it final args = jsonDecode((call['arguments'] as String?) ?? '{}') as Map<String, dynamic>; final out = await runTool(call['name'] as String, args); // consent gate here pending.add({ 'type': 'function_call_output', 'call_id': cid, 'output': jsonEncode(out), }); }); }
var tools = []any{map[string]any{ "type": "function", "name": "tide_height", "description": "Current tide height in metres for a port.", "parameters": map[string]any{ "type": "object", "properties": map[string]any{"port": map[string]any{"type": "string"}}, "required": []string{"port"}, }}} type itemFrame struct { Type string `json:"type"` Item struct { Type string `json:"type"` Name string `json:"name"` CallID string `json:"call_id"` Arguments string `json:"arguments"` } `json:"item"` // response.function_call_arguments.done carries the SAME finished call // with these fields at the ROOT. Dedupe the two on call_id. CallID string `json:"call_id"` Arguments string `json:"arguments"` } // Per-turn state: ids already executed, outputs waiting to go back. var seen = map[string]bool{} var pending []map[string]any func onFrame(c *websocket.Conn, data []byte) error { var f itemFrame if err := json.Unmarshal(data, &f); err != nil { return nil } if f.Type == "session.created" { // once per socket; per-turn tools re-bill the schema every turn return c.WriteJSON(map[string]any{"type": "session.update", "session": map[string]any{"model": "auto", "tools": tools}}) } if f.Type == "response.done" { // ONE response.create for the turn, after it closes — never one // per call, which asks for overlapping spoken replies. seen = map[string]bool{} if len(pending) == 0 { return nil } for _, item := range pending { if err := c.WriteJSON(map[string]any{ "type": "conversation.item.create", "item": item}); err != nil { return err } } pending = nil return c.WriteJSON(map[string]any{"type": "response.create"}) } // Two frame types carry one finished call. Take whichever arrives. callID, arguments := f.CallID, f.Arguments switch { case f.Type == "response.output_item.done" && f.Item.Type == "function_call": callID, arguments = f.Item.CallID, f.Item.Arguments case f.Type == "response.function_call_arguments.done": default: return nil } if callID == "" || seen[callID] { return nil } // duplicate frame: drop seen[callID] = true var args struct{ Port string `json:"port"` } if err := json.Unmarshal([]byte(arguments), &args); err != nil { return err } out, err := json.Marshal(map[string]any{"metres": 11.7, "port": args.Port}) if err != nil { return err } pending = append(pending, map[string]any{"type": "function_call_output", "call_id": callID, "output": string(out)}) return nil }
use std::collections::HashSet; // `seen` and `pending` are per-turn state owned by the read loop: // call_ids already executed, and the outputs waiting to go back. async fn on_frame<S>(ws: &mut S, e: &serde_json::Value, seen: &mut HashSet<String>, pending: &mut Vec<serde_json::Value>) -> Result<(), Box<dyn std::error::Error>> where S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin, { if e["type"] == "session.created" { let f = serde_json::json!({"type": "session.update", "session": { "model": "auto", "tools": [{"type": "function", "name": "tide_height", "description": "Current tide height in metres for a port.", "parameters": {"type": "object", "properties": {"port": {"type": "string"}}, "required": ["port"]}}]}}); ws.send(Message::Text(f.to_string())).await?; // once, not per turn return Ok(()); } if e["type"] == "response.done" { // ONE response.create for the whole turn, after it closes — never // one per call, which asks for overlapping spoken replies. if !pending.is_empty() { for item in pending.drain(..) { ws.send(Message::Text(serde_json::json!({ "type": "conversation.item.create", "item": item }).to_string())).await?; } ws.send(Message::Text(r#"{"type":"response.create"}"#.into())).await?; } seen.clear(); return Ok(()); } // Two frame types carry ONE finished call: the item frame, and the // arguments frame with the fields at the root. Take whichever came. let call = match e["type"].as_str().unwrap_or_default() { "response.output_item.done" if e["item"]["type"] == "function_call" => &e["item"], "response.function_call_arguments.done" => e, _ => return Ok(()), }; let Some(cid) = call["call_id"].as_str() else { return Ok(()); }; if !seen.insert(cid.to_string()) { return Ok(()) } // duplicate: drop it let args: serde_json::Value = serde_json::from_str(call["arguments"].as_str().unwrap_or("{}"))?; let out = serde_json::json!({"metres": 11.7, "port": args["port"]}).to_string(); pending.push(serde_json::json!({"type": "function_call_output", "call_id": cid, "output": out})); Ok(()) }
Be blunt about this, because a wrong assumption here costs a user their conversation:
| state | survives a reconnect? | what you must do |
|---|---|---|
| conversation history | no — it is per-socket | replay the turns you kept locally with conversation.item.create before the first response.create |
| session model | no | re-send session.update, or set ?model= on the new URL |
| tool declarations | no | re-declare once on session.created |
| input audio buffer | no | an uncommitted buffer is gone; re-capture |
| your API key + rate limit | yes — they are per key, not per socket | reconnect storms still count against RPM. Back off. |
Never retry an auth failure. A 401 invalid_api_key on the upgrade, or an
error frame with code: "invalid_api_key", is terminal — retrying it in
a loop is how a key gets rate-limited into a lockout. Retry transport failures and
503-class conditions with exponential backoff and jitter; on
429 rate_limit_exceeded honour Retry-After from the upgrade response
when you have it.
Liveness: ouroboros.telemetry is emitted every 8 tokens and as a 5 s
idle heartbeat. If you have seen neither a telemetry frame nor a pong for ~15 s, the socket is
dead even if the OS has not noticed. Refresh your watchdog on every inbound frame, not
only on pongs — otherwise a busy, healthy socket trips your own deadline.
import asyncio, json, random, websockets class Fatal(Exception): pass async def session_once(url, hdr, history, handle): async with websockets.connect(url, additional_headers=hdr, ping_interval=20, ping_timeout=20) as ws: await ws.recv() # session.created await ws.send(json.dumps({"type": "session.update", "session": {"model": "auto", "tools": TOOLS}})) for turn in history: # nothing carried over await ws.send(json.dumps({"type": "conversation.item.create", "item": turn})) async for raw in ws: e = json.loads(raw) if e.get("type") == "error" and \ e["error"].get("code") == "invalid_api_key": raise Fatal(e["error"]) # do NOT retry handle(e) async def run_forever(url, hdr, history, handle): delay = 0.5 while True: try: await session_once(url, hdr, history, handle) delay = 0.5 # clean close: reset except Fatal: raise except (OSError, websockets.WebSocketException) as exc: print("reconnect after", exc) await asyncio.sleep(delay + random.random() * 0.3) delay = min(delay * 2, 30) # jittered, capped
let delay = 500; let fatal = false; const history: unknown[] = []; // you keep this; the server does not function connect() { if (fatal) return; const ws = new WebSocket("wss://api.hawktalk.ai/v1/realtime?model=auto", { headers: { Authorization: `Bearer ${process.env.HAWKTALK_API_KEY}` } }); let lastSeen = Date.now(); const watchdog = setInterval(() => { // telemetry heartbeats every 5s idle; 15s of silence means it's gone if (Date.now() - lastSeen > 15000) ws.terminate(); }, 5000); ws.on("open", () => { delay = 500; }); ws.on("pong", () => { lastSeen = Date.now(); }); ws.on("message", (b: Buffer, isBinary: boolean) => { lastSeen = Date.now(); // EVERY frame, not just pongs if (isBinary) return; const e = JSON.parse(b.toString("utf8")); if (e.type === "session.created") { ws.send(JSON.stringify({ type: "session.update", session: { model: "auto", tools } })); for (const item of history) // replay: nothing survived ws.send(JSON.stringify({ type: "conversation.item.create", item })); } if (e.type === "error" && e.error?.code === "invalid_api_key") { fatal = true; // terminal — never loop on auth console.error("bad key, not retrying"); } }); ws.on("close", () => { clearInterval(watchdog); if (fatal) return; setTimeout(connect, delay + Math.random() * 300); delay = Math.min(delay * 2, 30000); }); ws.on("error", (err) => console.error("socket", err)); } connect();
import 'dart:convert'; import 'dart:math'; // Random import 'package:web_socket_channel/io.dart'; class Reconnector { Reconnector(this.key); final String key; final List<Map<String, dynamic>> history = []; Duration _delay = const Duration(milliseconds: 500); bool _fatal = false; Future<void> run() async { while (!_fatal) { final ch = IOWebSocketChannel.connect( Uri.parse('wss://api.hawktalk.ai/v1/realtime?model=auto'), headers: {'Authorization': 'Bearer $key'}, pingInterval: const Duration(seconds: 20)); try { await for (final raw in ch.stream) { if (raw is! String) continue; final e = jsonDecode(raw) as Map<String, dynamic>; if (e['type'] == 'session.created') { ch.sink.add(jsonEncode({'type': 'session.update', 'session': {'model': 'auto'}})); for (final item in history) { // replay — none of it carried ch.sink.add(jsonEncode( {'type': 'conversation.item.create', 'item': item})); } _delay = const Duration(milliseconds: 500); } if (e['type'] == 'error' && (e['error'] as Map)['code'] == 'invalid_api_key') { _fatal = true; // a bad key never fixes itself break; } } } catch (e) { print('socket dropped: $e'); } await ch.sink.close(); if (_fatal) break; await Future<void>.delayed(_delay + Duration(milliseconds: Random().nextInt(300))); if (_delay < const Duration(seconds: 30)) _delay *= 2; } } }
use rand::Rng; use std::time::Duration; async fn run_forever(key: &str) -> Result<(), Box<dyn std::error::Error>> { let mut delay_ms: u64 = 500; loop { match session_once(key).await { Ok(()) => delay_ms = 500, // Auth is terminal: bubble it out instead of hammering the gateway. Err(e) if e.to_string().contains("invalid_api_key") => return Err(e), Err(e) => eprintln!("reconnecting after {e}"), } let jitter = rand::thread_rng().gen_range(0..300); tokio::time::sleep(Duration::from_millis(delay_ms + jitter)).await; delay_ms = (delay_ms * 2).min(30_000); } }
var errBadKey = errors.New("invalid_api_key") func runForever(key string, history []map[string]any) { delay := 500 * time.Millisecond for { started := time.Now() err := sessionOnce(key, history) if errors.Is(err, errBadKey) { log.Fatal("invalid_api_key: not retrying") // terminal, by design } log.Printf("session ended: %v", err) if time.Since(started) > time.Minute { delay = 500 * time.Millisecond // it worked for a while: reset } time.Sleep(delay + time.Duration(rand.Intn(300))*time.Millisecond) if delay < 30*time.Second { delay *= 2 } } } func sessionOnce(key string, history []map[string]any) error { h := http.Header{} h.Set("Authorization", "Bearer "+key) c, resp, err := websocket.DefaultDialer.Dial( "wss://api.hawktalk.ai/v1/realtime?model=auto", h) if err != nil { if resp != nil && resp.StatusCode == http.StatusUnauthorized { return errBadKey } if resp != nil && resp.StatusCode == http.StatusTooManyRequests { if ra := resp.Header.Get("Retry-After"); ra != "" { log.Printf("rate limited, retry after %s", ra) } } return err } defer c.Close() // 15s of silence = dead: telemetry heartbeats at 5s idle. Bump the // deadline on EVERY frame — a pong-only refresh kills a busy socket. bump := func() error { return c.SetReadDeadline(time.Now().Add(15 * time.Second)) } if err := bump(); err != nil { return err } c.SetPongHandler(func(string) error { return bump() }) // Nothing survived the reconnect: re-declare, then replay. if err := c.WriteJSON(map[string]any{"type": "session.update", "session": map[string]any{"model": "auto", "tools": tools}}); err != nil { return err } for _, item := range history { if err := c.WriteJSON(map[string]any{ "type": "conversation.item.create", "item": item}); err != nil { return err } } for { typ, data, err := c.ReadMessage() if err != nil { return err } if err := bump(); err != nil { return err } if typ != websocket.TextMessage { continue } // binary: rejected by design if err := onFrame(c, data); err != nil { return err } } }
Everything above, assembled: connect, declare nothing fancy, capture at 24000 Hz, commit on your own VAD, play what comes back at the frame's own rate, and barge in the instant the user speaks over the model.
// pubspec: web_socket_channel: ^3.0.0 record: ^5.1.0 audioplayers: ^6.0.0 import 'dart:async'; import 'dart:convert'; import 'dart:math'; // sqrt — needed by _rms below import 'dart:typed_data'; import 'package:audioplayers/audioplayers.dart'; import 'package:record/record.dart'; import 'package:web_socket_channel/io.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; // wavPcm16() is the helper from 02.4 — keep it in a shared file. class HawkVoiceLoop { HawkVoiceLoop(this.apiKey); final String apiKey; late final WebSocketChannel _ch; final _rec = AudioRecorder(); final _player = AudioPlayer(); final _pcmOut = BytesBuilder(); final transcript = StreamController<String>.broadcast(); final visemes = StreamController<String>.broadcast(); StreamSubscription<List<int>>? _mic; int _outRate = 24000; bool _speaking = false; // user is talking (your vad owns this) bool _modelTalking = false; // model audio is playing Future<void> start() async { _ch = IOWebSocketChannel.connect( Uri.parse('wss://api.hawktalk.ai/v1/realtime?model=auto'), headers: {'Authorization': 'Bearer $apiKey'}, pingInterval: const Duration(seconds: 20), ); _ch.stream.listen(_onFrame, onError: (e) => transcript.addError(e), onDone: () => stop()); if (!await _rec.hasPermission()) throw StateError('mic permission denied'); final stream = await _rec.startStream(const RecordConfig( encoder: AudioEncoder.pcm16bits, sampleRate: 24000, numChannels: 1)); _mic = stream.listen(_onMic); // runs for the whole call, not one turn } // ---- uplink ------------------------------------------------------------- void _onMic(List<int> chunk) { final loud = _rms(Uint8List.fromList(chunk)) > 500; // crude local vad if (loud && _modelTalking) bargeIn(); // talk-over detected if (!loud && !_speaking) return; // silence: send nothing _speaking = true; _ch.sink.add(jsonEncode({'type': 'input_audio_buffer.append', 'audio': base64Encode(chunk)})); } /// YOUR end-of-speech, called from your hangover timer. The server emits no /// speech_started/speech_stopped — it does semantic endpointing only, and /// nothing happens until you commit. void endOfSpeech() { if (!_speaking) return; _speaking = false; _ch.sink.add(jsonEncode({'type': 'input_audio_buffer.commit'})); _ch.sink.add(jsonEncode({'type': 'response.create'})); } void bargeIn() { _player.stop(); // local flush first — 0 ms, not a round-trip _pcmOut.clear(); _modelTalking = false; _ch.sink.add(jsonEncode({'type': 'response.cancel'})); // rolls history back } // ---- downlink ----------------------------------------------------------- void _onFrame(dynamic raw) { if (raw is! String) return; // binary is rejected by design final e = jsonDecode(raw) as Map<String, dynamic>; switch (e['type']) { case 'session.created': _ch.sink.add(jsonEncode({'type': 'session.update', 'session': {'model': 'auto'}})); case 'conversation.item.input_audio_transcription.completed': transcript.add('you: ${e['transcript']}'); case 'response.text.delta': transcript.add('${e['delta'] ?? e['text'] ?? ''}'); case 'response.audio.delta': _modelTalking = true; _outRate = (e['x_sample_rate_hz'] as int?) ?? _outRate; final b64 = (e['delta'] ?? e['audio']) as String?; if (b64 != null) _pcmOut.add(base64Decode(b64)); case 'response.viseme.delta': final v = e['viseme']; if (v is String) visemes.add(v); case 'response.audio.done': final pcm = _pcmOut.takeBytes(); if (pcm.isNotEmpty) _player.play(BytesSource(wavPcm16(pcm, _outRate))); case 'response.done': _modelTalking = false; final x = (e['x_ouroboros'] ?? (e['response'] as Map?)?['x_ouroboros'] ?? {}) as Map; // nulls are "not measured" — render them as gaps, never as 0. print('stt=${x['stt_ms']} ttft=${x['ttft_ms']} ' 'audio=${x['first_audio_out_ms']} e2e=${x['end_to_end_ms']}'); case 'error': final err = e['error'] as Map; if (err['code'] == 'tts_not_wired' || err['code'] == 'stt_not_wired') { // Seam absent on this node: fall back to on-device speech. // You will never be handed fake audio or a guessed transcript. transcript.add('[seam ${err['code']} — using device speech]'); } else { transcript.addError(StateError('${err['code']}: ${err['message']}')); } default: break; // unknown frames are never fatal } } double _rms(Uint8List b) { final s = b.buffer.asInt16List(0, b.lengthInBytes ~/ 2); if (s.isEmpty) return 0; var acc = 0.0; for (final v in s) acc += v * v.toDouble(); return sqrt(acc / s.length); } Future<void> stop() async { await _mic?.cancel(); await _rec.stop(); await _player.stop(); await _ch.sink.close(); } }
// No build step needed beyond your bundler. Key comes from your backend. export class HawkVoiceLoop { private ws!: WebSocket; private inCtx = new AudioContext({ sampleRate: 24000 }); // server input default private outCtx = new AudioContext(); private queued: AudioBufferSourceNode[] = []; private playhead = 0; private carry = new Uint8Array(0); private speaking = false; private modelTalking = false; private silentFrames = 0; private text = ""; // this turn's reply, for the seam fallback private ttsSeamDown = false; /// Wire these to your UI; there is no `process` in a browser. onText?: (full: string) => void; onViseme?: (v: string) => void; async start(key: string) { this.ws = new WebSocket("wss://api.hawktalk.ai/v1/realtime?model=auto", ["openai-insecure-api-key." + key.trim()]); this.ws.onmessage = (ev) => { if (typeof ev.data !== "string") return; // text frames only this.onFrame(JSON.parse(ev.data)); }; this.ws.onerror = (e) => console.error("socket", e); const media = await navigator.mediaDevices.getUserMedia({ audio: { channelCount: 1, echoCancellation: true, noiseSuppression: true } }); const worklet = ` class Cap extends AudioWorkletProcessor { process(inputs) { const ch = inputs[0][0]; if (!ch) return true; const pcm = new Int16Array(ch.length); let peak = 0; for (let i = 0; i < ch.length; i++) { const s = Math.max(-1, Math.min(1, ch[i])); peak = Math.max(peak, Math.abs(s)); pcm[i] = s < 0 ? s * 0x8000 : s * 0x7fff; } this.port.postMessage({ pcm: pcm.buffer, peak }, [pcm.buffer]); return true; } } registerProcessor('cap', Cap);`; await this.inCtx.audioWorklet.addModule( URL.createObjectURL(new Blob([worklet], { type: "application/javascript" }))); const node = new AudioWorkletNode(this.inCtx, "cap"); this.inCtx.createMediaStreamSource(media).connect(node); node.port.onmessage = (ev) => this.onMic(ev.data.pcm, ev.data.peak); } // ---- uplink: YOUR vad decides everything ------------------------------- private onMic(buf: ArrayBuffer, peak: number) { const loud = peak > 0.02; if (loud && this.modelTalking) this.bargeIn(); if (!loud && !this.speaking) return; this.speaking = true; this.send({ type: "input_audio_buffer.append", audio: this.b64(buf) }); // 128-sample worklet blocks @24 kHz ~ 5.3 ms; 60 quiet blocks ~ 320 ms of // hangover. This number IS the user's perceived response latency. this.silentFrames = loud ? 0 : this.silentFrames + 1; if (this.silentFrames > 60) { this.silentFrames = 0; this.endOfSpeech(); } } endOfSpeech() { if (!this.speaking) return; this.speaking = false; this.text = ""; this.ttsSeamDown = false; this.send({ type: "input_audio_buffer.commit" }); this.send({ type: "response.create" }); } bargeIn() { for (const n of this.queued) { try { n.stop(); } catch {} } this.queued = []; this.playhead = 0; this.carry = new Uint8Array(0); this.modelTalking = false; speechSynthesis.cancel(); // kill the fallback voice too this.send({ type: "response.cancel" }); // history rolls back to the turn } // ---- downlink ----------------------------------------------------------- private onFrame(e: any) { switch (e.type) { case "session.created": this.send({ type: "session.update", session: { model: "auto" } }); break; case "conversation.item.input_audio_transcription.completed": console.log("you:", e.transcript); break; case "response.text.delta": this.text += e.delta ?? e.text ?? ""; this.onText?.(this.text); break; case "response.audio.delta": this.modelTalking = true; this.playPcm(e); break; case "response.viseme.delta": if (typeof e.viseme === "string") this.onViseme?.(e.viseme); break; case "response.done": { this.modelTalking = false; const x = e.x_ouroboros ?? e.response?.x_ouroboros ?? {}; console.log("timings", x); // nulls mean unmeasured, not zero // Seam was absent: speak the REAL reply with the browser voice, at the // point we know the whole text. Never fabricate audio to cover a 501. if (this.ttsSeamDown && this.text) speechSynthesis.speak(new SpeechSynthesisUtterance(this.text)); break; } case "error": if (e.error?.code === "tts_not_wired") { this.ttsSeamDown = true; // documented seam, not a bug } else console.error("api", e.error); break; default: break; // unknown frame: log it, never throw } } private playPcm(e: any) { const bin = atob(e.delta ?? e.audio ?? ""); let bytes = new Uint8Array(this.carry.length + bin.length); bytes.set(this.carry, 0); for (let i = 0; i < bin.length; i++) bytes[this.carry.length + i] = bin.charCodeAt(i); if (bytes.length % 2) { this.carry = bytes.slice(bytes.length - 1); bytes = bytes.slice(0, bytes.length - 1); } else this.carry = new Uint8Array(0); if (!bytes.length) return; const rate = e.x_sample_rate_hz ?? 24000; // the frame is the authority const pcm = new Int16Array(bytes.buffer, bytes.byteOffset, bytes.length / 2); const buf = this.outCtx.createBuffer(1, pcm.length, rate); const ch = buf.getChannelData(0); for (let i = 0; i < pcm.length; i++) ch[i] = pcm[i] / 0x8000; const node = this.outCtx.createBufferSource(); node.buffer = buf; node.connect(this.outCtx.destination); const at = Math.max(this.outCtx.currentTime, this.playhead); node.start(at); this.playhead = at + buf.duration; this.queued.push(node); node.onended = () => { this.queued = this.queued.filter(n => n !== node); }; } private b64(buf: ArrayBuffer): string { const b = new Uint8Array(buf); let s = ""; for (let i = 0; i < b.length; i += 0x8000) s += String.fromCharCode(...b.subarray(i, i + 0x8000)); return btoa(s); } private send(o: unknown) { if (this.ws.readyState === WebSocket.OPEN) this.ws.send(JSON.stringify(o)); } }
| code | meaning | what your client should do |
|---|---|---|
| invalid_request | 400-class: malformed frame or field | fix the frame; never retry unchanged |
| invalid_api_key | 401-class: bad or revoked key | terminal — stop, surface it, do not reconnect |
| model_not_found | 404-class: id or tier not on this node | fall back to "auto"; re-read GET /v1/models |
| rate_limit_exceeded | 429-class, per key | honour Retry-After when present, else jittered backoff |
| stt_not_wired / tts_not_wired | 501-class: the seam is absent on this node | fall back to device speech — this is a designed outcome, not a bug |
| backend_failed / stt_failed / tts_failed | 502-class | retry the turn once, then degrade |
| model_unavailable / backend_unavailable / compute_not_wired | 503-class | switch tier or degrade the feature; back off |
Every error frame has the same body: {"error":{"message","type","code","param"}}.
Branch on code, show message, and log the rest.
Status, stated plainly: the Realtime WebSocket session, per-key RPM limiting and the
usage ledger are shipped. REST STT/TTS return 501 on nodes where the seam is not
wired, and the same absence shows up here as stt_not_wired /
tts_not_wired. The HawkTalkLive lane mux at /live/brain is
preview and is not served on api.hawktalk.ai: it exists on a local
node only, loopback only, no auth wired — do not point production at it. Monthly token
quota enforcement in the request path is roadmap, not deployed.