Implementing WebSocket Client Connections with OpenAI Realtime Protocol in the Speech-to-Speech Library

To implement a WebSocket client connection with the OpenAI Realtime protocol in the huggingface/speech-to-speech repository, you connect to the /v1/realtime endpoint, exchange JSON events following the OpenAI schema, and stream PCM audio chunks while the server manages a dedicated PipelineUnit for your session.

The speech-to-speech library provides a complete, production-ready implementation of the OpenAI Realtime API over WebSocket (and WebRTC) transports. It isolates each client session in a pooled PipelineUnit that coordinates speech-to-text (STT), language model (LLM), and text-to-speech (TTS) components through an asynchronous event-driven architecture.

Architecture Overview

When a client connects, the system claims a pipeline unit from a managed pool and binds it to a RealtimeService instance that translates OpenAI protocol events into internal pipeline messages. This design ensures that state, queues, and handlers never leak between concurrent sessions.

Key components include:

• PipelineUnit (src/speech_to_speech/api/openai_realtime/pipeline_unit.py) – Holds all inter-process queues (input_queue, text_output_queue, output_queue), a RealtimeService instance, and a SessionState object that tracks the transport layer and session ID.
• RealtimeService (src/speech_to_speech/api/openai_realtime/service.py) – The core protocol engine that parses client events via parse_client_event(), validates them against internal models, updates per-connection ConnState, and dispatches events to specialized handlers (audio, conversation, response, session).
• WebSocketTransport (src/speech_to_speech/api/openai_realtime/transports.py) – Implements the concrete send_events() and send_audio_chunk() methods used by the send loop to stream data back to the client.
• Send loop (_send_loop_for in websocket_router.py) – Continuously drains the text-output and audio-output queues, batches raw PCM bytes up to MAX_AUDIO_BATCH_BYTES, and forwards them to the active transport while respecting SESSION_END propagation for graceful shutdown.

WebSocket Connection Lifecycle

Understanding the data flow through the server is essential for implementing a compatible client.

1. Connection and Session Registration

A client opens a WebSocket to /v1/realtime. The FastAPI route handler in websocket_router.py accepts the socket, wraps it in a WebSocketTransport, and calls _claim_unit() to acquire an idle PipelineUnit. It then invokes unit.service.register() to generate a unique session ID stored in SessionState.session_id. Immediately after registration, the server emits a session.created event to the client.

2. Event Ingestion

The client sends JSON events such as input_audio_buffer.append or conversation.item.create. The route handler calls _dispatch_client_event(), which uses RealtimeService.parse_client_event() to map raw JSON into typed OpenAI events. Valid events are routed to handler methods like handle_audio_append() or handle_conversation_item_create(), which push data onto the unit’s queues (unit.input_queue, unit.text_prompt_queue).

3. Pipeline Processing

Background workers consume from these queues and produce pipeline events (AssistantTextEvent, AudioOutput, SpeechStartedEvent). These internal events travel through the handler chain defined in src/speech_to_speech/api/openai_realtime/handlers/__init__.py.

4. Response Streaming

The _send_loop_for function reads from unit.text_output_queue first (to preserve ordering) and then from unit.output_queue. It dispatches text events via transport.send_events() and batches audio chunks via transport.send_audio_chunk().

5. Graceful Shutdown

When the client disconnects, _release_session() flushes all queues, enqueues a SESSION_END sentinel (defined in src/speech_to_speech/pipeline/control.py), and spawns _release_unit_after_drain(). The send loop sets session.drained once the sentinel traverses the handler chain, after which the unit returns to the idle pool.

Implementing a Custom WebSocket Client

Any WebSocket-capable language can connect to the server by following the OpenAI Realtime JSON schema. Below is a minimal Python implementation using the websockets library.

import asyncio
import json
import websockets

API_URL = "ws://localhost:8000/v1/realtime"

def make_event(event_type, **kwargs):
    """Helper to build OpenAI Realtime events."""
    payload = {"type": event_type, **kwargs}
    return json.dumps(payload)

async def realtime_client():
    async with websockets.connect(API_URL) as ws:
        # 1. Receive the mandatory session.created event

        greeting = json.loads(await ws.recv())
        print("Server:", greeting)

        # 2. Send audio (16-bit little-endian PCM at 16kHz)

        silence = b"\x00\x00" * 800  # 0.05s of silence

        await ws.send(make_event(
            "input_audio_buffer.append",
            audio=silence.hex(),  # Hex-encoded for JSON transport

        ))

        # 3. Commit the buffer to trigger STT

        await ws.send(make_event("input_audio_buffer.commit"))

        # 4. Create a user message

        await ws.send(make_event(
            "conversation.item.create",
            item={
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Hello, how are you?"}]
            }
        ))

        # 5. Request a model response with audio and text

        await ws.send(make_event(
            "response.create",
            response={"modalities": ["text", "audio"]}
        ))

        # 6. Stream events until response is complete

        while True:
            raw = await ws.recv()
            ev = json.loads(raw)
            print("←", ev["type"])
            if ev["type"] == "response.done":
                break

asyncio.run(realtime_client())

Key implementation details:

• Audio format: The server expects 16-bit little-endian PCM at 16kHz sample rate. Binary data must be hex-encoded for JSON transport.
• Event ordering: Always wait for session.created before sending client events.
• Response modalities: The response.create event accepts a modalities array specifying ["text"], ["audio"], or both.

Server Setup and Configuration

To run the WebSocket server locally, use the create_app factory function from websocket_router.py:

from speech_to_speech.api.openai_realtime.websocket_router import create_app
import uvicorn
import threading

# Pipeline units are typically constructed by the main entry point (s2s_pipeline.py)

# Each unit owns a configured STT, LLM, and TTS handler chain

pipeline_units = []  # Populate with configured PipelineUnit instances

stop_event = threading.Event()

app = create_app(pipeline_units, stop_event)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

The pipeline_units list must contain pre-initialized PipelineUnit objects. Refer to scripts/listen_and_play_realtime.py in the repository for a complete example that constructs these units with default STT, LLM, and TTS backends.

Summary

• Isolated sessions: Each WebSocket connection receives a dedicated PipelineUnit from a managed pool, ensuring state isolation and preventing resource leaks between clients.
• Protocol compliance: The RealtimeService class in service.py validates all incoming events against the OpenAI Realtime schema and dispatches them to the appropriate handlers.
• Ordered delivery: The _send_loop_for function guarantees that text and audio responses arrive in the correct sequence by draining queues in priority order and batching audio chunks efficiently.
• Clean teardown: The SESSION_END sentinel mechanism ensures that all pipeline stages flush their buffers and release resources before the unit returns to the idle pool.
• Language agnostic: Custom clients only need to implement the OpenAI Realtime JSON event schema over a standard WebSocket connection; the server handles all audio processing, transcription, and synthesis internally.

Frequently Asked Questions

What audio format does the OpenAI Realtime WebSocket endpoint expect?

The server expects 16-bit little-endian PCM audio at a 16kHz sample rate. When sending via WebSocket JSON events, you must hex-encode the binary PCM data (as shown in the input_audio_buffer.append event). The AudioHandler in the server resamples and processes this audio before enqueueing it for STT transcription.

How does the server handle concurrent client connections?

The server maintains a pool of PipelineUnit objects. When a client connects, the _claim_unit() function assigns an idle unit to that session. Each unit owns its own queues, cancel scopes, and RealtimeService instance. This architecture ensures that state never leaks between sessions, allowing the server to handle multiple concurrent real-time conversations without blocking.

Can I use WebRTC instead of WebSocket for real-time communication?

Yes. The repository implements both transports. While WebSocket uses the WebSocketTransport class for JSON and PCM streaming, the WebRTC implementation (webrtc_session.py) handles SDP exchange and media tracks. Both transports feed into the same PipelineUnit queues, so the choice of transport is transparent to the STT/LLM/TTS processing pipeline.

How do I properly terminate a session to ensure resources are released?

When your client disconnects, the server automatically triggers _release_session(), which flushes all queues and injects a SESSION_END control message (defined in src/speech_to_speech/pipeline/control.py). The _send_loop_for waits for this sentinel to traverse the entire handler chain before marking the session as drained. To force a reset mid-conversation, you can send a session.update event or simply close the WebSocket, and the server will gracefully recycle the unit back to the idle pool.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}