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:
wss://api.openai.com/v1/responses
wss://api.openai.com/v1/realtime?model=gpt-realtime
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.
How it works:
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:
previous_response_id — a reference to the last response
The server reuses the cached state instead of re-processing the entire conversation history.
Performance benefits:
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:
Key features:
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:
| Metric | HTTP | WebSocket (Responses API) | WebSocket (Realtime API) |
|---|---|---|---|
| Connection setup | Every request (~100–300ms) | Once (~100–300ms) | Once (~100–300ms) |
| 20+ tool call workflow | Baseline | ~40% faster | N/A |
| Voice round-trip latency | N/A (not designed for this) | N/A | <500ms |
| Memory overhead | Low (stateless) | Medium (cached state) | Medium-High (session state) |
| Implementation complexity | Low | Medium | Medium-High |
Let's implement a WebSocket connection to the Responses API for an agentic workflow.
ws for Node.js or websocket-client for Python)
Node.js example:
const WebSocket = require('ws');
// Connect to Responses API WebSocket endpoint
const ws = new WebSocket('wss://api.openai.com/v1/responses', {
headers: {
'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
'OpenAI-Beta': 'responses-api=v1'
}
});
ws.on('open', () => {
console.log('Connected to OpenAI Responses API');
// Send initial request
const initialMessage = {
model: 'gpt-4o',
messages: [
{ role: 'user', content: 'Help me analyze this codebase and suggest improvements.' }
],
tools: [
{
type: 'function',
function: {
name: 'read_file',
description: 'Read contents of a file',
parameters: {
type: 'object',
properties: {
path: { type: 'string', description: 'File path to read' }
},
required: ['path']
}
}
},
{
type: 'function',
function: {
name: 'search_code',
description: 'Search for code patterns',
parameters: {
type: 'object',
properties: {
pattern: { type: 'string', description: 'Regex pattern to search' }
},
required: ['pattern']
}
}
}
]
};
ws.send(JSON.stringify(initialMessage));
});
ws.on('message', (data) => {
const response = JSON.parse(data);
console.log('Received:', response);
// Check if model wants to call tools
if (response.choices[0].finish_reason === 'tool_calls') {
const toolCalls = response.choices[0].message.tool_calls;
// Execute tools (simplified)
const toolResults = toolCalls.map(call => ({
tool_call_id: call.id,
output: executeToolLocally(call.function.name, call.function.arguments)
}));
// Continue the conversation with tool results
const continuation = {
previous_response_id: response.id, // Reference previous response
input: toolResults
};
ws.send(JSON.stringify(continuation));
}
});
ws.on('error', (error) => {
console.error('WebSocket error:', error);
});
ws.on('close', () => {
console.log('Connection closed');
});
function executeToolLocally(name, args) {
// Your tool execution logic
if (name === 'read_file') {
const { path } = JSON.parse(args);
return fs.readFileSync(path, 'utf-8');
}
// ... other tools
}
Python example:
import websocket
import json
import os
def on_message(ws, message):
response = json.loads(message)
print(f"Received: {response}")
if response['choices'][0]['finish_reason'] == 'tool_calls':
tool_calls = response['choices'][0]['message']['tool_calls']
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
})
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")
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()
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()
input (new data—tool results, user messages, etc.). Don't resend the full messages array; the server already has that context.store: false in your initial request.
The Realtime API enables low-latency voice interactions. Here's how to implement it.
JavaScript (browser) example:
// 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();
});
}
python
import websocket
import json
import base64
import pyaudio
RATE = 24000
CHUNK = 4096
FORMAT = pyaudio.paInt16
CHANNELS = 1
audio = pyaudio.PyAudio()
def on_open(ws):
print("Connected to Realtime API")
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
}
}
}))
stream_microphone(ws)
def on_message(ws, message):
data = json.loads(message)
if data['type'] == 'response.audio.delta':
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()
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()
Event types. The Realtime API uses event-driven communication. Common events:
Client → Server:
session.update — configure session parametersinput_audio_buffer.append — send audio chunksconversation.item.create — add text messagesresponse.create — request AI response
Server → Client:
session.created — confirms session setupresponse.audio.delta — audio chunk from AIresponse.audio_transcript.delta — transcription of AI audioinput_audio_buffer.speech_started/stopped — VAD eventserror — error notifications
Voice Activity Detection. Choose between two VAD modes:
server_vad — basic voice activity detection based on audio volume and silence duration.semantic_vad — smarter detection that understands natural s vs. turn completion. Use this for more natural conversations where users might mid-thought.
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
Step 2: Configure headers. Add authentication headers:
Authorization: Bearer [REDACTED]
OpenAI-Beta: responses-api=v1
For the Realtime API, you can also use URL-based authentication:
wss://api.openai.com/v1/realtime?model=gpt-realtime
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" }
}
}
}
}
]
}
Step 5: Monitor responses. Apidog displays:
To test the continuation pattern with previous_response_id:
response.id in the response.
{
"previous_response_id": "resp_abc123",
"input": [
{
"tool_call_id": "call_xyz789",
"output": "{\"temperature\": 72, \"conditions\": \"sunny\"}"
}
]
}
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:
{{OPENAI_API_KEY}}
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.
Why the Responses API WebSocket:
Implementation pattern:
// 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...
Scenario: A phone support bot that handles customer inquiries with natural conversation flow.
Why the Realtime API WebSocket:
// 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))
}
})
Symptoms: The WebSocket connection never opens; immediate close event.
Common causes:
Authorization headerOpenAI-Beta: responses-api=v1
wss:// (not ws://) and the endpoint path
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.
Debugging code:
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)
});
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]
})
Symptoms: Application memory grows over time with a persistent connection.
Solution:
// 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
}
});
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.