Python SDK

# pip
pip install nolag

# poetry
poetry add nolag

# uv
uv add nolag

import asyncio
from nolag import NoLag, NoLagOptions

async def main():
    client = NoLag("dev-token-alice", NoLagOptions(url="ws://localhost:8080/ws"))

    # A room context prefixes topics with app/room, so this is demo/general/*
    room = client.set_app("demo").set_room("general")
    room.on("messages", lambda data, meta: print(f"Received: {data}"))

    # 'connect' fires after every successful connection, including
    # reconnects. Handlers are plain functions, so schedule the coroutine.
    client.on("connect", lambda: asyncio.ensure_future(room.subscribe("messages")))

    await client.connect()
    await room.emit("messages", {"hello": "world"})

    # Keep running
    await asyncio.sleep(60)

    # Disconnect when done (synchronous)
    client.disconnect()

asyncio.run(main())

from nolag import NoLag, NoLagOptions, QoS

options = NoLagOptions(
    url="ws://localhost:8080/ws",       # Your kraken's WebSocket URL (always set this)
    reconnect=True,                     # Reconnect automatically
    reconnect_interval=5.0,             # Seconds between reconnect attempts
    max_reconnect_attempts=10,          # Attempts before giving up (0 never reconnects)
    heartbeat_interval=30.0,            # Heartbeat in seconds (0 disables)
    qos=QoS.AT_LEAST_ONCE,              # Default QoS level
    load_balance=False,                 # Load-balance every subscription
    load_balance_group=None,            # Group name (default: the actor's id)
    debug=True,                         # Log protocol traffic via the "nolag" logger
)

client = NoLag("dev-token-alice", options)

import asyncio
from nolag import NoLag, NoLagOptions, SubscribeOptions

async def main():
    client = NoLag("dev-token-alice", NoLagOptions(url="ws://localhost:8080/ws"))
    room = client.set_app("demo").set_room("general")

    room.on("messages", lambda data, meta: print(data))

    async def resubscribe():
        await room.subscribe("messages", SubscribeOptions(filters=["vip"]))

    client.on("connect", lambda: asyncio.ensure_future(resubscribe()))
    client.on("disconnect", lambda reason: print(f"Disconnected: {reason}"))
    client.on("reconnect", lambda: print("Reconnected"))

    await client.connect()
    await room.set_presence({"username": "alice", "status": "online"})
    await asyncio.Event().wait()

asyncio.run(main())

room = client.set_app("demo").set_room("general")

# Subscribe, listen and publish within the room
await room.subscribe("messages")
room.on("messages", lambda data, meta: print(data))
await room.emit("messages", {"text": "Hello from general!"})

# The same, with full topic paths
await client.subscribe("demo/general/messages")
client.on("demo/general/messages", lambda data, meta: print(data))
await client.emit("demo/general/messages", {"text": "Hello!"})

# Filters and presence on the room
await room.set_filters("messages", ["lang:en", "lang:fr"])
await room.set_presence({"status": "online"})
members = await room.fetch_presence()

print(room.prefix)  # "demo/general"

await room.unsubscribe("messages")

from nolag import SubscribeOptions, QoS

room = client.set_app("demo").set_room("general")

# Basic subscription and handler
await room.subscribe("messages")
room.on("messages", lambda data, meta: print(data))

# With options (QoS, load balancing, filters)
await room.subscribe("messages", SubscribeOptions(
    qos=QoS.AT_LEAST_ONCE,
    load_balance=True,
    load_balance_group="workers",
    filters=["priority:high", "region:us"],
))

# With an acknowledgement callback: called with None on kraken's
# 'subscribed' frame, or with the error that refused the subscription
def on_subscribed(err):
    if err:
        print("Subscribe failed:", err)
    else:
        print("Subscribed")

await room.subscribe("messages", callback=on_subscribed)

await room.unsubscribe("messages")

from nolag import EmitOptions, QoS

# Publish any MessagePack-encodable data (dict, list, str, bytes, numbers)
await room.emit("messages", {"text": "Hello!"})

# With options
await room.emit("messages", {"online": True}, EmitOptions(
    qos=QoS.AT_LEAST_ONCE,
    retain=True,            # Keep the last message for later subscribers
    filter="priority:high", # Route to subscribers filtering on this value
    echo=False,             # Do not deliver this message back to this connection
))

# With an acknowledgement callback: called on kraken's 'published' frame
await room.emit("messages", {"text": "Hello"}, callback=lambda err: print(err or "sent"))

from nolag import SubscribeOptions, EmitOptions

room = client.set_app("demo").set_room("general")

# Subscribe with initial filters
await room.subscribe("messages", SubscribeOptions(filters=["region:us", "priority:high"]))
room.on("messages", lambda data, meta: print(data, meta.filter))

# Replace, add or remove filters on the existing subscription
await room.set_filters("messages", ["region:eu"])
await room.add_filters("messages", ["priority:critical"])
await room.remove_filters("messages", ["region:eu"])

# Clear all filters (receive everything on the topic again)
await room.set_filters("messages", [])

# Publish with a filter value
await room.emit("messages", {"msg": "US high priority"}, EmitOptions(filter="region:us"))

from nolag import ConnectionStatus

client.on("connect", lambda: print("Connected"))
client.on("disconnect", lambda reason: print(f"Disconnected: {reason}"))
client.on("reconnect", lambda: print("Reconnected"))

# Broker errors arrive as NoLagServerError (see Error handling)
client.on("error", lambda err: print(f"Error: {err}"))

# The app's hydration webhook response, once per subscribe.
# `topic` is the bare topic name, for example "messages".
client.on("hydration", lambda topic, data: print(f"Initial state for {topic}: {data}"))

if client.status == ConnectionStatus.CONNECTED:
    print("Connected as", client.actor_id)

def log_all(topic, data, meta):
    print(f"[{topic}] {data}")

client.on_any(log_all)

client.off_any(log_all)  # remove one handler
client.off_any()         # remove all of them

room = client.set_app("demo").set_room("general")

# Presence events are client-level and carry an ActorPresence
client.on("presence:join", lambda actor: print(f"Joined: {actor.actor_token_id} {actor.presence}"))
client.on("presence:update", lambda actor: print(f"Updated: {actor.presence}"))
client.on("presence:leave", lambda actor: print(f"Left: {actor.actor_token_id}"))

# Set presence in the room (re-sent automatically after reconnects)
await room.set_presence({"status": "online", "username": "alice"})

# The same, naming the room slug explicitly
await client.set_presence({"status": "online"}, room_id="general")

# The room's list from kraken: dicts with actorTokenId, presence and joinedAt
for entry in await room.fetch_presence():
    print(entry["actorTokenId"], entry["presence"])

# Local cache, built from presence events
for actor in client.get_all_presence():
    print(actor.actor_token_id, actor.presence)

# The actor needs a grant for the lobby (see Lobbies)
lobby = client.set_app("demo").set_lobby("online")

# A snapshot keyed by room slug, then actor id
snapshot = await lobby.subscribe()
print("Rooms:", list(snapshot.keys()))

# Events carry lobby_id, room_id, actor_id and data
lobby.on("presence:join", lambda e: print(f"{e.actor_id} joined {e.room_id}"))
lobby.on("presence:leave", lambda e: print(f"{e.actor_id} left {e.room_id}"))

latest = await lobby.fetch_presence()
lobby.unsubscribe()

import asyncio
from nolag import NoLag, NoLagOptions, NoLagServerError, NoLagEncodeError

def on_error(err):
    if isinstance(err, NoLagServerError):
        print(f"{err.error} ({err.code}) on {err.topic}: {err.hint}")
        if err.code == 42940:
            pass  # unknown_topic: no grant covers this address
    elif isinstance(err, NoLagEncodeError):
        print(f"Cannot encode payload for {err.op}: {err}")
    else:
        print(f"Error: {err}")

async def main():
    client = NoLag("dev-token-alice", NoLagOptions(url="ws://localhost:8080/ws"))
    client.on("error", on_error)

    try:
        await client.connect()
    except Exception as e:
        # The broker's reason, for example "access_denied"
        print(f"Connection failed: {e}")
        return

    room = client.set_app("demo").set_room("general")
    try:
        await room.emit("messages", {"data": "value"})
    except Exception as e:
        print(f"Emit failed: {e}")

    client.disconnect()

asyncio.run(main())

from nolag import NoLag, NoLagOptions, SubscribeOptions

# For every subscription on this client
client = NoLag("dev-token-alice", NoLagOptions(
    url="ws://localhost:8080/ws",
    load_balance=True,
    load_balance_group="task-workers",
))

# Or per subscription
room = client.set_app("demo").set_room("general")
await room.subscribe("messages", SubscribeOptions(
    load_balance=True,
    load_balance_group="task-workers",
))
room.on("messages", lambda data, meta: print(f"Processing: {data}"))

from nolag import (
    NoLag,              # The client
    NoLagOptions,       # Connection options
    SubscribeOptions,   # Subscription options
    EmitOptions,        # Publish options
    ConnectionStatus,   # Connection status enum
    ActorType,          # Actor type enum (client.actor_type after connect)
    QoS,                # QoS level enum
    MessageMeta,        # msg_id and filter of a received message
    ActorPresence,      # Presence entry
    LobbyPresenceEvent, # Lobby presence event
    NoLagServerError,   # Structured broker error (error, code, topic, hint, msg_ref)
    NoLagEncodeError,   # Payload could not be MessagePack-encoded
)