---
title: Python SDK
description: Complete reference for the NoLag Python SDK. Connect, subscribe, publish, and manage real-time messaging with async/await support.
---

# Python SDK

The official NoLag SDK for Python. Full async/await support with type hints
for Python 3.10+.

## Installation

```bash [Terminal]
# pip
pip install nolag

# poetry
poetry add nolag

# uv
uv add nolag
```

## Quick Start

```python [Python]
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](#rest-api-client) 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

```python [Python]
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.

```python [Python]
# 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.

```python [Python]
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

```python [Python]
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.

```python [Python]
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](/docs/concepts/filters).

## Connection Events

```python [Python]
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](/docs/concepts/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.

```python [Python]
# 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.

```python [Python]
# 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.

```python [Python]
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](/docs/concepts/qos).

## REST API Client

The SDK also includes a REST API client for managing apps, rooms, actors, and scopes:

```python [Python]
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:

```python [Python]
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](/docs/concepts/replay).

## Type Definitions

```python [Python]
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](/docs/agents/getting-started) - build multi-agent workflows with `nolag-agents`
- [Learn about Topics](/docs/concepts/topics)
- [Rooms](/docs/concepts/rooms)
- [Presence Tracking](/docs/concepts/presence)
- [Filters](/docs/concepts/filters)
- [Replay](/docs/concepts/replay)
- [REST API Reference](/docs/api-reference)