Python 88.3%
TypeScript 7.6%
Shell 4.1%
1#!/usr/bin/env python32"""STATUS: LIVE_VERIFIED 2026-09-18 — wss://api.openai.com/v1/realtime?model=gpt-realtime-mini, text-only turn, 14 server events, 3.6 s.34Observed order: session.created, session.updated, conversation.item.added, conversation.item.done, response.created,5response.output_item.added, conversation.item.added, response.content_part.added, response.output_text.delta,6response.output_text.done, response.content_part.done, conversation.item.done, response.output_item.done, response.done.7Costs a fraction of a cent (13 input / 3 output text tokens). Requires `websockets` (.venv has 17.1).8Run: .venv/bin/python examples/openai/realtime/websocket_text_only.py9"""10from __future__ import annotations1112import asyncio13import json14import os15import sys16from pathlib import Path1718sys.path.insert(0, str(Path(__file__).resolve().parents[3]))19from scripts.live import log_request # noqa: E402 (loads .env)2021MODEL = os.environ.get("ATLAS_REALTIME_MODEL", "gpt-realtime-mini")22URL = f"wss://api.openai.com/v1/realtime?model={MODEL}" # GA: no OpenAI-Beta header232425async def main() -> None:26 import websockets2728 seen: list[str] = []29 async with websockets.connect(URL, additional_headers={"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"}, max_size=2**24) as ws:30 first = json.loads(await ws.recv()); seen.append(first["type"]) # session.created31 await ws.send(json.dumps({"type": "session.update", "session": {"type": "realtime", "output_modalities": ["text"], "instructions": "Reply with OK only."}}))32 await ws.send(json.dumps({"type": "conversation.item.create", "item": {"type": "message", "role": "user",33 "content": [{"type": "input_text", "text": "Reply with OK."}]}}))34 await ws.send(json.dumps({"type": "response.create", "response": {"output_modalities": ["text"], "max_output_tokens": 16}}))35 while True:36 ev = json.loads(await asyncio.wait_for(ws.recv(), timeout=30)); seen.append(ev["type"])37 if ev["type"] == "response.output_text.delta":38 print("delta:", ev["delta"])39 if ev["type"] == "error":40 print("error:", ev["error"]); break41 if ev["type"] == "response.done":42 print("status:", ev["response"]["status"], "usage:", ev["response"]["usage"]); break43 log_request("openai", "WS", "/v1/realtime", 101, 0.0001, f"example websocket_text_only.py {len(seen)} server events")44 print("server events:", " → ".join(seen))454647if __name__ == "__main__":48 asyncio.run(main())49