How to Implement OpenAI Realtime WebSocket Protocol for Voice Agents

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

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, 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:

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:

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:

{
  "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:

{
  "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 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'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 definitions

---

Key Source Files Reference

File	Role	Critical Sections
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	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	Turn tracking	SpeculativeTurnTracker.observe, is_latest_after_reopen_grace
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 (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. 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.

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