Connect a Realtime session

WS/v1/realtime?model={model}

Token360 provides an OpenAI-compatible Realtime WebSocket endpoint for low-latency text and audio sessions.

Use this endpoint from a trusted server environment. Do not expose API keys in browser or mobile client code.

Connection


URL	`wss://api.token360.ai/v1/realtime?model=gpt-realtime-2.1`
Authentication	`Authorization: Bearer <TOKEN360_API_KEY>` during the WebSocket handshake
Model	`gpt-realtime-2.1`
Protocol	OpenAI-compatible GA Realtime events over WebSocket

Do not send the legacy OpenAI-Beta: realtime=v1 header.

Text session flow

Open the WebSocket and wait for session.created.
Send session.update with output_modalities: ["text"].
Wait for session.updated before sending the first item.
Send a conversation.item.create event containing input_text.
Send response.create.
Read response.output_text.delta events until response.done.

Waiting for session.updated is important. The default session configuration may use audio output, so a text-only client must explicitly select text before creating its first response.

Audio sessions

For speech-to-speech use cases, configure audio input and output through session.update, stream Base64 audio chunks with input_audio_buffer.append, then commit the input and create a response. Keep the audio format consistent for the whole session.

Do not apply the text-only output_modalities: ["text"] setting to an audio session.

Common client events


`session.update`	Configure output modality, instructions, tools, audio, and other session options
`conversation.item.create`	Add text or other conversation content
`input_audio_buffer.append`	Append a Base64-encoded audio chunk
`input_audio_buffer.commit`	Commit buffered audio as user input
`response.create`	Request a model response

Common server events


`session.created`	The connection is ready for session configuration
`session.updated`	The requested session configuration was accepted
`response.output_text.delta`	Incremental text output
`response.output_audio.delta`	Incremental Base64 audio output
`response.done`	Terminal response event containing status and usage
`error`	Request or session error

Errors and reconnects

Treat a non-upgraded WebSocket handshake as an authentication, authorization, model availability, or request validation failure and inspect the returned status and message.
For an error event, inspect error.code, error.type, error.param, and error.message when present.
Reconnect transiently closed sessions with bounded exponential backoff and jitter.
A new WebSocket creates a new session. Re-send the required session configuration and any application state that must be restored.
Realtime sessions are not permanent. Plan to reconnect before 60 minutes rather than relying on an indefinitely open connection.

Usage lookup

The terminal response.done event contains response usage. You can also reconcile a completed response through:

Shell

1curl "https://api.token360.ai/v1/billing/requests/RESPONSE_ID" \
2  -H "Authorization: Bearer $TOKEN360_API_KEY"

Request

Install the WebSocket client:

Shell

1pip install "websockets>=14"

Set TOKEN360_API_KEY in the environment, then run:

Python

1import asyncio
2import json
3import os
4
5import websockets
6
7
8URL = (
9    "wss://api.token360.ai/v1/realtime"
10    "?model=gpt-realtime-2.1"
11)
12
13
14async def receive_event(ws):
15    event = json.loads(await ws.recv())
16    if event.get("type") == "error":
17        raise RuntimeError(event.get("error", {}).get("message", "Realtime error"))
18    return event
19
20
21async def wait_for(ws, event_type):
22    while True:
23        event = await receive_event(ws)
24        if event.get("type") == event_type:
25            return event
26
27
28async def main():
29    headers = {
30        "Authorization": f"Bearer {os.environ['TOKEN360_API_KEY']}"
31    }
32
33    async with websockets.connect(URL, additional_headers=headers) as ws:
34        await wait_for(ws, "session.created")
35
36        await ws.send(json.dumps({
37            "type": "session.update",
38            "session": {
39                "type": "realtime",
40                "output_modalities": ["text"]
41            }
42        }))
43        await wait_for(ws, "session.updated")
44
45        await ws.send(json.dumps({
46            "type": "conversation.item.create",
47            "item": {
48                "type": "message",
49                "role": "user",
50                "content": [{
51                    "type": "input_text",
52                    "text": "Reply with exactly: Realtime is ready"
53                }]
54            }
55        }))
56        await ws.send(json.dumps({"type": "response.create"}))
57
58        output = []
59        while True:
60            event = await receive_event(ws)
61            if event.get("type") == "response.output_text.delta":
62                output.append(event.get("delta", ""))
63            if event.get("type") == "response.done":
64                response = event.get("response", {})
65                print("".join(output))
66                print(json.dumps(response.get("usage", {}), indent=2))
67                break
68
69
70asyncio.run(main())

The same event sequence applies to other server-side WebSocket libraries.

Was this page helpful?