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

> Learn to implement WebSocket client connections with the OpenAI Realtime protocol in the huggingface/speech-to-speech library. Stream audio and manage sessions effectively.

- Repository: [Hugging Face/speech-to-speech](https://github.com/huggingface/speech-to-speech)
- Tags: how-to-guide
- Published: 2026-08-09

---

**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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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.

```python
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`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py):

```python
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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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`](https://github.com/huggingface/speech-to-speech/blob/main/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.