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

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), 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/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/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). 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:

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:

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/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/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/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.

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