# How OpenAI Realtime WebSocket Protocol Works with the Speech-to-Speech Server

> Understand how OpenAI Realtime WebSocket protocol powers speech-to-speech conversations. Learn about its STT LM TTS processing chains and efficient audio streaming for real-time interaction.

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

---

**The OpenAI Realtime WebSocket protocol enables real-time speech-to-speech conversations by streaming PCM16 audio chunks through a persistent WebSocket connection, where each client session is handled by an isolated PipelineUnit running STT→LM→TTS processing chains.**

The Hugging Face `speech-to-speech` repository implements a fully compatible OpenAI Realtime API server that processes audio in real-time using a modular pipeline architecture. This implementation maps the OpenAI Realtime WebSocket protocol events to a local speech-to-speech pipeline, enabling low-latency voice conversations with open-source models.

## Server Bootstrap and Architecture

The server entry point resides in [[`src/speech_to_speech/api/openai_realtime/server.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/server.py)](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/server.py), where the `RealtimeServer` class builds a FastAPI application via `create_app()`.

It spawns a **uvicorn** server in its own thread through the `run()` method, watching a `stop_event` for graceful shutdown. This architecture allows the server to manage a pool of pipeline units while maintaining clean separation between the HTTP/WebSocket layer and the audio processing backend.

## Connection Handling and Session Management

The WebSocket route is defined at `@app.websocket("/v1/realtime")` in [[`websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py)](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py). When a client connects, the server performs three critical operations:

1. **Transport Wrapping**: Creates a `WebSocketTransport` instance (defined in [[`transports.py`](https://github.com/huggingface/speech-to-speech/blob/main/transports.py)](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/transports.py)) around the raw `WebSocket` object.
2. **Unit Claiming**: Calls `_claim_unit(transport)` to atomically select the first idle **`PipelineUnit`** from the pool and attach a fresh `SessionState`.
3. **Session Registration**: Registers a unique session ID with the core `RealtimeService` via `unit.service.register()`, which returns the identifier used in all subsequent events.

If all units are busy, the server returns an `error` event with type `session_limit_reached` and closes the socket (lines 66-75 of the router).

## Processing Client Events

All client JSON messages conform to the OpenAI Realtime schema, including `input_audio_buffer.append`, `input_audio_buffer.commit`, and `response.create`. The server processes these through `await ws.receive_json()` in the main event loop.

The `_dispatch_client_event()` method parses raw dictionaries via `service.parse_client_event()` and routes them accordingly:

- **Audio chunks** flow into `unit.input_queue` for the STT front-end.
- **Response creation** triggers cancellation of ongoing responses, generates new response tokens, and sends acknowledgements.
- **Session updates** and conversation items forward directly to the language model.

Invalid events generate OpenAI-style `error` events through `service.make_error()`, ensuring protocol compliance.

## Streaming Server Responses

Each `PipelineUnit` maintains a dedicated background coroutine `_send_loop_for` started in the FastAPI lifespan. This loop handles bidirectional streaming with three primary responsibilities:

**Text Events**: Pulls from `unit.text_output_queue` (containing `assistant.audio.delta` and `assistant.output_audio_transcript.delta`) and forwards them through `transport.send_events`.

**Audio Output**: Batches PCM bytes from `unit.output_queue` into **6,400-byte chunks** (the OpenAI protocol limit) and transmits them via `transport.send_audio_chunk`.

**Control Flow**: Monitors sentinel events including `SESSION_END`, `PIPELINE_END`, and `response.audio.done` to trigger cleanup, drain pending responses, and release the unit via `_release_session`.

The loop respects **interrupt-response** settings, discarding stale generations when clients begin speaking during assistant output using the unit's `cancel_scope`.

## Session Lifecycle and Pool Health

When clients disconnect or sessions end, `_release_session` executes a careful cleanup sequence:

1. Flushes all queues through `_clean_unit()`.
2. Enqueues a `SESSION_END` control message to allow in-flight handlers to complete.
3. Starts `_release_unit_after_drain`, an asynchronous task waiting for `SESSION_END` propagation through the handler chain.

If draining exceeds `SESSION_END_DRAIN_TIMEOUT_S`, the server logs a warning. After `SESSION_END_QUARANTINE_TIMEOUT_S`, stuck units are quarantined and reported via the `/v1/pool` endpoint, preventing resource leaks.

## WebRTC Alternative Transport

Beyond WebSocket, the server supports the **OpenAI GA Realtime "calls" SDP handshake** at `/v1/realtime/calls` (lines 79-126 of [`websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py)). This optional flow:

- Claims a `PipelineUnit` without initial transport.
- Creates an `aiortc.RTCPeerConnection` wrapped in `WebRTCSession`.
- Connects media tracks to the pipeline via `append_pcm()`.
- Routes events through the `oai-events` data channel using identical JSON schemas.

## Implementation Example

### Minimal Python Client

Connect to the server and stream audio using standard WebSocket libraries:

```python
import asyncio
import base64
import json
import websockets

CHUNK_MS = 20
SAMPLE_RATE = 16000
BYTES_PER_SAMPLE = 2
CHUNK_BYTES = SAMPLE_RATE * BYTES_PER_SAMPLE * CHUNK_MS // 1000  # 640

async def stream_prompt(ws, pcm_bytes):
    for i in range(0, len(pcm_bytes), CHUNK_BYTES):
        chunk = pcm_bytes[i:i + CHUNK_BYTES]
        if len(chunk) < CHUNK_BYTES:
            chunk += b"\x00" * (CHUNK_BYTES - len(chunk))
        await ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(chunk).decode()
        }))
        await asyncio.sleep(CHUNK_MS / 1000)

async def run():
    async with websockets.connect(
        "ws://localhost:8765/v1/realtime",
        extra_headers=[("Authorization", "Bearer $HF_TOKEN")]
    ) as ws:
        first = json.loads(await ws.recv())
        print("Session:", first)
        
        await stream_prompt(ws, prompt_pcm)
        
        while True:
            event = json.loads(await ws.recv())
            print("←", event["type"])
            if event["type"] == "response.done":
                break

asyncio.run(run())

```

### Load Testing with Synthetic Client

Test multiple concurrent sessions using the provided script:

```bash
HF_TOKEN=hf_... python scripts/synthetic_conversation_realtime_client.py \
    --clients 2 --turns 5 --interval 8

```

This demonstrates the full OpenAI Realtime WebSocket protocol implementation including session lifecycle management, error handling, and pool metrics.

## Summary

- The **OpenAI Realtime WebSocket protocol** implementation resides in the Hugging Face `speech-to-speech` repository, providing full API compatibility through FastAPI and uvicorn.
- Each client connection claims an isolated **PipelineUnit** from a managed pool, ensuring STT→LM→TTS processing chains remain independent and stateful.
- Audio flows in **640-byte PCM16 chunks** (20ms at 16kHz) via `input_audio_buffer.append` events, while responses stream back as 6400-byte batches.
- The **send loop** handles both text transcripts and binary audio, monitoring sentinel events like `SESSION_END` for graceful cleanup.
- **WebRTC transport** at `/v1/realtime/calls` offers an alternative to WebSocket using the same event schema over data channels.
- Pool health monitoring via `/v1/pool` tracks unit utilization and quarantines stuck sessions after configurable timeouts.

## Frequently Asked Questions

### What audio format does the OpenAI Realtime WebSocket protocol expect?

The protocol expects **PCM16 audio** at 16kHz sample rate, streamed in 640-byte chunks representing 20 milliseconds of audio. Clients must base64-encode these chunks when sending `input_audio_buffer.append` events, matching the format defined in [[`websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py)](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py).

### How does the server handle multiple concurrent connections?

The server maintains a pool of **PipelineUnit** instances in [[`pipeline_unit.py`](https://github.com/huggingface/speech-to-speech/blob/main/pipeline_unit.py)](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/pipeline_unit.py). When a client connects via `/v1/realtime`, `_claim_unit()` atomically assigns the first idle unit. If all units are busy, the server returns a `session_limit_reached` error and closes the connection, preventing resource exhaustion.

### Can I use WebRTC instead of WebSocket for the OpenAI Realtime protocol?

Yes. The server optionally exposes `/v1/realtime/calls` for **WebRTC transport**, implementing the OpenAI GA Realtime "calls" SDP handshake. This creates an `aiortc.RTCPeerConnection` that routes events through an `oai-events` data channel using identical JSON schemas to the WebSocket implementation, as defined in lines 79-126 of [[`websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py)](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py).

### What happens when a client disconnects unexpectedly?

The server triggers `_release_session`, which flushes queues via `_clean_unit()` and enqueues a `SESSION_END` control message. An asynchronous drain task waits for pipeline completion; if draining exceeds `SESSION_END_DRAIN_TIMEOUT_S`, the unit enters quarantine status visible via the `/v1/pool` health endpoint, ensuring resources eventually return to the pool even after abnormal disconnections.