Python SDK — Realtime

Realtime — asyncio + websockets

Reliable WebSocket (TCP) plus an optional UDP fast lane — 1 TCP + 1 UDP per player

Constructor

Realtime(realtime_websocket_url: str = "wss://realtime.michitai.com")

Initializes the realtime WebSocket client with optional custom server URL. Requires pip install websockets.

get_token

Realtime.get_token(client: Client, player_token: str) -> TokenResponse

Generates a realtime authentication token (player must be in a realtime-enabled room).

connect

await realtime.connect(realtime_token: str) -> bool

Connects to the WebSocket server using the realtime token. Starts the message listener and the 20s heartbeat loop automatically.

send

await realtime.send(
    target: RoomTargetPlayer,
    command: str,
    data: Any = None,
    target_ids: list[int] | None = None,
)

Sends a message to ALL, HOST, OTHERS or SPECIFIC player ids with optional data.

connect_udp

await realtime.connect_udp(udp_port: int | None = None) -> bool

Binds one UDP endpoint alongside the WebSocket (max 1 TCP + 1 UDP per player). Datagrams authenticate with the server-issued connection id. None = the udp_port announced by the server.

send_unreliable

await realtime.send_unreliable(
    target: RoomTargetPlayer,
    command: str,
    data: Any = None,
    target_ids: list[int] | None = None,
)

Sends over the bound UDP channel — unordered, unacknowledged, latest-wins via seq. Ideal for 20-60Hz movement/state streams. Falls back to the WebSocket when UDP isn't bound.

Prediction helpers

PredictionBuffer.next_seq() / .record() / .reconcile(ack, state, apply, dt)
SnapshotBuffer.add(t, state, seq) / .sample(render_time, lerp)

Client-side movement prediction: record each input with its seq, adopt authoritative state on arrival, replay unacked inputs. SnapshotBuffer interpolates remote entities ~100ms in the past. In rooms/realtime/prediction.py.

disconnect

await realtime.disconnect()

Disconnects from the server and cleans up both channels and tasks.

Callbacks

realtime.on_receive = lambda command, data, sender: ...
realtime.on_connected = lambda: ...

on_receive(command, data, sender) fires for each incoming message; on_connected() fires once the socket is open. Both sync and async callbacks are supported.