openai-realtime-api-intro.py

import json
import os
from pprint import pprint
from typing import Any

import websocket
from dotenv import load_dotenv
from websocket import WebSocket

load_dotenv()

OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]


def on_message(ws: WebSocket, message: Any) -> None:
    payload = json.loads(str(message))
    pprint(payload)


def on_error(ws: WebSocket, error: Any) -> None:
    print("ERROR: ", error)


def on_open(ws: WebSocket) -> None:
    print("Connected to server.")
    ws.send(
        json.dumps(
            {
                "type": "response.create",
                "response": {
                    "modalities": ["text"],
                    "instructions": "Please assist the user.",
                },
            }
        )
    )


ws_endpoint = (
    "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"
)

ws = websocket.WebSocketApp(
    ws_endpoint,
    header={"Authorization": f"Bearer {OPENAI_API_KEY}", "OpenAI-Beta": "realtime=v1"},
    on_message=on_message,
    on_error=on_error,
    on_open=on_open,
)

ws.run_forever()