Python SDK
The official NoLag SDK for Python. Full async/await support with type hints for Python 3.10+.
Installation
# pip
pip install nolag
# poetry
poetry add nolag
# uv
uv add nolagQuick Start
import asyncio
from nolag import NoLag
# The slug returned when you created the app (slugs carry a random suffix)
APP_SLUG = "chat-a3f9"
async def main():
# Create client with your actor token
client = NoLag("your-actor-token")
# Connect to NoLag
await client.connect()
# Scope to an app and a room. The room must already exist.
room = client.set_app(APP_SLUG).set_room("general")
# Subscribe to a topic in the room
await room.subscribe("messages")
# Listen for messages on that topic
def on_message(data, meta):
print(f"Received: {data}")
room.on("messages", on_message)
# Publish a message
await room.emit("messages", {"hello": "world"})
# Keep running
await asyncio.sleep(60)
# Disconnect when done (synchronous)
client.disconnect()
asyncio.run(main())Topics live inside rooms. The only hierarchy is app/room/topic: the room context above addresses chat-a3f9/general/messages. Topic names are single tokens ([a-zA-Z0-9_:-], no /). Rooms never exist implicitly: create them via the REST API or the portal first, or subscribing returns unknown_topic (42940). App slugs always get a random 4-hex suffix, so read slug from the create response and pass that to set_app().
Configuration
from nolag import NoLag, NoLagOptions, QoS
options = NoLagOptions(
url="wss://broker.nolag.app/ws", # Custom broker URL
reconnect=True, # Auto-reconnect on disconnect
reconnect_interval=5.0, # Seconds between reconnect attempts
max_reconnect_attempts=10, # Max reconnect attempts (default 10; 0 disables reconnection)
heartbeat_interval=30.0, # Heartbeat interval in seconds (0 to disable)
qos=QoS.AT_LEAST_ONCE, # Default QoS level
load_balance=False, # Enable load balancing
load_balance_group=None, # Load balance group name
debug=True, # Enable debug logging
)
client = NoLag("your-actor-token", options)There is no unlimited mode: reconnection stops after max_reconnect_attempts
consecutive failures, and 0 never reconnects.
Fluent API
The fluent API lets you scope operations to a specific app and room, so you don't need to repeat the full topic path every time.
# Scope topics to an app and room. APP_SLUG is the suffixed slug from the create response
room = client.set_app(APP_SLUG).set_room("general")
# Subscribe, listen, and emit within the room
await room.subscribe("messages")
room.on("messages", lambda data, meta: print(data))
await room.emit("messages", {"text": "Hello from general!"})
# This is equivalent to using full topic paths (with APP_SLUG = "chat-a3f9"):
# await client.subscribe("chat-a3f9/general/messages")
# client.on("chat-a3f9/general/messages", handler)
# await client.emit("chat-a3f9/general/messages", data)
# Room also supports filter management
await room.set_filters("messages", ["lang:en", "lang:fr"])
await room.add_filters("messages", ["lang:de"])
await room.remove_filters("messages", ["lang:fr"])
# Room-scoped presence
await room.set_presence({"status": "online"})
members = await room.fetch_presence()
# Unsubscribe within the room
await room.unsubscribe("messages")Subscribing to Topics
subscribe() registers interest in a topic.
Use on() separately to attach a message handler.
from nolag import SubscribeOptions, QoS
room = client.set_app(APP_SLUG).set_room("general")
# Basic subscription + handler
await room.subscribe("messages")
room.on("messages", lambda data, meta: print(data))
# With options (QoS, filters, load balancing)
options = SubscribeOptions(
qos=QoS.AT_LEAST_ONCE,
load_balance=True,
load_balance_group="workers",
filters=["priority:high", "region:us"]
)
await room.subscribe("tasks", options)
room.on("tasks", lambda data, meta: print(f"Task: {data}"))
# Client-level equivalent: the full app/room/topic path
await client.subscribe(f"{APP_SLUG}/general/messages")
client.on(f"{APP_SLUG}/general/messages", lambda data, meta: print(data))
# Unsubscribe
await room.unsubscribe("messages")Publishing Messages
from nolag import EmitOptions, QoS
# Publish any data (dict, list, string, bytes, etc.)
await room.emit("messages", {"text": "Hello!"})
# With options
options = EmitOptions(
qos=QoS.AT_LEAST_ONCE,
retain=True, # Broker keeps the last message for new subscribers
filter="priority:high", # Publish with a filter value
echo=False, # Per-connection flag; see the note below
)
await room.emit("status", {"online": True}, options)
# Client-level equivalent: the full path
await client.emit(f"{APP_SLUG}/general/messages", {"text": "Hello!"})Publishers never receive their own messages. The broker drops a message before delivering it to the actor that published it, whatever echo is set to. Append your own message to the UI locally. echo=False only adds a per-connection drop for the rare case of two connections sharing one actor token.
Filters
Filters let subscribers receive only messages matching specific criteria.
Publishers tag messages with a filter value via EmitOptions.filter,
and subscribers set which filters they care about.
from nolag import SubscribeOptions, EmitOptions
room = client.set_app(APP_SLUG).set_room("ops")
# Subscribe with initial filters
await room.subscribe("events", SubscribeOptions(
filters=["region:us", "priority:high"]
))
room.on("events", lambda data, meta: print(data, meta.filter))
# Replace all filters for a topic
await room.set_filters("events", ["region:eu"])
# Add filters to the existing set
await room.add_filters("events", ["priority:critical"])
# Remove specific filters
await room.remove_filters("events", ["region:eu"])
# Clear all filters (switches back to wildcard / receive all)
await room.set_filters("events", [])
# Publish with a filter value
await room.emit("events", {"msg": "US high-priority"},
EmitOptions(filter="region:us"))No filters = wildcard: Subscribing without filters receives all messages on the topic, including filtered publishes. Subscribing with filters only receives messages published with a matching filter. Messages published without a filter are only delivered to wildcard (no-filter) subscribers. Each filter is its own routing key, not a privacy boundary. Max 100 filters per topic; values must not contain /, #, + or |. See Filters.
Connection Events
from nolag import ConnectionStatus
# Listen for connection events
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}"))
# Hydration: with a hydration webhook configured on the app, the broker sends
# the webhook's response body once per subscribe. `topic` is the bare topic name.
client.on("hydration", lambda topic, data: print(f"Initial state for {topic}: {data}"))
# Check connection status
if client.status == ConnectionStatus.CONNECTED:
print("We're connected!")See Webhooks for configuring hydration.
Wildcard Handler (on_any / off_any)
Use on_any() to receive every message regardless of topic.
The handler receives (topic, data, meta), where topic is the full
app/room/topic path.
# Listen for ALL messages regardless of topic
def log_all(topic, data, meta):
print(f"[{topic}] {data} (from: {meta.sender})")
client.on_any(log_all)
# Remove a specific on_any handler
client.off_any(log_all)
# Remove ALL on_any handlers
client.off_any()Presence
Presence is scoped to rooms. Set it through the room context; presence events arrive on the client.
# Set presence scoped to a room (recommended)
room = client.set_app(APP_SLUG).set_room("general")
await room.set_presence({"status": "online", "username": "alice"})
# Or set presence with an explicit room ID
await client.set_presence({"status": "online"}, room_id="general")
# Fetch the presence list for a room (list of dicts from the broker)
actors = await room.fetch_presence()
for actor in actors:
print(f"{actor['actorTokenId']}: {actor['presence']}")
# Get cached presence for a specific actor
actor = client.get_presence("some-actor-token-id")
if actor:
print(f"{actor.actor_token_id}: {actor.presence}")
# Get all cached presence data
for actor in client.get_all_presence():
print(f"{actor.actor_token_id}: {actor.presence}")
# Clear your presence
await client.clear_presence()
# Presence events are client-level. Each carries actor_token_id and presence
client.on("presence:join", lambda actor: print(f"Joined: {actor.actor_token_id}"))
client.on("presence:leave", lambda actor: print(f"Left: {actor.actor_token_id}"))
client.on("presence:update", lambda actor: print(f"Updated: {actor.presence}"))client.set_presence() without a room_id is deprecated and is not broadcast.
Error Handling
Broker-side failures (a room that does not exist, a topic the actor cannot
write) are asynchronous: they arrive on the error event as NoLagServerError
with error (the machine name), code, topic and hint. Local failures
(not connected, a payload that cannot be msgpack-encoded) raise from the call.
Register the error handler before connect() so nothing is missed during the
handshake.
import asyncio
from nolag import NoLag, NoLagServerError, NoLagEncodeError
APP_SLUG = "chat-a3f9"
def on_error(err):
if isinstance(err, NoLagServerError):
# A structured frame from the broker (protocol v2)
print(f"{err.error} ({err.code}) on {err.topic}: {err.hint}")
if err.error == "unknown_topic":
# 42940: the room has not been created. Create it via the REST API.
pass
elif isinstance(err, NoLagEncodeError):
print(f"Cannot encode payload for {err.op}: {err}")
else:
print(f"Error: {err}")
async def main():
client = NoLag("your-actor-token")
client.on("error", on_error)
try:
await client.connect()
except Exception as e:
print(f"Connection failed: {e}")
return
room = client.set_app(APP_SLUG).set_room("general")
# Local failures raise; broker failures go to on_error
try:
await room.emit("messages", {"data": "value"})
except Exception as e:
print(f"Emit failed: {e}")
# Disconnect is synchronous, no await needed
client.disconnect()
asyncio.run(main())QoS Levels
| Level | Name | Description |
|---|---|---|
| 0 | AT_MOST_ONCE | Fire and forget on the broker hop |
| 1 | AT_LEAST_ONCE | Acknowledged on the broker hop (default) |
| 2 | EXACTLY_ONCE | Deduplicated on the broker hop |
QoS is a broker-hop setting, not an end-to-end guarantee. The level is validated (0, 1 or 2) and passed to the broker's internal MQTT hop. The WebSocket leg has an optional publish ack and no resend, so no level is an end-to-end delivery guarantee. Deduplicate with an idempotency key if a message must not be applied twice. See QoS.
REST API Client
The SDK also includes a REST API client for managing apps, rooms, actors, and scopes:
from nolag import NoLagApi, AppCreate, RoomCreate, ActorCreate
async def main():
# Create API client with project-scoped API key
async with NoLagApi("nlg_live_xxx.secret") as api:
# List apps: PaginatedResult with data and pagination
apps = await api.apps.list()
print(f"{len(apps.data)} of {apps.pagination.total} apps "
f"(page {apps.pagination.page} of {apps.pagination.page_count})")
# Create a new app. Without topics every subscribe is unknown_topic
app = await api.apps.create(AppCreate(
name="my-chat-app",
description="A real-time chat application",
topics=["messages"],
))
print(app.slug) # e.g. "my-chat-app-a3f9": pass this to set_app()
# Create a room in the app (rooms must exist before clients subscribe)
room = await api.rooms.create(app.app_id, RoomCreate(
name="general",
slug="general"
))
# Create an actor (save the access token!)
actor = await api.actors.create(ActorCreate(
name="web-client",
actor_type="device" # device, user, service, session, agent, orchestrator, observer
))
print(f"Actor token: {actor.access_token}")
# Scopes are paginated like apps; rooms and actors lists are plain lists
scopes = await api.scopes.list()
print(scopes.pagination.total)Load Balancing
Load balancing is opt-in. Without it every subscriber receives every message.
With it, subscribers that share a load_balance_group form a group and each
message is delivered to one member of the group:
from nolag import NoLag, NoLagOptions, SubscribeOptions
# Option 1: Enable load balancing globally via NoLagOptions
client = NoLag("your-actor-token", NoLagOptions(
load_balance=True,
load_balance_group="task-workers"
))
# Option 2: Enable per-subscription
room = client.set_app(APP_SLUG).set_room("image-processing")
await room.subscribe(
"tasks",
SubscribeOptions(
load_balance=True,
load_balance_group="task-workers"
)
)
room.on("tasks", lambda data, meta: print(f"Processing: {data}"))Workers whose actor type holds a persistent session (agent, orchestrator)
also have messages queued while they are away; see Replay.
Type Definitions
from nolag import (
# WebSocket Client
NoLag, # Main client class
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, # Message metadata
ActorPresence, # Presence info
NoLagServerError, # Structured broker error (error, code, topic, hint)
NoLagEncodeError, # Payload could not be msgpack-encoded
# REST API Client
NoLagApi, # REST API client
NoLagApiError, # API error class
PaginatedResult, # data + pagination
Pagination, # total, page, page_count
App, AppCreate, AppUpdate,
Room, RoomCreate, RoomUpdate,
Actor, ActorWithToken, ActorCreate, ActorUpdate,
Scope, ScopeCreate, ScopeUpdate,
)ActorCreate.actor_type accepts device, user, service, session, agent,
orchestrator or observer. Only agent and orchestrator hold a persistent
broker session.
Requirements
- Python 3.10+
- websockets >= 12.0
- msgpack >= 1.0.0
- aiohttp >= 3.9.0
Next Steps
- AI Agents SDK - build multi-agent workflows with
nolag-agents - Learn about Topics
- Rooms
- Presence Tracking
- Filters
- Replay
- REST API Reference