openai_realtime_api_websocket_text_message_exchange.py

1import asyncio
2import os
3import json
4import websockets
5
6async def hello_realtime():
7    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"
8    headers = {
9        "Authorization": "Bearer " + os.getenv("OPENAI_API_KEY"),
10        "OpenAI-Beta": "realtime=v1",
11    }
12
13    async with websockets.connect(url, extra_headers=headers) as ws:
14        # Receive the initial session.created event
15        res = await ws.recv()
16        print(json.loads(res))
17
18        # Send a text message to the assistant
19        event = {
20            "type": "conversation.item.create",
21            "item": {
22                "type": "message",
23                "role": "user",
24                "content": [
25                    {
26                        "type": "input_text",
27                        "text": "Hello, how are you?"
28                    }
29                ]
30            }
31        }
32        await ws.send(json.dumps(event))
33
34        # Request a response
35        await ws.send(json.dumps({"type": "response.create"}))
36
37        # Listen for the response events
38        async for message in ws:
39            event = json.loads(message)
40            if event["type"] == "response.audio_transcript.delta":
41                print(event["delta"], end="", flush=True)
42            elif event["type"] == "response.done":
43                print("\nResponse finished.")
44                break
45
46if __name__ == "__main__":
47    asyncio.run(hello_realtime())