# How to Implement OpenAI Realtime WebSocket Protocol for Voice Agents

> Implement OpenAI Realtime WebSocket protocol for voice agents using RealtimeService WebSocketTransport and PipelineUnit for seamless VAD STT LLM TTS streaming.

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

---

**Build a production-ready OpenAI Realtime WebSocket protocol for voice agents by leveraging `RealtimeService` for event parsing and pipeline orchestration, `WebSocketTransport` for JSON event handling, and `PipelineUnit` for VAD → STT → LLM → TTS streaming.**

The Hugging Face **speech-to-speech** repository provides a complete, open-source implementation of the OpenAI Realtime API that supports both WebSocket and WebRTC transports. This guide walks through the architecture, event lifecycle, and practical implementation details needed to build custom voice agents using the Realtime WebSocket protocol.

---

## Core Architecture of the Realtime WebSocket Protocol

The implementation centers on three cooperating components that handle the OpenAI Realtime WebSocket protocol end-to-end.

### RealtimeService: Protocol Engine

The **`RealtimeService`** class in [`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) serves as the protocol's brain. It parses incoming client events, maintains per-connection state via **`ConnState`**, and dispatches work to the processing pipeline.

Key responsibilities include:

- **Event parsing** – Maps raw JSON to typed Pydantic models using `_EVENT_TYPE_TO_MODEL` (lines 73-81)
- **State management** – Tracks `session_id`, `RuntimeConfig`, and pipeline unit assignments per connection
- **Pipeline dispatch** – Routes audio chunks and generation requests to `PipelineUnit` queues
- **Outbound event generation** – Converts pipeline outputs (`PipelineEvent`) back to protocol events

### WebSocket Router and Transport

The **`WebSocketTransport`** and router in [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) handle the network layer. The `create_app` factory (lines 59-67) mounts the `/v1/realtime` WebSocket endpoint that:

1. Claims a **`PipelineUnit`** from the thread pool via `_claim_unit`
2. Starts the **send loop** (`_send_loop_for`, lines 122-138) to drain output queues
3. Dispatches client events through `_dispatch_client_event` to the appropriate handlers
4. Releases the unit on disconnect

### PipelineUnit: Audio Processing Pipeline

Each client connection owns one **`PipelineUnit`** that runs four stages on background threads:

| Stage | Input | Output |
|-------|-------|--------|
| **VAD** (Voice Activity Detection) | Raw PCM chunks | `speech_started` / `speech_stopped` events |
| **STT** (Speech-to-Text) | Speech segments | Transcript text |
| **LLM** | Conversation context | Assistant text, tool calls |
| **TTS** (Text-to-Speech) | Assistant text or tool results | PCM audio bytes |

The unit's `input_queue` receives `AudioChunkItem` objects from the service, while its output queues feed the send loop that streams events back to clients.

---

## OpenAI Realtime WebSocket Event Lifecycle

Understanding the event flow is essential for implementing compliant clients. The protocol uses bidirectional JSON messages with specific event types.

### Inbound Client Events (Client → Server)

| Event | Handler Location | Action |
|-------|----------------|--------|
| `input_audio_buffer.append` | `AudioHandler.handle_audio_append` | Splits base64 PCM into 512-sample chunks (see `CHUNK_SIZE_BYTES`, service.py lines 65-69) and enqueues to `unit.input_queue` |
| `input_audio_buffer.commit` | `AudioHandler.handle_audio_commit` | Signals end-of-audio to VAD, triggering `speech_stopped` |
| `session.update` | `SessionHandler.handle_session_update` | Merges JSON into `RuntimeConfig` for dynamic threshold adjustment |
| `conversation.item.create` | `ConversationHandler.handle_conversation_item_create` | Injects messages into chat context without LLM invocation |
| `response.create` | `ResponseHandler.handle_response_create` | Starts new response, sends `response.created`, triggers LLM via `GenerateResponseRequest` |
| `response.cancel` | `ResponseHandler.handle_response_cancel` | Aborts active response, flushes queues, sends `response.done` with status *cancelled* |

### Outbound Server Events (Server → Client)

These events are generated by `RealtimeService.dispatch_pipeline_event` (service.py lines 47-54) and forwarded by the send loop:

- `input_audio_buffer.speech_started` / `speech_stopped` – VAD state changes
- `conversation.item.input_audio_transcription.delta` / `completed` – Incremental and final transcripts
- `response.audio.delta` – Base64-encoded PCM chunks (20ms batches)
- `response.audio_transcript.done` – Final assistant text
- `response.function_call_arguments.done` – Tool call payloads for client execution

All state is scoped to `ConnState.session_id`, ensuring pipeline unit reclamation cannot leak data between clients.

---

## Implementing a Minimal OpenAI Realtime WebSocket Client

This Python client demonstrates the complete protocol flow: connecting, streaming audio, committing the buffer, requesting a response, and processing events.

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

WS_URL = "ws://localhost:8765/v1/realtime"
CHUNK_MS = 20
SAMPLE_RATE = 16000
BYTES_PER_SAMPLE = 2
CHUNK_BYTES = SAMPLE_RATE * BYTES_PER_SAMPLE * CHUNK_MS // 1000  # 640 bytes

async def stream_pcm(ws, pcm_bytes):
    """Stream PCM audio using OpenAI Realtime input_audio_buffer.append events."""
    for i in range(0, len(pcm_bytes), CHUNK_BYTES):
        chunk = pcm_bytes[i:i + CHUNK_BYTES]
        # Pad final chunk to maintain timing

        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 main():
    async with websockets.connect(WS_URL) as ws:
        # Receive session.created handshake

        print(">>", await ws.recv())

        # Load 16kHz PCM16 mono audio

        with open("prompt.wav", "rb") as f:
            pcm = f.read()

        # Stream audio chunks

        await stream_pcm(ws, pcm)
        
        # Commit buffer to trigger processing

        await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))

        # Request assistant response

        await ws.send(json.dumps({
            "type": "response.create",
            "response": {
                "model": "openai/gpt-4o-mini",
                "instructions": "You are a helpful assistant."
            }
        }))

        # Process events until response completes

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

asyncio.run(main())

```

This mirrors the implementation in [`scripts/synthetic_conversation_realtime_client.py`](https://github.com/huggingface/speech-to-speech/blob/main/scripts/synthetic_conversation_realtime_client.py), which serves as the reference production client.

---

## Testing with the Built-In Synthetic Client

The repository includes a full-featured load-testing client that handles VAD-aware silence detection and realistic turn-taking.

First, start the server:

```bash
uv run speech-to-speech serve \
    --stt parakeet-tdt \
    --llm_backend transformers \
    --tts kokoro \
    --model_name "Qwen/Qwen3-4B-Instruct-2507" \
    --enable_live_transcription

```

Then launch synthetic conversations:

```bash
python scripts/synthetic_conversation_realtime_client.py \
    --clients 1 \
    --turns 10 \
    --interval 5

```

The synthetic client (lines 88-122) uses macOS `say` for prompt generation, implements proper event state machines, and validates protocol compliance—making it ideal for testing custom implementations.

---

## Handling Function Calls and Tool Use

The OpenAI Realtime WebSocket protocol supports tool calling through specific event types. When the LLM emits a function call, the server sends:

```json
{
  "type": "response.function_call_arguments.done",
  "name": "get_weather",
  "arguments": "{\"location\": \"San Francisco\"}",
  "call_id": "call_abc123"
}

```

Execute the tool client-side, then return results via `conversation.item.create`:

```json
{
  "type": "conversation.item.create",
  "item": {
    "type": "function_call_output",
    "function_call_id": "call_abc123",
    "output": "{\"temperature\": 72, \"unit\": \"F\"}"
  }
}

```

The `ConversationHandler` in [`handlers/conversation.py`](https://github.com/huggingface/speech-to-speech/blob/main/handlers/conversation.py) queues these outputs during active responses via `flush_deferred_items`, ensuring proper ordering with assistant generation.

---

## Extending the OpenAI Realtime WebSocket Protocol

### Adding Custom Client Events

1. **Define the Pydantic model** in `openai.types.realtime` matching the JSON schema
2. **Register in `_EVENT_TYPE_TO_MODEL`** (service.py lines 73-81)
3. **Implement handler** in `RealtimeService`
4. **Add dispatch case** in [`websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py)'s `_dispatch_client_event`

### Customizing Audio Parameters

- Override `AudioHandler.handle_audio_append` for alternative chunking strategies
- Modify `PIPELINE_SAMPLE_RATE` (default 16kHz, service.py line 65)
- Adjust `CHUNK_SIZE_BYTES` for latency/throughput tradeoffs

### Swapping Pipeline Stages

- Provide custom `LLMHandler` implementation
- Update `RealtimeService._pipeline_dispatch` to route `AssistantTextEvent` to new handlers
- Maintain queue type compatibility with [`queue_types.py`](https://github.com/huggingface/speech-to-speech/blob/main/queue_types.py) definitions

---

## Key Source Files Reference

| File | Role | Critical Sections |
|------|------|-------------------|
| [`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) | Protocol engine | Connection lifecycle (`register`/`unregister`), event parsing (`parse_client_event`), pipeline dispatch (`_dispatch_pipeline_event`) |
| [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) | WebSocket endpoint | `create_app`, `_claim_unit`, `_dispatch_client_event`, `_send_loop_for` |
| `src/speech_to_speech/api/openai_realtime/handlers/*.py` | Domain handlers | `AudioHandler.handle_audio_append`, `ResponseHandler.handle_response_create`, `ConversationHandler.handle_conversation_item_create` |
| [`src/speech_to_speech/pipeline/speculative_turns.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/pipeline/speculative_turns.py) | Turn tracking | `SpeculativeTurnTracker.observe`, `is_latest_after_reopen_grace` |
| [`scripts/synthetic_conversation_realtime_client.py`](https://github.com/huggingface/speech-to-speech/blob/main/scripts/synthetic_conversation_realtime_client.py) | Reference client | Lines 88-122 for event state machine implementation |

---

## Summary

- **Initialize** with `create_app(pool, stop_event)` to expose `/v1/realtime` WebSocket and `/v1/realtime/calls` WebRTC endpoints
- **Stream audio** as base64 PCM at 16kHz using `input_audio_buffer.append`, then `commit` to trigger VAD processing
- **Request responses** with `response.create`; handle incremental output via `response.audio.delta` and `response.audio_transcript.delta`
- **Cancel gracefully** using `response.cancel`, which flushes queues and emits `response.done` with *cancelled* status
- **Extend protocol** by adding Pydantic models to `_EVENT_TYPE_TO_MODEL` and registering handlers in the router dispatch block

---

## Frequently Asked Questions

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

The protocol expects **16kHz PCM16 mono audio** encoded as base64 strings in `input_audio_buffer.append` events. The `CHUNK_SIZE_BYTES` constant in [`service.py`](https://github.com/huggingface/speech-to-speech/blob/main/service.py) (lines 65-69) defines 512-sample chunks (640 bytes at 16kHz/16-bit). Clients should stream chunks every 20ms to maintain real-time latency.

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

Each WebSocket connection claims one **`PipelineUnit`** from a thread pool via `_claim_unit` in [`websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py). The unit encapsulates all processing state, and `ConnState.session_id` isolates data between clients. When a client disconnects, the unit releases back to the pool for reuse.

### Can I use the OpenAI Realtime protocol with custom STT or TTS models?

Yes. The `PipelineUnit` accepts pluggable handlers through the `pipeline` package. Implement custom `STTHandler` or `TTSHandler` classes, then inject them during `RealtimeService` initialization. The `RealtimeService._pipeline_dispatch` method routes events to your handlers based on type annotations.

### How do I implement voice activity detection (VAD) tuning?

Send `session.update` events to modify `RuntimeConfig` thresholds dynamically. The `SessionHandler.handle_session_update` merges JSON payloads into per-session configuration, affecting subsequent VAD decisions without interrupting active processing.