pythonadvanced

OpenAI Realtime API WebSocket

Connect to the OpenAI Realtime API via WebSocket for low-latency voice and text streaming.

pythonPress ⌘/Ctrl + Shift + C to copy

import asyncio
import json
import websockets

async def realtime_chat(api_key: str, user_message: str) -> None:
    url = 'wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview'
    headers = {'Authorization': f'Bearer {api_key}', 'OpenAI-Beta': 'realtime=v1'}

    async with websockets.connect(url, additional_headers=headers) as ws:
        # Send user turn
        await ws.send(json.dumps({'type': 'conversation.item.create', 'item': {'type': 'message', 'role': 'user', 'content': [{'type': 'input_text', 'text': user_message}]}}))
        await ws.send(json.dumps({'type': 'response.create'}))

        # Stream response
        while True:
            event = json.loads(await ws.recv())
            if event['type'] == 'response.text.delta':
                print(event['delta'], end='', flush=True)
            elif event['type'] == 'response.done':
                print()
                break

asyncio.run(realtime_chat('YOUR_API_KEY', 'Hello!'))