sodp

Python asyncio client for the State-Oriented Data Protocol (SODP) — a WebSocket-based protocol for continuous state synchronization.

Instead of polling or request/response, SODP streams every change as a minimal delta to all connected subscribers. One mutation to a 100-field object sends exactly the changed fields.

→ Protocol spec & server

Install

pip install sodp-client

Requires Python 3.11+ and a running asyncio event loop.

Quick start

import asyncio
from sodp import SodpClient

async def main():
    client = SodpClient("ws://localhost:7777")
    await client.ready

    # Subscribe
    def on_player(value, meta):
        print(f"player: {value}  version={meta.version}")

    unsub = client.watch("game.player", on_player)

    # Mutate
    await client.set("game.player", {"name": "Alice", "health": 100})
    await client.patch("game.player", {"health": 80})   # only health changes

    await asyncio.sleep(1)

    unsub()        # remove this callback
    client.close() # close the connection

asyncio.run(main())

Authentication

# Static token
client = SodpClient("wss://sodp.example.com", token="eyJhbG...")

# Dynamic token provider — called on every connect/reconnect
async def get_token() -> str:
    async with aiohttp.ClientSession() as s:
        return await (await s.get("/api/sodp-token")).text()

client = SodpClient("wss://sodp.example.com", token_provider=get_token)

# Sync provider is also accepted
client = SodpClient(url, token_provider=lambda: os.environ["SODP_TOKEN"])

API reference

SodpClient(url, *, ...)

client = SodpClient(
    url,                        # WebSocket URL, e.g. "ws://localhost:7777"
    token=None,                 # static JWT string
    token_provider=None,        # callable → str | Awaitable[str]; supersedes token
    reconnect=True,             # auto-reconnect on disconnect
    reconnect_delay=1.0,        # base reconnect delay in seconds (doubles per attempt)
    max_reconnect_delay=30.0,   # maximum reconnect delay in seconds
    on_connect=None,            # called each time the connection is established
    on_disconnect=None,         # called each time the connection drops
)

The client connects immediately in the background. Use await client.ready (or await client) to wait for the first successful authentication before sending commands.

await client.ready

Awaitable that resolves once the client is connected and authenticated. You can also await client directly:

await client.ready  # explicit
await client        # same thing

client.watch(key, callback) → unsub

Subscribe to a state key. callback(value, meta) fires on every update and immediately with the cached value if the key is already known.

• value — current state (any JSON-compatible type), or None if the key has no value yet

• meta.version — monotonically increasing version number (int)

• meta.initialized — False when the key has never been written to the server

• meta.source — origin of this callback invocation:

  • "cache" — fired synchronously from watch() with an already-cached value
  • "init" — the server's STATE_INIT baseline (initial load or post-reconnect)
  • "delta" — an incremental mutation (DELTA frame)

  Use meta.source — not meta.initialized — to distinguish the initial baseline from subsequent changes. initialized only tells you whether the key has ever been written on the server.

callback may be a plain function or an async function.

Returns an unsubscribe callable. Multiple watch() calls for the same key share a single server subscription.

def on_player(value, meta):
    if not meta.initialized:
        return
    print(value["name"], value["health"])

unsub = client.watch("game.player", on_player)

# Async callback also works:
async def on_score(value, meta):
    await db.update_score(value)

unsub2 = client.watch("game.score", on_score)

client.state(key) → StateRef

Returns a key-scoped handle for cleaner per-key code:

player = client.state("game.player")

unsub = player.watch(lambda v, m: print(v))

await player.set({"name": "Alice", "health": 100, "position": {"x": 0, "y": 0}})
await player.patch({"health": 80})           # only health changes
await player.set_in("/position/x", 5)        # atomic nested field update
await player.delete()                        # remove the key entirely
await player.presence("/alice", {"line": 1}) # session-lifetime path

current = player.get()                       # cached snapshot
player.unwatch()                             # cancel subscription

await client.call(method, args) → data

Invoke a built-in server method:

await client.call("state.set", {"state": "game.score", "value": {"value": 0}})

Convenience methods

await client.set("game.score", {"value": 42})
await client.patch("game.player", {"health": 80})
await client.presence("collab.cursors", "/alice", {"name": "Alice", "line": 3})

client.unwatch(key)

Cancel the server subscription and clear all local state for a key.

client.get(key) → Any

Synchronously read the cached value without subscribing. Returns None if the key is not being watched or has no value.

client.is_watching(key) → bool

Returns True if this client has an active subscription for key.

client.close()

Gracefully close the connection and stop reconnecting.

Presence

Presence binds a nested path to the session lifetime. The server automatically removes it and notifies all watchers when the client disconnects for any reason — no ghost cursors or stale "online" flags:

# Bind cursor to this session — auto-removed if the process crashes or disconnects
await client.presence("collab.cursors", "/alice", {"name": "Alice", "line": 1})

# Or via StateRef:
cursors = client.state("collab.cursors")
await cursors.presence("/alice", {"name": "Alice", "line": 1})

Auto-reconnect & RESUME

The client reconnects with exponential backoff (1 s → 2 s → 4 s → … → 30 s). After reconnecting:

• Keys with a known version send RESUME — the server replays missed deltas, then resumes live streaming
• Keys never seen yet send WATCH — you receive the current snapshot

No data is lost during short disconnections as long as the server's delta log is not full (1 000 deltas per key).

StateRef API summary

ref = client.state("my.key")

ref.watch(callback)          # subscribe; returns unsub callable
ref.get()                    # cached value
ref.is_watching()            # True if subscribed
ref.unwatch()                # cancel subscription + clear local state
await ref.set(value)         # replace full value
await ref.patch(partial)     # deep-merge partial
await ref.set_in(path, val)  # set nested field by JSON Pointer
await ref.delete()           # remove key from server
await ref.presence(path, v)  # session-lifetime path binding

FastAPI example

from contextlib import asynccontextmanager
from fastapi import FastAPI
from sodp import SodpClient

client: SodpClient

@asynccontextmanager
async def lifespan(app: FastAPI):
    global client
    client = SodpClient("ws://sodp-server:7777", token=os.environ["SODP_TOKEN"])
    await client.ready
    yield
    client.close()

app = FastAPI(lifespan=lifespan)

@app.post("/score/{value}")
async def set_score(value: int):
    await client.set("game.score", {"value": value})
    return {"ok": True}