Realtime

WSSwss://voicebot-serverless-stg.botnoigroup.com/v1/call

Mint a call token with your access token, open the WebSocket with it, stream microphone audio in and receive transcripts and TTS back.

Connect

1. Mint a call token on your server
Authenticate with an access token. Copy it from the Developers page of the Botnoi Voice console and send it on every request as Authorization: Bearer <access_token>. Post the agent you want to talk to:
```
curl -s -X POST https://backend-voicebot-stg.botnoigroup.com/call/tokens \
  -H "Authorization: Bearer $ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"agent_id": "<agent_id>", "channel": "web"}'
```
Your call token comes back in ephemeral_token (vb_live_…). It is single-use and short-lived (currently about two minutes): a failed connect or a second connect needs a new one.
Your access token never reaches the browser. Mint on your server and hand the page only the call token. The endpoint is Create a call token; every status it can return is under Errors below.
2. Connect from the browser
Open wss://voicebot-serverless-stg.botnoigroup.com/v1/call?token=<call_token>. No headers, and no other query parameters.
The socket opens before the token is checked, so onopen is not success — wait for opened.

Name	In	Required	Type	Description
token	query	Yes	string	Your call token from Create a call token. Single-use.

The `open` you send

{
  "version": "2",
  "type": "open",
  "seq": 1,
  "id": "<uuid — your session id>",
  "position": "POST-ANSWER",
  "parameters": {
    "conversationId": "<your id, [A-Za-z0-9_-]{1,128}>",
    "media": [
      {
        "type": "audio/L16",
        "sampleRateHz": 16000,
        "channels": [
          "external"
        ]
      }
    ]
  }
}

Send open as your first frame (seq 1). The agent is chosen by your token — any agent_id in inputVariables is ignored.

Every frame you send carries seq, starting at 1 and going up by one; frames out of order end the call. Server frames echo the seq they answer as clientseq.

opened echoes the media entry the server selected, with startPaused: false. Send audio only after it arrives.

Rules

After opened, send microphone frames continuously — about 20 ms each, no WAV header.
On every TTS burst, send playback_started when it starts playing and playback_completed when it finishes. These are required: when the agent is not interruptible, the server does not listen to the microphone until it receives playback_completed.
On barge_in, stop playing TTS and clear the playback queue.
To hang up, send close with reason: "end"; the server replies closed. Frames can still arrive after closed until the socket closes — ignore them; they are not an error.
A call has a server-side maximum length, currently about ten minutes. At the limit the server sends disconnect, then closes the socket.

Migrating from `/v1/preview_call`

Authenticate with a token query parameter — your call token — instead of api_key and agent_id.
You send open yourself; the proxy used to send it for you.
Close codes 1008 and 1011 no longer occur; expect 4001, 3000 and 3003.
No headers of any kind.
inputVariables.agent_id is ignored — the token picks the agent.

This box goes away when /v1/preview_call is retired.

Binary audio

After opened, both directions use the selected encoding with no WAV header. Send about 20 ms per binary frame: sample_rate × 0.02 × bytes_per_sample × channels. 16 kHz L16 mono is 640 bytes; 8 kHz µ-law mono is 160 bytes.

Client messages

JSON you send, starting with open, plus raw binary audio once opened arrives.

type: openYour first frame, seq 1. The agent comes from your call token; wait for opened before sending audio.

Name	Type	Required	Description
type	"open"	Yes	Message type.
seq	1	Yes	Your first sequence number.
id	string	Yes	A session id you choose; use a UUID.
position	string	Yes	"POST-ANSWER".
parameters.conversationId	string	Yes	<your id, [A-Za-z0-9_-]{1,128}>
parameters.media	Media[]	Yes	[{"type":"audio/L16","sampleRateHz":16000,"channels":["external"]}]

type: pingApplication keepalive. Not a WebSocket ping.

Name	Type	Required	Description
type	"ping"	Yes	Message type.
seq	integer	Yes	Previous client seq + 1.

type: playback_startedClient began playing a TTS burst.

Name	Type	Required	Description
type	"playback_started"	Yes	Message type.

type: playback_completedClient finished a TTS burst. When in-flight audio drains, recognition unmutes.

Name	Type	Required	Description
type	"playback_completed"	Yes	Message type.

type: closeEnd the session. Do not send more audio after this.

Name	Type	Required	Description
type	"close"	Yes	Message type.
parameters.reason	"end" \| "disconnect"	Yes	How the client is ending the call.

type: binary audioRaw PCM16 or µ-law frames. No WAV header. After opened only.

Name	Type	Required	Description
(binary frame)	bytes	Yes	20 ms recommended. Encoding must match opened.parameters.media[0].

Server messages

JSON text frames the server sends, plus binary TTS. Every JSON envelope also carries version, seq, clientseq, and id.

type: openedReply to your open. Safe to send audio; echoes the selected media.

Name	Type	Required	Description
type	"opened"	Yes	Message type.
parameters.startPaused	boolean	Yes	false
parameters.media	Media[]	Yes	Single selected descriptor. Match this on the wire.

type: pongReply to ping.

Name	Type	Required	Description
type	"pong"	Yes	Message type.

type: eventTranscripts and barge-in.

Name	Type	Required	Description
type	"event"	Yes	Message type.
parameters.entities[].type	string	Yes	user_turn_response \| bot_turn_response \| barge_in
parameters.entities[].data	object	Yes	user: { transcript.result.text, is_final, timestamp? }. bot jsonConfig: { output, action, response_type, phone_transfer }. close-end may send { disposition: match }. barge_in: {} — stop TTS and clear the playback queue; keep sending mic.

type: closedReply to close. Socket will end.

Name	Type	Required	Description
type	"closed"	Yes	Message type.

type: disconnectServer is ending the session.

Name	Type	Required	Description
type	"disconnect"	Yes	Message type.
parameters.reason	"completed" \| "error" \| "unauthorized"	Yes	Why the server closed.
parameters.info	string	Yes	Human-readable detail, e.g. sequence mismatch or bot_id not in allow_list.

type: binary audioBot TTS in the negotiated encoding. May arrive as several frames.

Name	Type	Required	Description
(binary frame)	bytes	Yes	Play it, then send playback_started / playback_completed.

Errors

Minting a call token

Status	detail	What it means
401	`Not authenticated` / `Invalid token` / `Token has expired` / `User not found`	Your access token is missing or invalid. Sign in to the console and copy it again.
402	a reason, or `credit_refused`	Not enough credit to start a call. `detail` says why.
403	`agent_id not allowed`	The agent isn't yours, or isn't linked to your account yet.
409	`connector_not_provisioned`	Your account isn't ready for calls yet. Try again in a few minutes.
422	field errors	`agent_id` is missing or blank, or `channel` isn't `web`.
500	`connector_error`	Server-side error. Retry; contact support if it persists.
502	`connector_store_unavailable` / `token_mint_upstream_error`	An upstream service failed. Retry.
503	`token_mint_not_configured`	Preview calls aren't enabled on this environment.

During the call

Close code	What it means
4001	Token invalid, expired, already used, or no capacity — mint a new token. The server does not say which.
3000	Protocol error, sent after a `disconnect` frame: a `seq` out of order, or an agent that isn't allowed.
3003	Billing refused the call.
1000	Normal end: after `closed`, at the maximum call length, or when the agent ended the call.

1008 and 1011 do not occur on this WebSocket.

Message sequence

One call in wire order, as a browser client sees it. publish is what you send; subscribe is what the server sends back. Audio rows stand in for raw binary frames — they are not JSON on the wire.

openpublish

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19{
  "version":  "2",
  "type":  "open",
  "seq":  1,
  "id":  "00000000-0000-4000-8000-000000000001",
  "position":  "POST-ANSWER",
  "parameters":  {
    "conversationId":  "conv-example-1",
    "media":  [
      {
        "type":  "audio/L16",
        "sampleRateHz":  16000,
        "channels":  [
          "external"
        ]
      }
    ]
  }
}

openedsubscribe

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19{
  "version":  "2",
  "type":  "opened",
  "seq":  1,
  "clientseq":  1,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {
    "startPaused":  false,
    "media":  [
      {
        "type":  "audio/L16",
        "sampleRateHz":  16000,
        "channels":  [
          "external"
        ]
      }
    ]
  }
}

Greeting TTS: raw PCM16 little-endian, 16 kHz, no WAV header. It starts before the greeting text.

audiosubscribe

eventsubscribe

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24{
  "version":  "2",
  "type":  "event",
  "seq":  2,
  "clientseq":  1,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {
    "entities":  [
      {
        "type":  "bot_turn_response",
        "data":  {
          "output":  "สวัสดีค่ะ ยินดีต้อนรับค่ะ",
          "action":  "None",
          "response_type":  "text",
          "phone_transfer":  null,
          "turn":  {
            "id":  "t1",
            "rev":  0
          }
        }
      }
    ]
  }
}

The next greeting chunk's audio, again before its text.

audiosubscribe

eventsubscribe

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24{
  "version":  "2",
  "type":  "event",
  "seq":  3,
  "clientseq":  1,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {
    "entities":  [
      {
        "type":  "bot_turn_response",
        "data":  {
          "output":  "วันนี้มีอะไรให้ช่วยไหมคะ",
          "action":  "None",
          "response_type":  "text",
          "phone_transfer":  null,
          "turn":  {
            "id":  "t1",
            "rev":  0
          }
        }
      }
    ]
  }
}

playback_startedpublish

1
2
3
4
5
6
7{
  "version":  "2",
  "type":  "playback_started",
  "seq":  2,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {}
}

playback_completedpublish

1
2
3
4
5
6
7{
  "version":  "2",
  "type":  "playback_completed",
  "seq":  3,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {}
}

Microphone audio: 20 ms frames of PCM16, 640 bytes each at 16 kHz mono, sent continuously.

audiopublish

eventsubscribe

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23{
  "version":  "2",
  "type":  "event",
  "seq":  4,
  "clientseq":  3,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {
    "entities":  [
      {
        "type":  "user_turn_response",
        "data":  {
          "transcript":  {
            "result":  {
              "text":  "hello"
            }
          },
          "is_final":  false,
          "timestamp":  1767225600.52
        }
      }
    ]
  }
}

eventsubscribe

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23{
  "version":  "2",
  "type":  "event",
  "seq":  5,
  "clientseq":  3,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {
    "entities":  [
      {
        "type":  "user_turn_response",
        "data":  {
          "transcript":  {
            "result":  {
              "text":  "hello world"
            }
          },
          "is_final":  true,
          "timestamp":  1767225600.98
        }
      }
    ]
  }
}

Reply TTS: raw PCM16, 16 kHz, before the reply text.

audiosubscribe

eventsubscribe

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24{
  "version":  "2",
  "type":  "event",
  "seq":  6,
  "clientseq":  3,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {
    "entities":  [
      {
        "type":  "bot_turn_response",
        "data":  {
          "output":  "hello world",
          "action":  "None",
          "response_type":  "text",
          "phone_transfer":  null,
          "turn":  {
            "id":  "t2",
            "rev":  0
          }
        }
      }
    ]
  }
}

playback_startedpublish

1
2
3
4
5
6
7{
  "version":  "2",
  "type":  "playback_started",
  "seq":  4,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {}
}

playback_completedpublish

1
2
3
4
5
6
7{
  "version":  "2",
  "type":  "playback_completed",
  "seq":  5,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {}
}

closepublish

1
2
3
4
5
6
7
8
9{
  "version":  "2",
  "type":  "close",
  "seq":  6,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {
    "reason":  "end"
  }
}

eventsubscribe

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17{
  "version":  "2",
  "type":  "event",
  "seq":  7,
  "clientseq":  6,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {
    "entities":  [
      {
        "type":  "bot_turn_response",
        "data":  {
          "disposition":  "match"
        }
      }
    ]
  }
}

closedsubscribe

1
2
3
4
5
6
7
8{
  "version":  "2",
  "type":  "closed",
  "seq":  8,
  "clientseq":  6,
  "id":  "00000000-0000-4000-8000-000000000001",
  "parameters":  {}
}

Frames may still arrive for a few seconds after closed — ignore them; they are not an error.

The socket closes with code 1000.

Request

WSSwss://voicebot-serverless-stg.botnoigroup.com/v1/call?token=…

Download

On barge_in, stop TTS and clear the queue. Python: pip install websockets sounddevice numpy. Node: npm i ws and ffmpeg on the PATH.

<!doctype html>
<!--
  Realtime example: talk to one of your agents from a browser.

  1. On your server — never in this page — mint a call token with your access token:

       curl -s -X POST https://backend-voicebot-stg.botnoigroup.com/call/tokens \
         -H "Authorization: Bearer $ACCESS_TOKEN" \
         -H "Content-Type: application/json" \
         -d '{"agent_id": "<agent_id>", "channel": "web"}'

     The token is in "ephemeral_token". It works once and expires in about two minutes.
  2. Open this file, paste the call token, press Connect.
-->
<html lang="en">
<meta charset="utf-8" />
<title>Preview Call</title>
<style>
  body { font: 14px/1.4 system-ui, sans-serif; max-width: 40rem; margin: 2rem auto; }
  label { display: block; margin: 0.5rem 0; }
  input { width: 100%; }
  pre { background: #111; color: #ddd; padding: 1rem; min-height: 8rem; overflow: auto; }
</style>
<label>WebSocket URL <input id="wsUrl" autocomplete="off" placeholder="wss://voicebot-serverless-stg.botnoigroup.com/v1/call" /></label>
<label>Call token <input id="callToken" autocomplete="off" placeholder="vb_live_..." /></label>
<p>
  <button id="connect" type="button">Connect</button>
  <button id="hangup" type="button" disabled>Hang up</button>
</p>
<pre id="log"></pre>
<script>
const DEFAULT_WS = "wss://voicebot-serverless-stg.botnoigroup.com/v1/call";
function defaultWsUrl() {
  return DEFAULT_WS;
}
const FRAME_MS = 20;
const PLAYBACK_IDLE_MS = 600;
const PLAYBACK_LOOKAHEAD_SEC = 0.03;
// Coalesce 20 ms TTS frames into one buffer per burst. One source node per
// frame at a hardware-rate context buzzes at each node boundary.
const BURST_MS = 80;

const logEl = document.getElementById("log");
const connectBtn = document.getElementById("connect");
const hangupBtn = document.getElementById("hangup");

let ws = null;
let seq = 0;
let sessionId = "";
let opened = false;
let sampleRate = 16000;
let audioCtx = null;
let mediaStream = null;
let micNode = null;
let masterGain = null;
let micRemainder = new Float32Array(0);
let playbackActive = false;
let playbackIdle = null;
let pingTimer = null;
let nextPlayTime = 0;
let playbackGen = 0;
let dropBotUntil = 0;
const ttsSources = new Set();
let pendingChunks = [];
let burstTimer = null;

function log(line) {
  logEl.textContent += line + "\n";
  logEl.scrollTop = logEl.scrollHeight;
}

function sendJson(type, parameters = {}) {
  if (!ws || ws.readyState !== WebSocket.OPEN) return;
  ws.send(JSON.stringify({
    version: "2",
    type,
    seq: ++seq,
    id: sessionId,
    parameters,
  }));
}

function floatToL16(samples) {
  const out = new ArrayBuffer(samples.length * 2);
  const view = new DataView(out);
  for (let i = 0; i < samples.length; i++) {
    const s = Math.max(-1, Math.min(1, samples[i]));
    view.setInt16(i * 2, s < 0 ? s * 0x8000 : s * 0x7fff, true);
  }
  return out;
}

function l16ToFloat(buffer) {
  const view = new DataView(buffer);
  const out = new Float32Array(buffer.byteLength / 2);
  for (let i = 0; i < out.length; i++) out[i] = view.getInt16(i * 2, true) / 32768;
  return out;
}

function sendMic(float32) {
  const frame = Math.max(1, Math.round(sampleRate * FRAME_MS / 1000));
  const merged = new Float32Array(micRemainder.length + float32.length);
  merged.set(micRemainder);
  merged.set(float32, micRemainder.length);
  let offset = 0;
  while (offset + frame <= merged.length) {
    if (ws && ws.readyState === WebSocket.OPEN) {
      ws.send(floatToL16(merged.subarray(offset, offset + frame)));
    }
    offset += frame;
  }
  micRemainder = offset < merged.length ? merged.slice(offset) : new Float32Array(0);
}

function resample(input, fromRate, toRate) {
  if (fromRate === toRate) return input;
  const ratio = fromRate / toRate;
  const out = new Float32Array(Math.max(1, Math.round(input.length / ratio)));
  for (let i = 0; i < out.length; i++) {
    const pos = i * ratio;
    const idx = Math.floor(pos);
    const frac = pos - idx;
    const s0 = input[idx] || 0;
    const s1 = input[idx + 1] === undefined ? s0 : input[idx + 1];
    out[i] = s0 + frac * (s1 - s0);
  }
  return out;
}

function createAudioContext() {
  // Open the context at the wire rate so every buffer plays sample-aligned
  // instead of going through a per-node resampler.
  if (audioCtx && audioCtx.sampleRate !== sampleRate) {
    try { audioCtx.close(); } catch (_) {}
    audioCtx = null;
  }
  if (!audioCtx || audioCtx.state === "closed") {
    audioCtx = new AudioContext({ sampleRate });
    masterGain = audioCtx.createGain();
    masterGain.gain.value = 1;
    masterGain.connect(audioCtx.destination);
  }
  if (audioCtx.state === "suspended") void audioCtx.resume();
}

function playTts(buffer) {
  if (!opened || !audioCtx || !masterGain || !buffer.byteLength) return;
  if (dropBotUntil && Date.now() < dropBotUntil) return;
  pendingChunks.push(l16ToFloat(buffer));
  if (burstTimer) clearTimeout(burstTimer);
  burstTimer = setTimeout(flushBurst, BURST_MS);
}

function flushBurst() {
  burstTimer = null;
  const chunks = pendingChunks;
  pendingChunks = [];
  if (!opened || !audioCtx || !masterGain || !chunks.length) return;
  const total = chunks.reduce((n, c) => n + c.length, 0);
  const samples = new Float32Array(total);
  let offset = 0;
  for (const c of chunks) { samples.set(c, offset); offset += c.length; }
  if (!samples.length) return;
  const now = audioCtx.currentTime;
  masterGain.gain.cancelScheduledValues(now);
  masterGain.gain.setValueAtTime(1, now);
  const playBuffer = audioCtx.createBuffer(1, samples.length, sampleRate);
  playBuffer.getChannelData(0).set(samples);
  const startAt = Math.max(now + PLAYBACK_LOOKAHEAD_SEC, nextPlayTime);
  nextPlayTime = startAt + playBuffer.duration;
  const src = audioCtx.createBufferSource();
  src.buffer = playBuffer;
  src.connect(masterGain);
  ttsSources.add(src);
  src.onended = () => {
    ttsSources.delete(src);
    if (playbackActive && !botStillPlaying()) schedulePlaybackCompleted();
  };
  src.start(startAt);
  if (!playbackActive) {
    playbackActive = true;
    sendJson("playback_started");
    log("→ playback_started");
  }
  schedulePlaybackCompleted();
}

function botStillPlaying() {
  if (ttsSources.size > 0) return true;
  return !!(audioCtx && nextPlayTime > audioCtx.currentTime + 0.01);
}

function schedulePlaybackCompleted() {
  clearTimeout(playbackIdle);
  playbackIdle = setTimeout(() => {
    playbackIdle = null;
    if (!playbackActive) return;
    if (botStillPlaying()) {
      schedulePlaybackCompleted();
      return;
    }
    playbackActive = false;
    nextPlayTime = 0;
    sendJson("playback_completed");
    log("→ playback_completed");
  }, PLAYBACK_IDLE_MS);
}

function stopPlaybackAndClearQueue() {
  dropBotUntil = Date.now() + 2000;
  pendingChunks = [];
  clearTimeout(burstTimer);
  burstTimer = null;
  playbackGen += 1;
  nextPlayTime = 0;
  if (masterGain && audioCtx) {
    const now = audioCtx.currentTime;
    masterGain.gain.cancelScheduledValues(now);
    masterGain.gain.setValueAtTime(0, now);
  }
  for (const src of ttsSources) {
    try { src.stop(); } catch (_) {}
  }
  ttsSources.clear();
  clearTimeout(playbackIdle);
  playbackIdle = null;
  if (playbackActive) {
    playbackActive = false;
    sendJson("playback_completed");
    log("→ playback_completed");
  }
}

async function startMic() {
  createAudioContext();
  if (audioCtx.state === "suspended") await audioCtx.resume();
  mediaStream = await navigator.mediaDevices.getUserMedia({
    audio: { channelCount: 1, echoCancellation: true, noiseSuppression: true },
  });
  const source = audioCtx.createMediaStreamSource(mediaStream);
  micNode = audioCtx.createScriptProcessor(audioCtx.sampleRate <= 8000 ? 256 : 512, 1, 1);
  micNode.onaudioprocess = (event) => {
    let samples = event.inputBuffer.getChannelData(0);
    if (audioCtx.sampleRate !== sampleRate) {
      samples = resample(samples, audioCtx.sampleRate, sampleRate);
    }
    sendMic(samples);
  };
  source.connect(micNode);
  micNode.connect(audioCtx.destination);
  log("mic streaming");
}

function teardownAudio() {
  opened = false;
  dropBotUntil = 0;
  pendingChunks = [];
  clearTimeout(burstTimer);
  burstTimer = null;
  for (const src of ttsSources) {
    try { src.stop(); } catch (_) {}
  }
  ttsSources.clear();
  nextPlayTime = 0;
  try { micNode?.disconnect(); } catch (_) {}
  micNode = null;
  try { masterGain?.disconnect(); } catch (_) {}
  masterGain = null;
  mediaStream?.getTracks().forEach((track) => track.stop());
  mediaStream = null;
  micRemainder = new Float32Array(0);
  if (audioCtx) {
    audioCtx.close();
    audioCtx = null;
  }
}

/** Stop audio and close the socket without sending anything more. */
function endCall() {
  clearInterval(pingTimer);
  pingTimer = null;
  const openWs = ws;
  ws = null;
  sessionId = "";
  teardownAudio();
  if (openWs && openWs.readyState <= WebSocket.OPEN) openWs.close(1000);
  connectBtn.disabled = false;
  hangupBtn.disabled = true;
}

/** Hang up: say close if the call reached opened, give it a moment to land, then end. */
function hangup() {
  const openWs = ws;
  if (openWs && openWs.readyState === WebSocket.OPEN && sessionId && opened) {
    openWs.send(JSON.stringify({
      version: "2",
      type: "close",
      seq: ++seq,
      id: sessionId,
      parameters: { reason: "end" },
    }));
    ws = null;
    setTimeout(() => openWs.close(1000), 300);
  }
  endCall();
}

function onBargeIn() {
  log("barge_in — stop TTS and clear queue");
  stopPlaybackAndClearQueue();
}

const wsUrlInput = document.getElementById("wsUrl");
const callTokenInput = document.getElementById("callToken");
// A new key: the old example saved the console proxy URL under previewCall.wsUrl.
wsUrlInput.value = localStorage.getItem("previewCall.callTokenWsUrl") || defaultWsUrl();

function connect() {
  const wsBase = wsUrlInput.value.trim() || defaultWsUrl();
  const callToken = callTokenInput.value.trim();
  if (!callToken) {
    log("paste a call token minted on your server");
    return;
  }
  hangup();
  createAudioContext();
  let url;
  try {
    url = new URL(wsBase);
  } catch (_) {
    log("invalid WebSocket URL: " + wsBase);
    return;
  }
  url.searchParams.set("token", callToken);
  // A call token works once: clear it so the next call starts from a fresh one.
  callTokenInput.value = "";
  localStorage.setItem("previewCall.callTokenWsUrl", wsBase);
  log("connecting " + url.origin + url.pathname);
  seq = 0;
  // crypto.randomUUID needs a secure context: file:// and localhost are, a plain-http LAN host is not.
  sessionId = crypto.randomUUID();
  const socket = new WebSocket(url);
  ws = socket;
  socket.binaryType = "arraybuffer";
  connectBtn.disabled = true;
  hangupBtn.disabled = false;

  // The socket opens before the token is checked: send open, then wait for opened.
  socket.onopen = () => {
    socket.send(JSON.stringify({
      version: "2",
      type: "open",
      seq: ++seq,
      id: sessionId,
      position: "POST-ANSWER",
      parameters: {
        conversationId: "browser-" + Date.now(),
        media: [{"type":"audio/L16","sampleRateHz":16000,"channels":["external"]}],
      },
    }));
  };

  socket.onmessage = (event) => {
    const payload = event.data;
    if (payload instanceof ArrayBuffer) {
      playTts(payload);
      return;
    }
    if (typeof Blob !== "undefined" && payload instanceof Blob) {
      payload.arrayBuffer().then(playTts);
      return;
    }
    const msg = JSON.parse(payload);
    if (msg.type === "opened") {
      const media = msg.parameters && msg.parameters.media && msg.parameters.media[0];
      if (media && media.sampleRateHz) sampleRate = media.sampleRateHz;
      opened = true;
      log("opened " + sessionId + " @ " + sampleRate + " Hz");
      void startMic().catch((err) => {
        if (err && err.name === "NotAllowedError") log("mic permission denied — allow microphone in the browser");
        else if (err && err.name === "NotFoundError") log("no microphone found");
        else log(String(err));
      });
      pingTimer = setInterval(() => sendJson("ping"), 20000);
      return;
    }
    if (msg.type === "pong") return;
    if (msg.type === "barge_in") {
      onBargeIn();
      return;
    }
    if (msg.type === "event") {
      const entities = (msg.parameters && msg.parameters.entities) || [];
      log("← event " + entities.map((entity) => entity.type).join(","));
      for (const entity of entities) {
        if (entity.type === "bot_turn_response" && entity.data && entity.data.output) {
          log("bot: " + entity.data.output);
        }
        if (entity.type === "user_turn_response" && entity.data && entity.data.is_final) {
          dropBotUntil = 0;
          if (masterGain && audioCtx) {
            masterGain.gain.setValueAtTime(1, audioCtx.currentTime);
          }
          const text = entity.data.transcript && entity.data.transcript.result && entity.data.transcript.result.text;
          if (text) log("user: " + text);
        }
        if (entity.type === "barge_in") onBargeIn();
      }
      return;
    }
    if (msg.type === "closed" || msg.type === "disconnect") {
      // The call is over; frames may still arrive until the socket closes — ignore them.
      log(msg.type);
      endCall();
      return;
    }
    log("← " + msg.type);
  };
  socket.onclose = (event) => {
    const extra = event.code === 4001
      ? " (call token expired or already used — mint a new one)"
      : event.code === 1006
        ? " (abnormal close — check the URL and network)"
        : "";
    log("closed " + event.code + " " + (event.reason || "") + extra);
    if (ws === socket) endCall();
  };
  socket.onerror = () => log("ws error — cannot reach " + (wsUrlInput.value.trim() || defaultWsUrl()));
}

connectBtn.onclick = connect;
hangupBtn.onclick = hangup;
</script>
</html>

"""Realtime client: mint a call token, then talk to an agent from this machine.

Your access token stays on this machine: it mints a single-use call token,
and only the call token goes on the WebSocket.

AEC: duplex I/O so mic/speaker frames align, NLMS, auto delay, then a residual
gate — silence while the bot is in the speakers unless leftover is user speech.

pip install -r preview-call-requirements.txt
ACCESS_TOKEN=… python preview-call-example.py --agent-id agt_…
"""
from __future__ import annotations

import argparse
import asyncio
import json
import math
import os
import re
import ssl
import sys
import threading
import time
import urllib.error
import urllib.request
import uuid
from urllib.parse import quote

import numpy as np
import sounddevice as sd
import websockets

DEFAULT_WS_URL = "wss://voicebot-serverless-stg.botnoigroup.com/v1/call"
DEFAULT_MINT_URL = "https://backend-voicebot-stg.botnoigroup.com/call/tokens"
OPEN_MEDIA = [{"type":"audio/L16","sampleRateHz":16000,"channels":["external"]}]

FRAME_MS = 20
PLAYBACK_IDLE_S = 0.6
TTS_PREROLL_MS = 80
DROP_BOT_S = 2.0
MIC_Q_MAX = 5
AEC_BUDGET_S = FRAME_MS / 1000 * 0.9
AEC_DELAY_MS = int(os.environ.get("VOICEBOT_AEC_DELAY_MS", "-1"))
AEC_TAPS = int(os.environ.get("VOICEBOT_AEC_TAPS", "512"))
AEC_MAX_DELAY_MS = int(os.environ.get("VOICEBOT_AEC_MAX_DELAY_MS", "200"))


def log(*args: object) -> None:
    print(*args, flush=True)


def load_config() -> tuple[str, str, str, str, bool]:
    parser = argparse.ArgumentParser(description="Realtime voice client")
    parser.add_argument(
        "--url",
        default=os.environ.get("VOICEBOT_WS_URL", DEFAULT_WS_URL),
        help="Call Token WebSocket URL (env: VOICEBOT_WS_URL)",
    )
    parser.add_argument(
        "--mint-url",
        default=os.environ.get("VOICEBOT_MINT_URL", DEFAULT_MINT_URL),
        help="POST /call/tokens URL (env: VOICEBOT_MINT_URL)",
    )
    parser.add_argument(
        "--agent-id",
        default=os.environ.get("VOICEBOT_AGENT_ID", ""),
        help="Agent to talk to (env: VOICEBOT_AGENT_ID)",
    )
    parser.add_argument(
        "--aec",
        choices=("on", "off"),
        default=os.environ.get("VOICEBOT_AEC", "on"),
        help="Client echo canceller (default: on). Use off only on a headset.",
    )
    args = parser.parse_args()
    # The access token comes from the environment only — never a flag, which lands in shell history.
    access_token = os.environ.get("ACCESS_TOKEN", "").strip()
    if not access_token:
        sys.exit("Set ACCESS_TOKEN to your access token from the Developers page of the console")
    agent_id = args.agent_id.strip() or input("Agent id: ").strip()
    if not agent_id:
        sys.exit("An agent id is required")
    ws_base = args.url.strip().rstrip("/")
    if not ws_base.startswith(("ws://", "wss://")):
        sys.exit(f"--url must start with ws:// or wss://, got {ws_base!r}")
    return ws_base, args.mint_url.strip(), access_token, agent_id, args.aec == "on"


def mint_call_token(mint_url: str, access_token: str, agent_id: str) -> str:
    """POST /call/tokens with the access token; returns the single-use call token."""
    request = urllib.request.Request(
        mint_url,
        data=json.dumps({"agent_id": agent_id, "channel": "web"}).encode(),
        headers={
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json",
            # Some firewalls refuse urllib's default "Python-urllib/3.x" User-Agent; name yours.
            "User-Agent": "voicebot-preview-call-example/1.0",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=15, context=make_ssl_context()) as response:
            body = json.loads(response.read())
    except urllib.error.HTTPError as exc:
        try:
            detail = json.loads(exc.read()).get("detail")
        except Exception:
            detail = None
        sys.exit(f"Could not mint a call token: HTTP {exc.code} {detail or ''}".rstrip())
    except (urllib.error.URLError, OSError) as exc:
        sys.exit(f"Could not reach {mint_url}: {explain_error(exc)}")
    token = body.get("ephemeral_token")
    if not isinstance(token, str) or not token:
        sys.exit("Could not mint a call token: no ephemeral_token in the response")
    return token


def make_ssl_context() -> ssl.SSLContext:
    try:
        import certifi
    except ImportError:
        log("warning: certifi not installed; Mac users may hit SSL errors. pip install certifi")
        return ssl.create_default_context()
    return ssl.create_default_context(cafile=certifi.where())


def ensure_audio() -> None:
    try:
        sd.query_devices()
    except Exception as exc:
        sys.exit(
            "Cannot open audio (sounddevice/PortAudio).\n"
            "  Linux:   sudo apt install libportaudio2\n"
            "  macOS:   allow mic/speaker for Terminal in System Settings\n"
            "  Windows: check the default microphone and speakers\n"
            f"  Detail:  {exc}"
        )


def explain_error(exc: BaseException) -> str:
    name = type(exc).__name__
    # A malformed URL makes websockets quote it whole, call token included: never print that.
    msg = re.sub(r"token=[^&\s'\"]+", "token=***", str(exc))
    lower = f"{name} {msg}".lower()
    if "certificate" in lower or "ssl" in name.lower():
        return (
            "SSL certificate verify failed. Install certifi (`pip install certifi`) "
            "or check a corporate proxy intercepting TLS."
        )
    if any(s in msg for s in ("PortAudio", "Invalid number of channels", "No Default", "PaError")):
        return (
            "Audio device error. On Linux install libportaudio2; "
            "elsewhere check mic/speaker permissions."
        )
    if "4001" in lower:
        return "Call token expired or already used — mint a new one."
    if any(s in lower for s in ("timed out", "timeout", "name or service", "nodename", "getaddrinfo")):
        return f"Cannot reach the server. Check the URL and network.\n  Detail: {name}: {msg}"
    return f"{name}: {msg}"


class Seq:
    def __init__(self) -> None:
        self.value = 0

    def next(self) -> int:
        self.value += 1
        return self.value


class TtsBuffer:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._buf = bytearray()
        self._primed = False
        self.preroll_bytes = 0

    def set_preroll(self, nbytes: int) -> None:
        with self._lock:
            self.preroll_bytes = max(0, nbytes)

    def push(self, data: bytes) -> None:
        with self._lock:
            self._buf.extend(data)

    def clear(self) -> None:
        with self._lock:
            self._buf.clear()
            self._primed = False

    def pull(self, nbytes: int) -> bytes:
        with self._lock:
            if not self._primed:
                if len(self._buf) < max(self.preroll_bytes, nbytes):
                    return b""
                self._primed = True
            n = min(nbytes, len(self._buf))
            chunk = bytes(self._buf[:n])
            del self._buf[:n]
            return chunk

    def pending(self) -> int:
        with self._lock:
            return len(self._buf)

    def is_primed(self) -> bool:
        with self._lock:
            return self._primed

    def mark_idle(self) -> None:
        with self._lock:
            self._primed = False


class PlayedRef:
    """Samples that actually went to the speaker — the AEC far-end."""

    def __init__(self, maxlen: int) -> None:
        self._lock = threading.Lock()
        self._buf = np.zeros(maxlen, dtype=np.float32)
        self._n = 0
        self.maxlen = maxlen

    def push_bytes(self, pcm: bytes) -> None:
        if not pcm:
            return
        samples = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) * (1.0 / 32768.0)
        n = int(samples.size)
        with self._lock:
            start = self._n % self.maxlen
            end = start + n
            if end <= self.maxlen:
                self._buf[start:end] = samples
            else:
                first = self.maxlen - start
                self._buf[start:] = samples[:first]
                self._buf[: n - first] = samples[first:]
            self._n += n

    def recent(self, count: int, delay: int) -> np.ndarray:
        out = np.zeros(count, dtype=np.float32)
        with self._lock:
            start = self._n - delay - count
            if start < 0:
                valid = count + start
                if valid <= 0:
                    return out
                start = 0
                dest = count - valid
            else:
                dest = 0
                valid = count
            idx = start % self.maxlen
            take = valid
            first = min(take, self.maxlen - idx)
            out[dest : dest + first] = self._buf[idx : idx + first]
            if first < take:
                out[dest + first : dest + take] = self._buf[: take - first]
        return out


class NlmsAec:
    """NLMS with a ring buffer — no per-sample memmove."""

    def __init__(self, taps: int = 512, mu: float = 0.12) -> None:
        self.taps = int(taps)
        self.w = np.zeros(self.taps, dtype=np.float32)
        self.x = np.zeros(self.taps, dtype=np.float32)
        self.i = 0
        self.mu = np.float32(mu)

    def reset(self) -> None:
        self.w.fill(0)
        self.x.fill(0)
        self.i = 0

    def process(self, near: np.ndarray, far: np.ndarray, adapt: bool = True) -> np.ndarray:
        err = np.empty(len(near), dtype=np.float32)
        w = self.w
        x = self.x
        taps = self.taps
        i = self.i
        mu = self.mu
        for k in range(len(near)):
            i = (i + 1) % taps
            x[i] = far[k]
            head = x[i::-1]
            n_head = int(head.size)
            echo = float(w[:n_head] @ head)
            pwr = float(head @ head)
            if n_head < taps:
                tail = x[:i:-1]
                echo += float(w[n_head:] @ tail)
                pwr += float(tail @ tail)
            e = float(near[k] - echo)
            if adapt and pwr > 1e-6:
                scale = mu * np.float32(e) / (pwr + 1e-4)
                w[:n_head] += scale * head
                if n_head < taps:
                    w[n_head:] += scale * x[:i:-1]
            err[k] = e
        self.i = i
        return err


class EchoCanceller:
    def __init__(self, sample_rate: int, taps: int, delay_ms: int = -1) -> None:
        self.sr = sample_rate
        self.played = PlayedRef(sample_rate * 3)
        self.nlms = NlmsAec(taps=taps)
        self.max_delay = max(1, int(sample_rate * AEC_MAX_DELAY_MS / 1000))
        self.fixed_delay = delay_ms >= 0
        self.delay = max(0, int(sample_rate * (delay_ms if delay_ms >= 0 else 20) / 1000))
        self._frame = 0

    def on_played(self, pcm: bytes) -> None:
        self.played.push_bytes(pcm)

    def _maybe_update_delay(self, near: np.ndarray) -> None:
        if self.fixed_delay:
            return
        self._frame += 1
        if self._frame % 12 != 0:
            return
        n = int(near.size)
        hist = self.played.recent(n + self.max_delay, 0)
        if float(np.dot(hist, hist)) < 1e-4:
            return
        corr = np.correlate(hist, near, mode="valid")
        peak_i = int(np.argmax(np.abs(corr)))
        peak = float(np.abs(corr[peak_i]))
        med = float(np.median(np.abs(corr))) + 1e-8
        if peak < 4.0 * med:
            return
        new_delay = int(np.clip(self.max_delay - peak_i, 0, self.max_delay))
        self.delay = int(0.7 * self.delay + 0.3 * new_delay)

    def process(self, pcm: bytes, playback_active: bool = False) -> bytes:
        t0 = time.perf_counter()
        near = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) * (1.0 / 32768.0)
        n = int(near.size)
        silence = b"\x00" * (n * 2)
        self._maybe_update_delay(near)
        if time.perf_counter() - t0 > AEC_BUDGET_S:
            return silence
        far = self.played.recent(n, self.delay)
        far_pow = float(np.mean(far * far))
        near_pow = float(np.mean(near * near))
        speaking = playback_active or far_pow >= 1e-7
        if not speaking:
            return pcm
        denom = math.sqrt(max(near_pow * far_pow, 0.0)) * n + 1e-8
        corr = abs(float(np.dot(near, far))) / denom
        if playback_active:
            double_talk = near_pow > 6e-4 and corr < 0.08
        else:
            double_talk = near_pow > 3e-4 and corr < 0.12
        err = self.nlms.process(near, far, adapt=not double_talk)
        if time.perf_counter() - t0 > AEC_BUDGET_S:
            return silence
        err_pow = float(np.mean(err * err))
        if playback_active:
            user_speech = double_talk and err_pow > 1.5e-3 and err_pow > 0.6 * near_pow
        else:
            user_speech = double_talk and err_pow > 8e-4 and err_pow > 0.5 * near_pow
        if not user_speech:
            return silence
        return np.clip(err * 32768.0, -32768, 32767).astype(np.int16).tobytes()


async def main(ws_base: str, call_token: str, aec_enabled: bool) -> None:
    url = f"{ws_base}?token={quote(call_token)}"
    seq = Seq()
    session_id = str(uuid.uuid4())
    closed_by_server = False
    sample_rate = 16000
    mic_q: asyncio.Queue[bytes] = asyncio.Queue(maxsize=MIC_Q_MAX)
    tts = TtsBuffer()
    aec = EchoCanceller(sample_rate, taps=AEC_TAPS, delay_ms=AEC_DELAY_MS)
    playback_active = False
    drop_bot_until = 0.0
    stream = None
    connect_kwargs: dict = {"max_size": 2**20, "open_timeout": 20}
    if ws_base.startswith("wss://"):
        connect_kwargs["ssl"] = make_ssl_context()

    log(f"connecting {ws_base}")
    async with websockets.connect(url, **connect_kwargs) as ws:
        loop = asyncio.get_running_loop()
        # The socket opens before the token is checked: send open, then wait for opened.
        await ws.send(
            json.dumps(
                {
                    "version": "2",
                    "type": "open",
                    "seq": seq.next(),
                    "id": session_id,
                    "position": "POST-ANSWER",
                    "parameters": {"conversationId": f"python-{int(time.time())}", "media": OPEN_MEDIA},
                }
            )
        )
        log("sent open, waiting for opened")

        async def send_json(typ: str, parameters: dict | None = None) -> None:
            await ws.send(
                json.dumps(
                    {
                        "version": "2",
                        "type": typ,
                        "seq": seq.next(),
                        "id": session_id,
                        "parameters": parameters or {},
                    }
                )
            )

        def enqueue_mic(pcm: bytes) -> None:
            while mic_q.qsize() >= MIC_Q_MAX:
                try:
                    mic_q.get_nowait()
                except asyncio.QueueEmpty:
                    break
            try:
                mic_q.put_nowait(pcm)
            except asyncio.QueueFull:
                pass

        def on_duplex(indata, outdata, frames, time_info, status) -> None:
            need = frames * 2
            chunk = tts.pull(need)
            if len(chunk) < need:
                chunk = chunk + b"\x00" * (need - len(chunk))
            outdata[:need] = chunk
            aec.on_played(chunk)
            loop.call_soon_threadsafe(enqueue_mic, bytes(indata[:need]))

        async def stop_playback_and_clear_queue() -> None:
            nonlocal playback_active, drop_bot_until
            drop_bot_until = time.monotonic() + DROP_BOT_S
            tts.clear()
            if playback_active:
                playback_active = False
                await send_json("playback_completed")
                log("→ playback_completed")
            log("TTS cleared")

        async def playback_watch() -> None:
            nonlocal playback_active
            idle_since = None
            while True:
                await asyncio.sleep(0.03)
                pending = tts.pending()
                # Wait until preroll is playing — an empty queue before start
                # would otherwise fire completed before anything plays.
                if pending > 0 and stream is not None:
                    idle_since = None
                    if tts.is_primed() and not playback_active:
                        playback_active = True
                        await send_json("playback_started")
                        log("→ playback_started")
                    continue
                if not playback_active:
                    continue
                if idle_since is None:
                    idle_since = time.monotonic()
                elif time.monotonic() - idle_since >= PLAYBACK_IDLE_S:
                    playback_active = False
                    idle_since = None
                    tts.mark_idle()
                    await send_json("playback_completed")
                    log("→ playback_completed")

        async def mic_sender() -> None:
            while True:
                pcm = await mic_q.get()
                if not aec_enabled:
                    await ws.send(pcm)
                    continue
                t0 = time.perf_counter()
                out = aec.process(pcm, playback_active=playback_active)
                if time.perf_counter() - t0 > AEC_BUDGET_S:
                    out = b"\x00" * len(pcm)
                await ws.send(out)

        watch_task = asyncio.create_task(playback_watch())
        mic_task = asyncio.create_task(mic_sender())

        try:
            async for raw in ws:
                if isinstance(raw, bytes):
                    if time.monotonic() < drop_bot_until:
                        continue
                    tts.push(raw)
                    continue
                msg = json.loads(raw)
                typ = msg.get("type")
                if typ == "opened":
                    media = (msg.get("parameters") or {}).get("media") or []
                    if media and isinstance(media[0], dict):
                        rate = media[0].get("sampleRateHz") or media[0].get("rate")
                        if rate:
                            sample_rate = int(rate)
                    frame = max(1, int(sample_rate * FRAME_MS / 1000))
                    tts.set_preroll(max(1, int(sample_rate * TTS_PREROLL_MS / 1000)) * 2)
                    aec.sr = sample_rate
                    aec.max_delay = max(1, int(sample_rate * AEC_MAX_DELAY_MS / 1000))
                    if AEC_DELAY_MS >= 0:
                        aec.delay = int(sample_rate * AEC_DELAY_MS / 1000)
                    log(
                        f"opened {session_id} @ {sample_rate} Hz — duplex AEC "
                        f"{'on' if aec_enabled else 'off'} "
                        f"(delay={'auto' if AEC_DELAY_MS < 0 else str(AEC_DELAY_MS) + 'ms'} "
                        f"taps={AEC_TAPS} preroll={TTS_PREROLL_MS}ms)"
                    )
                    try:
                        stream = sd.RawStream(
                            samplerate=sample_rate,
                            channels=1,
                            dtype="int16",
                            blocksize=frame,
                            callback=on_duplex,
                        )
                        stream.start()
                    except Exception as exc:
                        log(explain_error(exc))
                        break

                    async def ping() -> None:
                        while True:
                            await asyncio.sleep(20)
                            await send_json("ping")

                    asyncio.create_task(ping())
                    continue

                if typ == "barge_in":
                    log("barge_in — stop TTS and clear queue")
                    await stop_playback_and_clear_queue()
                    continue

                if typ == "event":
                    entities = (msg.get("parameters") or {}).get("entities") or []
                    log("← event", [entity.get("type") for entity in entities])
                    for entity in entities:
                        data = entity.get("data") or {}
                        if entity.get("type") == "bot_turn_response" and data.get("output"):
                            log("bot:", data["output"])
                        if entity.get("type") == "user_turn_response" and data.get("is_final"):
                            text = (
                                (data.get("transcript") or {})
                                .get("result", {})
                                .get("text")
                            )
                            if text:
                                log("user:", text)
                        if entity.get("type") == "barge_in":
                            log("barge_in — stop TTS and clear queue")
                            await stop_playback_and_clear_queue()
                    continue

                if typ == "pong":
                    continue

                if typ in {"closed", "disconnect"}:
                    # The call is over; frames may still arrive until the socket closes.
                    log(typ, msg.get("parameters") or {})
                    closed_by_server = True
                    break
        finally:
            watch_task.cancel()
            mic_task.cancel()
            if stream:
                stream.stop()
                stream.close()
            tts.clear()
            if stream is not None and not closed_by_server:
                await send_json("close", {"reason": "end"})


if __name__ == "__main__":
    ws_base, mint_url, access_token, agent_id, aec_enabled = load_config()
    ensure_audio()
    call_token = mint_call_token(mint_url, access_token, agent_id)
    log("minted a call token")
    try:
        asyncio.run(main(ws_base, call_token, aec_enabled))
    except KeyboardInterrupt:
        log("\nbye")
    except Exception as exc:
        sys.exit(explain_error(exc))

/**
 * Realtime Node client — mints a call token, then 20 ms L16 mic frames + TTS playback.
 * Your access token stays in this process; only the single-use call token goes on the WebSocket.
 *
 *   npm i ws
 *   ffmpeg on PATH (Windows: set VOICEBOT_DSHOW_DEVICE=audio=Your Mic Name)
 *
 *   ACCESS_TOKEN=… VOICEBOT_AGENT_ID=agt_… node preview-call-example.mjs
 */
import { spawn } from "node:child_process";
import { randomUUID } from "node:crypto";
import WebSocket from "ws";

const WS_BASE = process.env.VOICEBOT_WS_URL ?? "wss://voicebot-serverless-stg.botnoigroup.com/v1/call";
const MINT_URL = process.env.VOICEBOT_MINT_URL ?? "https://backend-voicebot-stg.botnoigroup.com/call/tokens";
const ACCESS_TOKEN = process.env.ACCESS_TOKEN;
const AGENT_ID = process.env.VOICEBOT_AGENT_ID;
if (!ACCESS_TOKEN || !AGENT_ID) {
  console.error("Set ACCESS_TOKEN (from the Developers page of the console) and VOICEBOT_AGENT_ID");
  process.exit(1);
}

const FRAME_MS = 20;
const PLAYBACK_IDLE_MS = 600;

/** POST /call/tokens with the access token; returns the single-use call token. */
async function mintCallToken() {
  let response;
  try {
    response = await fetch(MINT_URL, {
      method: "POST",
      headers: { Authorization: `Bearer ${ACCESS_TOKEN}`, "Content-Type": "application/json" },
      body: JSON.stringify({ agent_id: AGENT_ID, channel: "web" }),
    });
  } catch (err) {
    console.error(`Could not reach ${MINT_URL}: ${err.cause?.message ?? err.message}`);
    process.exit(1);
  }
  const body = await response.json().catch(() => ({}));
  if (!response.ok || typeof body.ephemeral_token !== "string") {
    console.error(`Could not mint a call token: HTTP ${response.status} ${body.detail ?? ""}`.trim());
    process.exit(1);
  }
  return body.ephemeral_token;
}

const callToken = await mintCallToken();
console.log("minted a call token");
const url = new URL(WS_BASE);
url.search = "";
url.searchParams.set("token", callToken);

let seq = 0;
let opened = false;
let hangingUp = false;
const sessionId = randomUUID();
let sampleRate = 16000;
let mic = null;
let play = null;
let playIdle = null;
let playbackActive = false;
let pingTimer = null;
let frameBuf = Buffer.alloc(0);

function sendJson(ws, type, parameters = {}) {
  if (ws.readyState !== WebSocket.OPEN) return;
  ws.send(
    JSON.stringify({
      version: "2",
      type,
      seq: ++seq,
      id: sessionId,
      parameters,
    }),
  );
}

function frameBytes(rate) {
  return Math.max(1, Math.round((rate * FRAME_MS) / 1000)) * 2;
}

function micArgs(rate) {
  const out = ["-hide_banner", "-loglevel", "error", "-ac", "1", "-ar", String(rate), "-f", "s16le", "pipe:1"];
  if (process.platform === "darwin") return ["-f", "avfoundation", "-i", ":0", ...out];
  if (process.platform === "linux") return ["-f", "pulse", "-i", "default", ...out];
  const device = process.env.VOICEBOT_DSHOW_DEVICE?.trim() || "audio=Microphone";
  return ["-f", "dshow", "-i", device, ...out];
}

function startMic(ws, rate) {
  const proc = spawn("ffmpeg", micArgs(rate), { stdio: ["ignore", "pipe", "pipe"] });
  const size = frameBytes(rate);
  proc.stderr.on("data", (chunk) => {
    const text = String(chunk).trim();
    if (text) console.error("[ffmpeg mic]", text);
  });
  proc.stdout.on("data", (chunk) => {
    frameBuf = Buffer.concat([frameBuf, chunk]);
    while (frameBuf.length >= size) {
      const frame = frameBuf.subarray(0, size);
      frameBuf = frameBuf.subarray(size);
      if (ws.readyState === WebSocket.OPEN) ws.send(frame);
    }
  });
  return proc;
}

function ensurePlayer(rate) {
  if (play && !play.killed) return play;
  const args =
    process.platform === "win32"
      ? ["-nodisp", "-autoexit", "-loglevel", "error", "-f", "s16le", "-ar", String(rate), "-ac", "1", "-i", "pipe:0"]
      : [
          "-hide_banner",
          "-loglevel",
          "error",
          "-f",
          "s16le",
          "-ar",
          String(rate),
          "-ac",
          "1",
          "-i",
          "pipe:0",
          ...(process.platform === "darwin" ? ["-f", "audiotoolbox", "-"] : ["-f", "pulse", "default"]),
        ];
  const bin = process.platform === "win32" ? "ffplay" : "ffmpeg";
  play = spawn(bin, args, { stdio: ["pipe", "ignore", "pipe"] });
  play.stderr.on("data", (chunk) => {
    const text = String(chunk).trim();
    if (text) console.error("[play]", text);
  });
  play.on("close", () => {
    play = null;
  });
  return play;
}

function stopPlaybackAndClearQueue(ws) {
  clearTimeout(playIdle);
  playIdle = null;
  if (play) {
    try { play.stdin.destroy(); } catch {}
    try { play.kill("SIGKILL"); } catch {}
    play = null;
  }
  if (playbackActive) {
    playbackActive = false;
    sendJson(ws, "playback_completed");
  }
}

function pushTts(ws, chunk, rate) {
  const proc = ensurePlayer(rate);
  if (!playbackActive) {
    playbackActive = true;
    sendJson(ws, "playback_started");
  }
  try {
    proc.stdin.write(chunk);
  } catch {
    play = null;
  }
  clearTimeout(playIdle);
  playIdle = setTimeout(() => {
    if (!playbackActive) return;
    playbackActive = false;
    sendJson(ws, "playback_completed");
  }, PLAYBACK_IDLE_MS);
}

function cleanup(ws, sendClose) {
  clearInterval(pingTimer);
  pingTimer = null;
  mic?.kill("SIGTERM");
  mic = null;
  if (play && !play.killed) {
    play.stdin.end();
    play.kill("SIGTERM");
  }
  play = null;
  if (sendClose && opened && ws.readyState === WebSocket.OPEN) {
    sendJson(ws, "close", { reason: "end" });
  }
}

const ws = new WebSocket(url, { maxPayload: 2 ** 20 });
console.log("connecting", url.origin + url.pathname);

// The socket opens before the token is checked: send open, then wait for opened.
ws.on("open", () => {
  ws.send(
    JSON.stringify({
      version: "2",
      type: "open",
      seq: ++seq,
      id: sessionId,
      position: "POST-ANSWER",
      parameters: { conversationId: `node-${Date.now()}`, media: [{"type":"audio/L16","sampleRateHz":16000,"channels":["external"]}] },
    }),
  );
});

ws.on("message", (data, isBinary) => {
  if (isBinary) {
    pushTts(ws, Buffer.isBuffer(data) ? data : Buffer.from(data), sampleRate);
    return;
  }
  const msg = JSON.parse(String(data));
  if (msg.type === "opened") {
    if (hangingUp) return;
    opened = true;
    const media = msg.parameters?.media?.[0];
    const rate = media?.sampleRateHz ?? media?.rate;
    if (rate) sampleRate = Number(rate);
    console.log(`opened ${sessionId} @ ${sampleRate} Hz — streaming mic`);
    mic = startMic(ws, sampleRate);
    pingTimer = setInterval(() => sendJson(ws, "ping"), 20_000);
    return;
  }
  if (msg.type === "event") {
    for (const entity of msg.parameters?.entities ?? []) {
      if (entity.type === "bot_turn_response" && entity.data?.output) {
        console.log("bot:", entity.data.output);
      }
      if (entity.type === "user_turn_response" && entity.data?.is_final) {
        const text = entity.data.transcript?.result?.text;
        if (text) console.log("user:", text);
      }
      if (entity.type === "barge_in") {
        console.log("barge_in — stop TTS and clear queue");
        stopPlaybackAndClearQueue(ws);
      }
    }
    return;
  }
  if (msg.type === "closed" || msg.type === "disconnect") {
    console.log(msg.type, msg.parameters ?? {});
    cleanup(ws, false);
    if (hangingUp && msg.type === "closed") ws.close(1000);
  }
});

ws.on("close", (code) => {
  if (code === 4001) console.error("call token expired or already used — mint a new one");
  cleanup(ws, false);
  // Your own hang-up ends with 1000, or 1005 if the server's reply carries no code.
  process.exit(code === 1000 || (hangingUp && code === 1005) ? 0 : 1);
});
ws.on("error", (err) => {
  console.error(err.message);
  cleanup(ws, false);
  process.exit(1);
});
// Ctrl+C hangs up: send close, wait for closed, then close the socket with 1000.
// A second Ctrl+C gives up waiting and quits at once.
process.on("SIGINT", () => {
  if (hangingUp) {
    ws.terminate();
    process.exit(130);
  }
  hangingUp = true;
  cleanup(ws, true);
  setTimeout(() => ws.close(1000), 2000);
});

Realtime

Connect

1. Mint a call token on your server

2. Connect from the browser

The open you send

Rules

Migrating from /v1/preview_call

Binary audio

Client messages

Server messages

Errors

Minting a call token

During the call

Message sequence

The `open` you send

Migrating from `/v1/preview_call`