GPT Realtime 2.1

# pip install "websockets>=14"
import asyncio
import base64
import json
import os

import websockets

URL = "wss://api.modellix.ai/api/v1/live/openai/gpt-realtime-2.1"


async def main():
    headers = {"Authorization": f"Bearer {os.environ['MODELLIX_API_KEY']}"}
    async with websockets.connect(URL, additional_headers=headers, max_size=16 * 1024 * 1024) as ws:
        # 1. The server speaks first. Wait for session.created.
        created = json.loads(await ws.recv())
        if created["type"] != "session.created":
            raise RuntimeError(created)

        # 2. Configure the session.
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "type": "realtime",
                "instructions": "Answer briefly.",
                "output_modalities": ["audio"],
                "audio": {
                    "output": {
                        "format": {"type": "audio/pcm", "rate": 24000},
                        "voice": "marin",
                    }
                },
            },
        }))

        # 3. Add a user message and request a response.
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Introduce yourself in one sentence."}],
            },
        }))
        await ws.send(json.dumps({"type": "response.create"}))

        # 4. Read events until the response is done.
        with open("reply.pcm", "wb") as output:
            async for raw in ws:
                event = json.loads(raw)
                if event["type"] == "response.output_audio.delta":
                    output.write(base64.b64decode(event["delta"]))
                elif event["type"] == "response.output_audio_transcript.delta":
                    print(event["delta"], end="", flush=True)
                elif event["type"] == "error":
                    raise RuntimeError(event["error"])
                elif event["type"] == "response.done":
                    print("\nUsage:", event["response"]["usage"])
                    return


asyncio.run(main())