Voice chat with OpenAI GPT-Live
Talks to the model through your microphone and speakers, with live transcripts in the terminal. It needs no server, only an API key.
OPENAI_API_KEY=sk-... uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/openai_live.py
It connects with WARP, which saves round trips when the call starts. --no-warp
connects the usual way, and -v logs how long each step took after the offer.
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = ["wrtc>=0.0.1", "sounddevice", "httpx"]
# ///
#
# Copyright 2022-2026 Ilya (Marshal) <https://github.com/MarshalX>.
#
# Dedicated to the public domain under CC0, see the LICENSE file of the examples.
#
"""Talk to OpenAI GPT-Live from the terminal: your microphone goes to the model, its voice to your speakers.
No server needed: the script posts its SDP offer to the OpenAI API with your key, as the guide
https://developers.openai.com/api/docs/guides/voice-webrtc?api=live describes for a backend.
export OPENAI_API_KEY=sk-...
uv run openai_live.py
uv run openai_live.py --voice gleam --instructions "You are a pirate. Keep answers short."
uv run openai_live.py --list-devices
uv installs the dependencies declared above. Without a download, straight from GitHub:
uv run https://raw.githubusercontent.com/MarshalX/python-webrtc/main/examples/openai_live.py
Or with pip: pip install wrtc sounddevice httpx, then python openai_live.py.
It connects with WARP, which saves round trips when the call starts: --no-warp turns it off, -v logs the timings.
Press Ctrl+C to hang up. Without headphones the microphone is muted while the assistant speaks, so it doesn't
hear itself; with headphones pass --barge-in to be able to interrupt it.
"""
from __future__ import annotations
import argparse
import asyncio
import contextlib
import json
import os
import signal
import sys
import threading
import time
from array import array
from typing import ClassVar, TypedDict
import httpx
import sounddevice
from typing_extensions import NotRequired
import webrtc
API_URL = 'https://api.openai.com/v1/live/sessions'
SAMPLE_RATE = 48000
FRAME = SAMPLE_RATE // 100 # 10 ms of samples, as WebRTC sends them
VOICE_LEVEL = 0.02 # the peak level above which audio counts as speech
ECHO_TAIL = 0.6 # seconds the microphone stays muted after the assistant stops
MAX_PLAYBACK = 0.5 # seconds of assistant audio buffered before the oldest is dropped
SILENT_MIC_WARNING = 10 # seconds of silence before the microphone is suspected
# the trials of WARP, whose DTLS 1.3 is on by default
WARP_TRIALS = {'WebRTC-Sctp-Snap': 'Enabled', 'WebRTC-IceHandshakeDtls': 'Enabled'}
WARP_CHANNEL_ID = 4 # the event channel's, sent to the API as dcid
class Session(TypedDict, total=False):
"""The session of a reply of the API."""
id: str
class Transport(TypedDict):
"""The transport of a reply of the API: the answer."""
sdp: str
class SessionReply(TypedDict):
"""The reply of the API to a new session."""
session: NotRequired[Session]
transport: Transport
class Usage(TypedDict, total=False):
"""The usage of a session."""
seconds: float
class ContextWindow(TypedDict, total=False):
"""How full the context of the model is."""
usage_ratio: object
class ApiError(TypedDict, total=False):
"""An error of the API."""
message: str
class ErrorReply(TypedDict, total=False):
"""The reply of the API to a request that failed."""
error: ApiError
class Message(TypedDict, total=False):
"""An event of the API on the event channel: the fields used here."""
type: str
delta: str
reason: str
usage: Usage
error: ApiError
context_window: ContextWindow
class Console:
"""Human-readable output: timestamped status lines and live transcripts that share the terminal."""
COLORS: ClassVar[dict[str, str]] = {
'dim': '2',
'red': '31',
'green': '32',
'yellow': '33',
'blue': '34',
'magenta': '35',
'cyan': '36',
}
def __init__(self, *, verbose: bool) -> None:
self.verbose = verbose
self.color = sys.stdout.isatty() and os.environ.get('NO_COLOR', '') == ''
self.speaker: str | None = None # who the open transcript line belongs to
def paint(self, text: str, color: str) -> str:
"""The text in a color, if the terminal shows colors."""
return f'\033[{self.COLORS[color]}m{text}\033[0m' if self.color else text
def _end_transcript(self) -> None:
if self.speaker is not None:
print(flush=True)
self.speaker = None
def log(self, icon: str, text: str, color: str | None = None) -> None:
"""Prints a timestamped status line."""
self._end_transcript()
line = f'{self.paint(time.strftime("%H:%M:%S"), "dim")} {icon} {text}'
print(self.paint(line, color) if color is not None else line, flush=True)
def info(self, text: str) -> None:
"""Prints a status line."""
self.log('•', text)
def ok(self, text: str) -> None:
"""Prints a line of a success."""
self.log('✓', text, 'green')
def warn(self, text: str) -> None:
"""Prints a line of a problem."""
self.log('!', text, 'yellow')
def error(self, text: str) -> None:
"""Prints a line of a failure."""
self.log('✗', text, 'red')
def debug(self, text: str) -> None:
"""Prints a line in verbose mode only."""
if self.verbose:
self.log('·', text, 'dim')
def transcript(self, speaker: str, delta: str) -> None:
"""Appends a fragment to the speaker's line; fragments have no end marker, so a new speaker starts a line."""
if speaker != self.speaker:
self._end_transcript()
label, color = ('You', 'cyan') if speaker == 'user' else ('Assistant', 'magenta')
print(self.paint(f'\n{label}: ', color), end='')
self.speaker = speaker
print(delta, end='', flush=True)
def peak(samples: bytes | bytearray) -> float:
"""The peak level of 16-bit samples, from 0 to 1."""
values = array('h', samples)
return max(*values, -min(values)) / 32768 if len(values) > 0 else 0
class Microphone:
"""Captures 10 ms chunks of 16-bit mono audio from a device into an asyncio queue."""
def __init__(self, device: str | int | None, loop: asyncio.AbstractEventLoop) -> None:
self.queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=50)
self._loop = loop
self._stream = sounddevice.RawInputStream(
samplerate=SAMPLE_RATE, channels=1, dtype='int16', blocksize=FRAME, device=device, callback=self._on_audio
)
self.name = sounddevice.query_devices(self._stream.device)['name']
def _on_audio(self, data: memoryview, *_: object) -> None:
self._loop.call_soon_threadsafe(self._put, bytes(data))
def _put(self, chunk: bytes) -> None:
if self.queue.full():
self.queue.get_nowait()
self.queue.put_nowait(chunk)
def start(self) -> None:
"""Starts capturing."""
self._stream.start()
def close(self) -> None:
"""Stops capturing and releases the device."""
self._stream.stop()
self._stream.close()
class Speakers:
"""Plays 16-bit audio on a device from a small buffer, opened on the first chunk to match its format."""
def __init__(self, device: str | int | None) -> None:
self.device = device
self.name = sounddevice.query_devices(device, 'output')['name']
self._stream: sounddevice.RawOutputStream | None = None
self._buffer = bytearray()
self._lock = threading.Lock()
self._limit = 0
def play(self, samples: bytes | bytearray, sample_rate: int, channels: int) -> None:
"""Queues samples, opening the device on the first ones."""
if self._stream is None:
self._limit = int(sample_rate * MAX_PLAYBACK) * channels * 2
self._stream = sounddevice.RawOutputStream(
samplerate=sample_rate, channels=channels, dtype='int16', device=self.device, callback=self._on_need
)
self._stream.start()
with self._lock:
self._buffer += samples
if len(self._buffer) > self._limit:
del self._buffer[: len(self._buffer) - self._limit]
def _on_need(self, out: memoryview, *_: object) -> None:
size = len(out)
with self._lock:
chunk = self._buffer[:size]
del self._buffer[:size]
out[: len(chunk)] = chunk
out[len(chunk) :] = bytes(size - len(chunk))
def close(self) -> None:
"""Releases the device."""
if self._stream is not None:
self._stream.stop()
self._stream.close()
class LiveCall:
"""A call with the model: the microphone and the speakers, the connection and its event channel."""
def __init__(self, args: argparse.Namespace, console: Console) -> None:
self.args = args
self.console = console
self.pc: webrtc.RTCPeerConnection | None = None
self.events: webrtc.RTCDataChannel | None = None
self.microphone: Microphone | None = None
self.speakers: Speakers | None = None
self.tasks: list[asyncio.Future[None]] = []
self.session_closed = asyncio.Event()
self.assistant_spoke_at = 0.0 # when the assistant's audio was last above the speech level
self.offered_at = 0.0
async def run(self, hang_up: asyncio.Event) -> None:
"""Connects, then talks until hung up."""
console, args = self.console, self.args
loop = asyncio.get_running_loop()
self.microphone = Microphone(args.input_device, loop)
self.speakers = Speakers(args.output_device)
console.info(f'Microphone: {self.microphone.name}')
console.info(f'Speakers: {self.speakers.name}')
if not args.barge_in:
console.info('Echo guard on: the mic is muted while the assistant speaks (--barge-in turns it off)')
self.pc = pc = webrtc.RTCPeerConnection()
pc.on('connectionstatechange', self._on_connection_state)
pc.on('track', self._on_track)
generator = webrtc.MediaStreamTrackGenerator('audio')
pc.add_track(generator)
# created before the offer, so the offer negotiates it
init = None if args.no_warp else webrtc.RTCDataChannelInit(negotiated=True, id=WARP_CHANNEL_ID)
self.events = pc.create_data_channel('oai-events', init)
self.events.on('open', self._on_events_open)
self.events.on('message', self._on_message)
offer = await pc.create_offer()
self.offered_at = time.monotonic()
await pc.set_local_description(offer)
sdp = await self._gathered(pc)
console.info(f'Creating a {args.model} session...')
answer = await self._create_session(sdp)
await pc.set_remote_description(webrtc.RTCSessionDescriptionInit('answer', answer))
self.microphone.start()
writer = generator.writable.get_writer()
self.tasks.append(asyncio.ensure_future(self._send_microphone(writer, self.microphone)))
await hang_up.wait()
await self.close()
async def _gathered(self, pc: webrtc.RTCPeerConnection) -> str:
"""Waits for the local ICE candidates, which go in the offer since there is no trickling, returns the offer."""
done = asyncio.Event()
def on_gathering_state(_event: webrtc.Event) -> None:
if pc.ice_gathering_state == 'complete':
done.set()
pc.on('icegatheringstatechange', on_gathering_state)
if pc.ice_gathering_state != 'complete':
try:
await asyncio.wait_for(done.wait(), 5)
except asyncio.TimeoutError:
self.console.warn('ICE gathering is slow, sending the candidates found so far')
if pc.local_description is None:
msg = 'No local description'
raise CallError(msg)
return pc.local_description.sdp
async def _create_session(self, sdp: str) -> str:
session: dict[str, object] = {'model': self.args.model}
if self.args.instructions:
session['instructions'] = self.args.instructions
if self.args.voice:
session['audio'] = {'output': {'voice': self.args.voice}}
transport: dict[str, object] = {'type': 'webrtc', 'sdp': sdp}
if not self.args.no_warp:
transport['dcid'] = WARP_CHANNEL_ID
try:
async with httpx.AsyncClient(timeout=30) as client:
response = await client.post(
API_URL,
headers={'Authorization': f'Bearer {self.args.api_key}'},
json={'session': session, 'transport': transport},
)
except httpx.HTTPError as e:
msg = f'Could not reach the OpenAI API: {e or type(e).__name__}'
raise CallError(msg) from None
if response.is_error:
raise CallError(_describe_http_error(response))
reply: SessionReply = response.json()
self.console.ok(f'Session created: {reply.get("session", {}).get("id", "?")}')
return reply['transport']['sdp']
def _elapsed(self) -> str:
return f'{(time.monotonic() - self.offered_at) * 1000:.0f} ms after the offer'
def _on_events_open(self, _event: webrtc.Event) -> None:
self.console.ok('Event channel open')
self.console.debug(f'Event channel open {self._elapsed()}')
def _on_connection_state(self, _event: webrtc.Event) -> None:
if self.pc is None:
msg = 'No connection'
raise CallError(msg)
state = self.pc.connection_state
messages = {
'connecting': ('info', 'Connecting audio...'),
'connected': ('ok', 'Audio connected'),
'disconnected': ('warn', 'Connection interrupted, waiting for it to recover...'),
'failed': ('error', 'Connection failed (a firewall may block UDP)'),
}
if state in messages:
level, text = messages[state]
getattr(self.console, level)(text)
if state == 'connected':
self.console.debug(f'Connected {self._elapsed()}')
def _on_track(self, event: webrtc.RTCTrackEvent) -> None:
if self.speakers is None:
msg = 'No speakers'
raise CallError(msg)
self.console.debug(f"Receiving the assistant's {event.track.kind} track")
self.tasks.append(asyncio.ensure_future(self._play(event.track, self.speakers)))
async def _send_microphone(self, writer: webrtc.WritableStreamDefaultWriter, microphone: Microphone) -> None:
"""Sends the microphone to the model, silence instead while the echo guard holds it."""
console, loop = self.console, asyncio.get_running_loop()
silence = bytes(FRAME * 2)
timestamp, heard, started = 0, False, loop.time()
while True:
chunk = await microphone.queue.get()
if not heard:
if peak(chunk) > VOICE_LEVEL:
heard = True
console.ok('Microphone is picking up sound')
elif loop.time() - started > SILENT_MIC_WARNING:
heard = True
console.warn(
f'The microphone has been silent for {SILENT_MIC_WARNING} s: '
'check its permission and input level'
)
guarded = not self.args.barge_in and time.monotonic() - self.assistant_spoke_at < ECHO_TAIL
data = webrtc.AudioData(
webrtc.AudioDataInit(
format='s16',
sample_rate=SAMPLE_RATE,
number_of_frames=FRAME,
number_of_channels=1,
timestamp=timestamp,
data=silence if guarded else chunk,
)
)
await writer.write(data)
timestamp += 10_000
async def _play(self, track: webrtc.MediaStreamTrack, speakers: Speakers) -> None:
"""Plays the assistant's audio."""
heard = False
async for data in webrtc.MediaStreamTrackProcessor(
webrtc.MediaStreamTrackProcessorInit(track, max_buffer_size=50)
).readable:
if not isinstance(data, webrtc.AudioData):
msg = f'expected audio data, not {data!r}'
raise TypeError(msg)
with data:
options = webrtc.AudioDataCopyToOptions(plane_index=0, format='s16')
samples = bytearray(data.allocation_size(options))
data.copy_to(samples, options)
rate, channels = int(data.sample_rate), data.number_of_channels
if peak(samples) > VOICE_LEVEL:
self.assistant_spoke_at = time.monotonic()
if not heard:
heard = True
self.console.ok(f"Receiving the assistant's voice ({rate} Hz, {channels} ch)")
speakers.play(samples, rate, channels)
def _on_message(self, event: webrtc.MessageEvent) -> None:
console = self.console
message = parse_message(event.data)
if message is None:
console.debug(f'Not JSON: {event.data!r}')
return
kind = message.get('type')
if kind == 'session.input_transcript.delta':
console.transcript('user', message.get('delta', ''))
elif kind == 'session.output_transcript.delta':
console.transcript('assistant', message.get('delta', ''))
elif kind == 'session.started':
# the API starts it: session.start would break WARP
console.ok('Session started, say something! (Ctrl+C to hang up)')
console.debug(f'Session started {self._elapsed()}')
elif kind == 'session.closed':
usage = message.get('usage', {})
reason = message.get('reason', '')
console.info(
f'Session closed{f" ({reason})" if reason != "" else ""}, {usage.get("seconds", "?")} s billed'
)
self.session_closed.set()
elif kind == 'error':
error = message.get('error', {})
text = error.get('message')
console.error(f'API error: {text if text is not None and text != "" else error}')
elif kind == 'session.usage.updated':
usage = message.get('usage', {})
ratio = message.get('context_window', {}).get('usage_ratio')
context = f', context {ratio:.0%} full' if isinstance(ratio, (int, float)) else ''
console.debug(f'Usage: {usage.get("seconds", "?")} s{context}')
else:
console.debug(f'Event {kind}')
async def close(self) -> None:
"""Ends the session gracefully, then releases the audio devices."""
console = self.console
if self.events is not None and self.events.ready_state == 'open':
console.info('Hanging up...')
self.events.send(json.dumps({'type': 'session.close'}))
try:
await asyncio.wait_for(self.session_closed.wait(), 5)
except asyncio.TimeoutError:
console.warn("The session didn't confirm closing")
for task in self.tasks:
task.cancel()
await asyncio.gather(*self.tasks, return_exceptions=True)
if self.pc is not None:
self.pc.close()
if self.microphone is not None:
self.microphone.close()
if self.speakers is not None:
self.speakers.close()
console.ok('Bye!')
def parse_message(data: str | bytes | webrtc.Blob) -> Message | None:
"""An event of the API from the event channel, None if it isn't JSON."""
if isinstance(data, webrtc.Blob):
msg = 'the binary type of the event channel is bytes, not blobs'
raise TypeError(msg)
try:
message: Message = json.loads(data)
except ValueError:
return None
return message
class CallError(Exception):
"""The call couldn't be set up."""
def _describe_http_error(response: httpx.Response) -> str:
try:
reply: ErrorReply = response.json()
except ValueError:
reply = {}
message = reply.get('error', {}).get('message', '')
hints = {
401: 'the API key is invalid',
403: 'the key has no access to this model',
404: 'unknown endpoint or model',
429: 'rate limit or quota exceeded',
}
summary = hints.get(response.status_code, response.reason_phrase)
return f'OpenAI API returned {response.status_code}, {summary}' + (f': {message}' if message != '' else '')
def parse_args() -> argparse.Namespace:
"""The options of the command line."""
parser = argparse.ArgumentParser(
description='Voice chat with OpenAI GPT-Live over WebRTC.',
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
parser.add_argument('--api-key', default=os.environ.get('OPENAI_API_KEY'), help='default: $OPENAI_API_KEY')
parser.add_argument('--model', default='gpt-live-1', help='the GPT-Live model')
parser.add_argument('--instructions', help='the system prompt')
parser.add_argument('--voice', help='quartz, ripple, vesper, willow, stone, gleam, meridian, bossa, tempo, ...')
parser.add_argument('--input-device', help='microphone name or index (see --list-devices), default: the system one')
parser.add_argument('--output-device', help='speakers name or index (see --list-devices), default: the system one')
parser.add_argument('--no-warp', action='store_true', help='connect without WARP, with the usual handshakes')
parser.add_argument('--barge-in', action='store_true', help='keep the mic open while the assistant speaks')
parser.add_argument('--list-devices', action='store_true', help='list audio devices and exit')
parser.add_argument('-v', '--verbose', action='store_true', help='log every event from the API')
args = parser.parse_args()
for name in ('input_device', 'output_device'):
value: str | None = getattr(args, name)
if value is not None and value.isdigit():
setattr(args, name, int(value))
return args
async def main() -> None:
"""Runs a call until Ctrl+C."""
args = parse_args()
if args.list_devices:
print(sounddevice.query_devices())
return
console = Console(verbose=args.verbose)
if not args.api_key:
console.error('No API key: set OPENAI_API_KEY or pass --api-key')
sys.exit(1)
if not args.no_warp:
webrtc.field_trials.update(WARP_TRIALS)
hang_up = asyncio.Event()
# on Windows, Ctrl+C raises KeyboardInterrupt instead
with contextlib.suppress(NotImplementedError):
asyncio.get_running_loop().add_signal_handler(signal.SIGINT, hang_up.set)
call = LiveCall(args, console)
try:
await call.run(hang_up)
except (CallError, sounddevice.PortAudioError) as e:
console.error(str(e))
await call.close()
sys.exit(1)
if __name__ == '__main__':
with contextlib.suppress(KeyboardInterrupt):
asyncio.run(main())