How to Use OpenAI WebSocket Mode API ?

TL;DR OpenAI offers two WebSocket API modes for different use cases: the Responses API WebSocket mode for agentic workflows with heavy tool calling (up to 40% faster for 20+ tool calls), and the Realtime API for low-latency voice and audio applications. Both use persistent WebSocket connections ins

dev.toFreitag 15:02 · 16.10.17 Min. Lesezeit

OpenAI offers two WebSocket API modes for different use cases: the Responses API WebSocket mode for agentic workflows with heavy tool calling (up to 40% faster for 20+ tool calls), and the Realtime API for low-latency voice and audio applications. Both use persistent WebSocket connections instead of stateless HTTP requests, reducing latency by eliminating repeated connection overhead and enabling event-driven, stateful interactions.

Every HTTP request to OpenAI's API pays a repeated tax: connection setup, authentication, and re-sent conversation state—even when you're continuing the same session. For applications making rapid-fire tool calls or streaming audio, that overhead is pure waste.

OpenAI's WebSocket API eliminates it with a single, persistent, bidirectional connection:

💡 Testing WebSocket connections traditionally required complex debugging tools. Apidog's WebSocket testing interface lets you establish connections, send events, and monitor responses in real time—essential for validating OpenAI WebSocket integrations before production deployment.

This guide covers both WebSocket modes: the Responses API for tool-heavy agent workflows, and the Realtime API for audio streaming. You'll learn when to use each, how to implement them, and how to test them effectively.

The OpenAI WebSocket API is an alternative transport to HTTP for interacting with OpenAI's models. Instead of opening a new connection per API call, you establish one long-lived connection that stays open for the duration of your session.

WebSocket connections start with an HTTP upgrade request, then switch to the WebSocket protocol. For OpenAI, connect to:

The wss:// scheme indicates a secure WebSocket connection (analogous to HTTPS for HTTP).

OpenAI provides two distinct WebSocket modes, each optimized for different use cases.

Use this mode for agentic workflows that make many sequential tool calls: coding assistants, orchestration systems, and autonomous agents that repeatedly call tools to complete complex tasks.

On an active connection, the service maintains one previous-response state in a connection-local in-memory cache (the most recent response). To continue a turn, you send only:

The server reuses the cached state instead of re-processing the entire conversation history.

For workflows with 20+ tool calls, OpenAI reports up to 40% faster end-to-end execution compared to HTTP, thanks to:

Compatibility: WebSocket mode works with both Zero Data Retention (ZDR) and store=false options, making it suitable for privacy-sensitive applications.

Use this mode for low-latency, streaming audio in voice-driven applications—speech-to-speech interactions where the model responds to audio input with audio output and handles interruptions naturally.

The Realtime API creates a stateful, event-driven session over WebSocket. You stream audio chunks in, and it streams back both transcriptions and generated audio. The connection supports:

Choosing between WebSocket and HTTP depends on your application's characteristics. Here's when each protocol excels.

High tool call volume: If your workflow makes 10+ sequential tool calls, the persistent connection eliminates repeated setup overhead. Each HTTP request requires:

WebSocket does this once, then reuses the connection.

Latency-sensitive applications: For real-time voice or chat where every millisecond counts, the persistent connection and streaming capabilities significantly reduce perceived latency.

Server-initiated updates: The server can push data to clients without polling. For long-running agent tasks, the server sends progress updates as events occur.

Simple request-response: For one-off API calls or workflows with 1–2 tool calls, HTTP is simpler to implement and debug, and infrastructure (load balancers, proxies) handles it well.

Stateless operations: If you don't need session state between requests, HTTP's stateless nature is an advantage—no connection management required.

Infrastructure constraints: Some environments (serverless functions, certain proxies) don't support long-lived WebSocket connections. HTTP works universally.

Based on OpenAI's documentation and community testing:

Let's implement a WebSocket connection to the Responses API for an agentic workflow.

import websocket import json import os def on_message ( ws, message ): response = json. loads ( message ) print ( f " Received: { response } " ) # Handle tool calls if response [ ' choices ' ][ 0 ][ ' finish_reason ' ] == ' tool_calls ': tool_calls = response [ ' choices ' ][ 0 ][ ' message ' ][ ' tool_calls ' ] # Execute tools tool_results = [] for call in tool_calls: result = execute_tool ( call [ ' function ' ][ ' name ' ], json. loads ( call [ ' function ' ][ ' arguments ' ])) tool_results. append ({ ' tool_call_id ': call [ ' id ' ], ' output ': result }) # Send continuation with only new input + previous_response_id continuation = { ' previous_response_id ': response [ ' id ' ], ' input ': tool_results } ws. send ( json. dumps ( continuation )) def on_error ( ws, error ): print ( f " Error: { error } " ) def on_close ( ws, close_status_code, close_msg ): print ( " Connection closed " ) def on_open ( ws ): print ( " Connected to OpenAI Responses API " ) # Send initial request initial_message = { ' model ': ' gpt-4o ', ' messages ': [ { ' role ': ' user ', ' content ': ' Analyze this codebase and suggest improvements. ' } ], ' tools ': [ { ' type ': ' function ', ' function ': { ' name ': ' read_file ', ' description ': ' Read file contents ', ' parameters ': { ' type ': ' object ', ' properties ': { ' path ': { ' type ': ' string ' } }, ' required ': [ ' path ' ] } } } ] } ws. send ( json. dumps ( initial_message )) def execute_tool ( name, args ): if name == ' read_file ': with open ( args [ ' path ' ], ' r ' ) as f: return f. read () #... other tools # Create WebSocket connection ws = websocket. WebSocketApp ( " wss://api.openai.com/v1/responses ", header = { " Authorization ": f " Bearer { os. environ [ ' OPENAI_API_KEY ' ] } ", " OpenAI-Beta ": " responses-api=v1 " }, on_open = on_open, on_message = on_message, on_error = on_error, on_close = on_close ) ws. run_forever () Enter fullscreen mode Exit fullscreen mode Key Implementation Details State management: The critical difference from HTTP is using previous_response_id in continuations. This tells the API to reuse cached state from the last response.

Input-only continuations: When continuing a turn, send only previous_response_id (references the cached response) and input (new data—tool results, user messages, etc.). Don't resend the full messages array; the server already has that context. Zero Data Retention: To use ZDR with WebSocket mode, include store: false in your initial request.

How to Use Realtime API WebSocket Mode The Realtime API enables low-latency voice interactions. Here's how to implement it.

// Connect to Realtime API const ws = new WebSocket ( `wss://api.openai.com/v1/realtime?model=gpt-realtime`, [ ' realtime ', ' openai-insecure-api-key. ' + process. env. OPENAI_API_KEY ] ); ws. addEventListener ( ' open ', () => { console. log ( ' Connected to Realtime API ' ); // Configure session ws. send ( JSON. stringify ({ type: ' session.update ', session: { modalities: [ ' text ', ' audio ' ], voice: ' alloy ', input_audio_format: ' pcm16 ', output_audio_format: ' pcm16 ', turn_detection: { type: ' server_vad ', // or 'semantic_vad' for smarter detection threshold: 0.5, prefix_padding_ms: 300, silence_duration_ms: 500 } } })); }); ws. addEventListener ( ' message ', ( event ) => { const message = JSON. parse ( event. data ); switch ( message. type ) { case ' session.created ': console. log ( ' Session created: ', message. session ); break; case ' conversation.item.created ': console. log ( ' New item: ', message. item ); break; case ' response.audio.delta ': // Received audio chunk - play it const audioChunk = base64ToArrayBuffer ( message. delta ); playAudioChunk ( audioChunk ); break; case ' response.audio_transcript.delta ': // Received transcript chunk console. log ( ' Transcript: ', message. delta ); break; case ' input_audio_buffer.speech_started ': console. log ( ' User started speaking ' ); break; case ' input_audio_buffer.speech_stopped ': console. log ( ' User stopped speaking ' ); break; case ' error ': console. error ( ' API error: ', message. error ); break; } }); // Send audio from microphone async function streamMicrophoneToAPI () { const stream = await navigator. mediaDevices. getUserMedia ({ audio: true }); const audioContext = new AudioContext ({ sampleRate: 24000 }); const source = audioContext. createMediaStreamSource ( stream ); const processor = audioContext. createScriptProcessor ( 4096, 1, 1 ); processor. onaudioprocess = ( e ) => { const inputData = e. inputBuffer. getChannelData ( 0 ); // Convert Float32 to Int16 PCM const pcmData = new Int16Array ( inputData. length ); for ( let i = 0; i < inputData. length; i ++ ) { pcmData [ i ] = Math. max ( - 32768, Math. min ( 32767, inputData [ i ] * 32768 )); } // Send to API ws. send ( JSON. stringify ({ type: ' input_audio_buffer.append ', audio: arrayBufferToBase64 ( pcmData. buffer ) })); }; source. connect ( processor ); processor. connect ( audioContext. destination ); } // Send text input function sendTextMessage ( text ) { ws. send ( JSON. stringify ({ type: ' conversation.item.create ', item: { type: ' message ', role: ' user ', content: [ { type: ' input_text ', text: text } ] } })); // Request response generation ws. send ( JSON. stringify ({ type: ' response.create ' })); } function playAudioChunk ( arrayBuffer ) { const audioContext = new AudioContext ({ sampleRate: 24000 }); audioContext. decodeAudioData ( arrayBuffer, ( buffer ) => { const source = audioContext. createBufferSource (); source. buffer = buffer; source. connect ( audioContext. destination ); source. start (); }); } Enter fullscreen mode Exit fullscreen mode Python example:

import websocket import json import base64 import pyaudio # Audio configuration RATE = 24000 CHUNK = 4096 FORMAT = pyaudio. paInt16 CHANNELS = 1 audio = pyaudio. PyAudio () def on_open ( ws ): print ( " Connected to Realtime API " ) # Configure session ws. send ( json. dumps ({ ' type ': ' session.update ', ' session ': { ' modalities ': [ ' text ', ' audio ' ], ' voice ': ' alloy ', ' input_audio_format ': ' pcm16 ', ' output_audio_format ': ' pcm16 ', ' turn_detection ': { ' type ': ' server_vad ', ' threshold ': 0.5, ' silence_duration_ms ': 500 } } })) # Start streaming microphone stream_microphone ( ws ) def on_message ( ws, message ): data = json. loads ( message ) if data [ ' type ' ] == ' response.audio.delta ': # Decode and play audio audio_chunk = base64. b64decode ( data [ ' delta ' ]) play_audio ( audio_chunk ) elif data [ ' type ' ] == ' response.audio_transcript.delta ': print ( f " Transcript: { data [ ' delta ' ] } ", end = '', flush = True ) elif data [ ' type ' ] == ' input_audio_buffer.speech_started ': print ( " \n [User speaking...] " ) elif data [ ' type ' ] == ' error ': print ( f " Error: { data [ ' error ' ] } " ) def stream_microphone ( ws ): stream = audio. open ( format = FORMAT, channels = CHANNELS, rate = RATE, input = True, frames_per_buffer = CHUNK ) def audio_thread (): while True: audio_data = stream. read ( CHUNK ) ws. send ( json. dumps ({ ' type ': ' input_audio_buffer.append ', ' audio ': base64. b64encode ( audio_data ). decode ( ' utf-8 ' ) })) import threading threading. Thread ( target = audio_thread, daemon = True ). start () def play_audio ( audio_chunk ): stream = audio. open ( format = FORMAT, channels = CHANNELS, rate = RATE, output = True ) stream. write ( audio_chunk ) stream. stop_stream () stream. close () # Create WebSocket connection ws = websocket.

WebSocketApp ( f " wss://api.openai.com/v1/realtime?model=gpt-realtime ", header = { " Authorization ": f " Bearer { os. environ [ ' OPENAI_API_KEY ' ] } " }, on_open = on_open, on_message = on_message ) ws. run_forever () Enter fullscreen mode Exit fullscreen mode Key Implementation Details Event types. The Realtime API uses event-driven communication. Common events:

Voice Activity Detection. Choose between two VAD modes:

Testing WebSocket APIs differs from HTTP testing—you need to maintain a connection, send events, and monitor bidirectional message flow. Apidog provides specialized WebSocket testing capabilities.

Step 1: Create a WebSocket request. In Apidog, create a new request and select "WebSocket" as the protocol. Enter your connection URL:

wss://api.openai.com/v1/responses Enter fullscreen mode Exit fullscreen mode Step 2: Configure headers. Add authentication headers:

Authorization: Bearer [REDACTED] OpenAI-Beta: responses-api=v1 Enter fullscreen mode Exit fullscreen mode For the Realtime API, you can also use URL-based authentication:

wss://api.openai.com/v1/realtime?model=gpt-realtime Enter fullscreen mode Exit fullscreen mode with the API key in the subprotocol header.

Step 3: Establish the connection. Click "Connect." Apidog shows:

Step 4: Send events. Use Apidog's message composer to send JSON events. For the Responses API:

{ "model": "gpt-4o", "messages": [ { "role": "user", "content": "What's the weather in San Francisco?" } ], "tools": [ { "type": "function", "function": { "name": "get_weather", "description": "Get current weather", "parameters": { "type": "object", "properties": { "location": { "type": "string" } } } } } ] } Enter fullscreen mode Exit fullscreen mode Step 5: Monitor responses. Apidog displays:

To test the continuation pattern with previous_response_id:

{ "previous_response_id": "resp_abc123", "input": [ { "tool_call_id": "call_xyz789", "output": "{ \" temperature \": 72, \" conditions \": \" sunny \" }" } ] } Enter fullscreen mode Exit fullscreen mode Observe the reduced latency compared to resending full context. Testing the Realtime API For the Realtime API, Apidog lets you:

This is particularly useful for debugging why your voice assistant might be cutting off users or not detecting speech properly.

Store API keys securely using Apidog's environment variables:

This lets you switch between development and production keys without editing requests.

Here are practical scenarios where OpenAI's WebSocket modes excel.

Scenario: A coding assistant that analyzes codebases, identifies issues, and makes improvements autonomously.

// Initial task ws. send ({ messages: [{ role: ' user ', content: ' Audit security vulnerabilities ' }], tools: }) // First response: model calls read_file ws. on ( ' message ', ( resp1 ) => { ws. send ({ previous_response_id: resp1. id, input: [ tool_result_1 ] }) }) // Second response: model calls search_code ws. on ( ' message ', ( resp2 ) => { ws. send ({ previous_response_id: resp2. id, input: [ tool_result_2 ] }) }) // Continue for 20+ iterations... Enter fullscreen mode Exit fullscreen mode Use Case 2: Voice Customer Service Bot Scenario: A phone support bot that handles customer inquiries with natural conversation flow.

// Stream phone audio to API phoneSystem. on ( ' audio ', ( chunk ) => { ws. send ({ type: ' input_audio_buffer.append ', audio: base64Encode ( chunk ) }) }) // Play AI responses immediately ws. on ( ' message ', ( event ) => { if ( event. type === ' response.audio.delta ' ) { phoneSystem. playAudio ( base64Decode ( event. delta )) } }) Enter fullscreen mode Exit fullscreen mode Troubleshooting Common Issues Connection Fails to Establish Symptoms: The WebSocket connection never opens; immediate close event.

Solution: Use Apidog to test the connection with detailed error messages. The request inspector shows exactly which headers are sent, making it easy to spot missing or incorrect API keys.

ws. on ( ' error ', ( error ) => { console. error ( ' Connection error: ', error ); }); ws. on ( ' close ', ( code, reason ) => { console. log ( `Closed with code ${ code }: ${ reason } ` ); // Common codes: // 1006: Abnormal closure (often auth issues) // 1008: Policy violation (invalid headers) }); Enter fullscreen mode Exit fullscreen mode High Latency Despite WebSocket Symptoms: The WebSocket connection works but isn't faster than HTTP.

Solution: Verify you're sending only new input in continuations:

// WRONG - sends full context every time ws. send ({ messages: [... allPreviousMessages, newMessage ], tools: }) // RIGHT - references cached state ws. send ({ previous_response_id: lastResponse. id, input: [ newMessage ] }) Enter fullscreen mode Exit fullscreen mode Memory Leaks in Long-Running Connections Symptoms: Application memory grows over time with a persistent connection.

// Clean up event listeners on reconnection function cleanupAndReconnect ( ws ) { ws. removeAllListeners (); ws. close (); const newWs = createConnection (); return newWs; } // Release audio buffers after playing function playAndRelease ( audioBuffer ) { const source = audioContext. createBufferSource (); source. buffer = audioBuffer; source. connect ( audioContext. destination ); source. start (); source. onended = () => { source. disconnect (); // Buffer will be garbage collected }; } // Limit message history const messageHistory = []; const MAX_HISTORY = 100; ws. on ( ' message ', ( data ) => { messageHistory. push ( data ); if ( messageHistory. length > MAX_HISTORY ) { messageHistory. shift (); // Remove oldest } }); Enter fullscreen mode Exit fullscreen mode Conclusion OpenAI's WebSocket API modes unlock new possibilities for AI applications. The Responses API WebSocket mode delivers up to 40% faster execution for agentic workflows with heavy tool calling, making it ideal for autonomous coding assistants and orchestration systems. The Realtime API provides sub-500ms latency for voice applications, enabling natural, interrupt-capable conversations.

Choosing the right mode depends on your use case:

The persistent, event-driven nature of WebSocket connections requires different testing approaches than HTTP. Test OpenAI's WebSocket APIs with Apidog's real-time WebSocket client—import your API key, establish connections, send events, and monitor responses with detailed logging. Try it free to validate your integrations before production deployment.

Quelle: dev.to (dev.to). Original öffnen