Skip to content

Python Websocket Module

The Connect to AI Agent samples are built on a small, reusable Python package — aculab.websocket — that implements the Aculab Cloud WSS media-streaming protocol once, so AI-specific bridges can concentrate on driving the downstream AI session. It is deliberately backend-agnostic: the same package is used by the Gemini Live sample and the OpenAI Realtime sample, and can be paired with any service that exposes a streaming audio API.

The source lives in the ip-acu-aculab-websocket repository. The runnable AI-agent samples referenced above live in the samples/ tree of the same repository.

Installation

The single top-level Python module is available as a wheel package from Downloads

And can be installed using pip install

then import from it:

from aculab.websocket import (
    AculabCloudWebsocketMixin,
    select_aculab_subprotocol,
    run_aculab_echo,
)

Runtime requirements: Python 3.11+ and websockets 12.0 or later.

What the module provides

aculab.websocket exports three top-level elements:

  • AculabCloudWebsocketMixin — the main class. It owns all of the JSON control framing, subprotocol-aware receive loop, playback chunking and per-connection lifecycle. A bridge is built by subclassing this mixin and overriding a small number of hooks.
  • select_aculab_subprotocol(connection, client_subprotocols) — a subprotocol negotiator to pass to websockets.serve(select_subprotocol=…). It selects v2.ws.cloud.aculab.com for request paths beginning with /v2/ and v1.ws.cloud.aculab.com otherwise, and rejects the handshake if neither is offered by Aculab Cloud.
  • run_aculab_echo(aculab_ws, log=…) — a loopback (echo) handler used for connectivity testing. It responds to audio_start with an audio_play_start and then reflects every binary audio frame back to the caller until audio_end or the connection closes. AculabCloudWebsocketMixin.serve routes any request path ending in echo to this handler automatically, so pointing a Connect action at wss://<host>:<port>/v2/echo gives an instant end-to-end audio path check without touching the AI backend.

The module also exposes a handful of constants that document Aculab Cloud's contract and can be reused by subclasses:

  • ACULAB_SUBPROTOCOL_V1
  • ACULAB_SUBPROTOCOL_V2
  • ACULAB_SUPPORTED_FORMAT — "16bit_PCM", the only caller-audio format the mixin bridges.
  • ACULAB_PLAY_CHUNK_MS — 20 ms, the slice size used when chunking outbound PCM to stay under Aculab Cloud's 1600-byte binary-frame limit.
  • DEFAULT_CALLER_RATE — 8000 Hz.
  • DEFAULT_INITIAL_BUFFER_MS — 300 ms, the jitter-buffer prewarm sent in audio_play_start.

What the mixin does for you

AculabCloudWebsocketMixin implements the whole Aculab-facing side of the bridge:

  • Handshake and subprotocol negotiation through select_aculab_subprotocol, wired up by serve below.
  • Inbound JSON control framing. The receive loop parses each text frame and dispatches on type:
    • audio_start — records the announced format, sample_rate and channels on the instance (self.caller_format, self.caller_rate, self.caller_channels), sets self.audio_running = True and calls on_caller_audio_start(rate, fmt, channels). If the announced format is anything other than 16bit_PCM the WS is closed with WS 1003 — the mixin does not decode µ-law / A-law.
    • audio_end — clears self.audio_running and calls on_caller_audio_end().
    • call_hangup — calls on_caller_hangup() and closes the WS.
  • Inbound binary audio. Each binary frame is passed straight to on_caller_audio_frame(pcm) (only while audio_running is set and the format is 16bit_PCM), so subclasses only ever see 16-bit mono PCM at self.caller_rate.
  • Outbound playback framing. send_playback_audio(pcm) lazily sends an audio_play_start on the first call (announcing 16bit_PCM at self.caller_rate), then slices the buffer into ~20 ms chunks so no binary frame exceeds Aculab Cloud's 1600-byte limit. send_playback_abort() and send_playback_end() emit the matching JSON control frames and reset the internal _playing flag so a subsequent send_playback_audio will start a fresh play.
  • Jitter-buffer prewarm. The default on_caller_audio_start calls start_playback_stream(), which sends audio_play_start immediately with a 300 ms initial_buffer_ms. Combined with drain_until_audio_start, this lets Aculab Cloud start filling its playback buffer while the (potentially slow) AI connect handshake is still in progress, hiding some of the AI's first-token latency from the caller.
  • Per-connection lifecycle. run() drains inbound frames until audio_start is seen (with a 2 s timeout), then concurrently runs the Aculab receive loop and run_session(), waits on self.closed, cancels both tasks and closes the WS.
  • Server plumbing. AculabCloudWebsocketMixin.serve(host, port, ssl=…, factory=…, stop=…) starts a websockets server with the subprotocol negotiator wired up, dispatches echo paths to run_aculab_echo and instantiates the subclass (or a caller-supplied factory) for every other connection. It runs forever unless stop is provided.

The mixin also exposes one attribute the owning coordinator can rely on:

  • self.closed — an asyncio.Event that is set when the Aculab side has finished, whether because of a peer close, a call_hangup, or a send failure. Subclasses should also set it when the downstream AI session decides the conversation is over, so run() tears both sides down together.

How to use it

A minimal bridge is built in three steps:

1. Subclass the mixin and initialise it. The constructor must call self._init_aculab(aculab_ws) before anything else — that installs the WS, sets defaults for caller_rate / caller_format / caller_channels and creates self.closed. Pass an optional log= keyword argument to route the mixin's log output through your own logging.Logger.

import asyncio

from aculab.websocket import AculabCloudWebsocketMixin


class MyBridge(AculabCloudWebsocketMixin):
    def __init__(self, aculab_ws):
        self._init_aculab(aculab_ws)
        # create whatever queues / clients the downstream AI needs, e.g.
        self.to_ai = asyncio.Queue()

2. Override the hooks you care about. Only on_caller_audio_frame is mandatory; the rest have sensible defaults:

Hook When it fires Typical override
on_caller_audio_frame(pcm: bytes) Every inbound binary frame while audio_running is set Resample from self.caller_rate to the AI's input rate and push the buffer onto the AI input queue.
on_caller_audio_start(rate, fmt, channels) On the audio_start JSON frame Default calls start_playback_stream() to prewarm playback; override if you want to defer that.
on_caller_audio_end() On the audio_end JSON frame Flush any queued audio to the AI.
on_caller_hangup() On the call_hangup JSON frame Cancel the AI session and set self.closed.
run_session() Once per connection, in parallel with the receive loop Open the AI session, iterate its event stream, and drive playback back to the caller (see below).

3. Drive playback from run_session. run_session is where the bridge lives. It should open the AI session, consume caller audio from whatever queue was populated in on_caller_audio_frame, and forward AI audio back to the caller using the mixin's send helpers:

async def run_session(self):
    async with open_ai_session(...) as ai:
        async for event in ai.events():
            if event.type == "audio":
                # event.pcm is 16-bit PCM at the AI's output rate; resample
                # to self.caller_rate before forwarding.
                await self.send_playback_audio(resample(event.pcm, ...))
            elif event.type == "interrupted":
                await self.send_playback_abort()
            elif event.type == "turn_complete":
                await self.send_playback_end()
            elif event.type == "closed":
                break
    self.closed.set()

The three send helpers cover the whole outbound side of the Aculab contract:

  • send_playback_audio(pcm) — 16-bit mono PCM at self.caller_rate; chunking, audio_play_start and frame-size limits are handled for you.
  • send_playback_abort() — cancel the current play (use on barge-in / interruption events from the AI).
  • send_playback_end() — mark the end of the current AI turn.

4. Start the server. serve handles TLS, subprotocol negotiation, echo routing and per-connection instantiation:

import ssl

ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.load_cert_chain("cloud/localhost.pem")

asyncio.run(MyBridge.serve("0.0.0.0", 8443, ssl=ctx))

Point an Aculab Cloud REST Connect action at wss://<host>:8443/v2/<anything> with audio format 16bit_PCM, and each call will get its own MyBridge instance. wss://<host>:8443/v2/echo gives the built-in loopback for smoke-testing the audio path independently of the AI backend.

Complete worked samples

Runnable AI-agent samples that use this package are included in ip-acu-aculab-websocket/samples/: