Server-side Usage

Sign credentials on a trusted Python server, serve them to your applications, and connect from a backend.

Before you start: install useceleris-server on a trusted server, and keep your Celeris client ID and signing secret in that server's configuration, never in an application you ship to users.

1. Understand the two responsibilities

useceleris-server signs access. It has no separate publishing transport: a backend that publishes or receives uses useceleris-client, as any other application does, but signs its own credentials through create_credential_provider().

Backend signer -> backend client -> Celeris -> subscribers

Applications you ship to users fetch credentials from an authenticated endpoint on your server instead; see Authentication.

2. Serve credentials to your applications

The endpoint authenticates the user, decides the claims server-side, and signs fresh for every request:

import os

from useceleris_server import SigningClaims, create_signer

signer = create_signer(
    client_id=os.environ["CELERIS_CLIENT_ID"],
    signing_secret=os.environ["CELERIS_SIGNING_SECRET"],
)


def credentials_for(user_id: str, channel_reference: str, may_write: bool) -> dict[str, str]:
    # Your application decided this user may access this channel.
    claims: SigningClaims = {
        "channels": {"kind": "restricted", "references": [channel_reference]},
        "permissions": {
            "kind": "restricted",
            "segments": [{"segment_id": "chat", "read": True, "write": may_write}],
        },
        "reference": user_id,  # the identity peers see; no colons
        "replay": {"lookback_ms": 30_000},
    }
    credentials = signer.sign(claims)

    return {"payload": credentials.payload, "signature": credentials.signature}

Call it from your framework's authenticated route (Django, Flask, FastAPI or anything else) after deciding whether the user may access the requested channel. Never sign permissions the caller asked for, and never cache or backdate credentials. sign() is synchronous and validates the claims, raising ConfigurationError naming the claim and rule that failed.

Scope every credential to what the user needs: restricted channels and segments are explicit, and "kind": "all" is an opt-in, never a default. Replay and echo default to off.

3. Connect from a backend

import asyncio
import os

from useceleris_client import ChannelError, MessageMetadata, ServerError, create_client, read_text, text_payload

from useceleris_server import CredentialRequest, SigningClaims, create_credential_provider, create_signer

signer = create_signer(
    client_id=os.environ["CELERIS_CLIENT_ID"],
    signing_secret=os.environ["CELERIS_SIGNING_SECRET"],
)


def claims_for(request: CredentialRequest) -> SigningClaims:
    if request.channel_reference != "demo-room":
        raise ValueError("Worker channel is not allowed.")

    return {
        "channels": {"kind": "restricted", "references": ["demo-room"]},
        "permissions": {
            "kind": "restricted",
            "segments": [{"segment_id": "chat", "read": True, "write": True}],
        },
        "reference": "demo-worker",
        "replay": (
            False
            if request.replay_lookback_ms is None
            else {"lookback_ms": min(request.replay_lookback_ms, 30_000)}
        ),
    }


def report(error: ChannelError) -> None:
    print(error.type if isinstance(error, ServerError) else error.code, error)


def receive(payload: bytes, metadata: MessageMetadata) -> None:
    print("Received:", read_text(payload))


async def main() -> None:
    client = create_client(
        credential_provider=create_credential_provider(signer=signer, claims=claims_for)
    )
    channel = client.channel("demo-room")
    chat = channel.segment("chat")
    channel.events().on_error(report)
    chat.on_message(receive)
    chat.subscribe()

    try:
        await channel.connect()
        await chat.publish(text_payload("Hello from the backend"))
        await asyncio.Event().wait()  # run until cancelled, e.g. Ctrl+C
    finally:
        await channel.close()


asyncio.run(main())

The claims function is the authority: nothing from the request turns into permission by itself. If it returns invalid claims, connect() fails with the client's generic Transport error; to see which claim and rule failed, sign its result directly with signer.sign(claims_for(request)), which raises a ConfigurationError naming them. This worker restricts its channel and segment and caps the requested replay at 30 seconds. It may also be an async def, for claims that need a database lookup. The provider calls it on every connection attempt and signs freshly; cancelling the attempt while it runs cancels it, and nothing is signed.

Publishing has the same local-acceptance semantics on a server as anywhere else: when publish() returns, the bytes were handed to the socket, not confirmed delivered.

4. Apply this to your service

Sign from server-owned configuration. For multi-tenant services, authorize each channel instead of widening to unrestricted claims. The signer holds no resources and needs no disposal; the client channel owns the socket. Use Reconnection and Recovery to decide how your service reconciles missed or duplicate updates.

Next: Python server API and Python client API.

Errors and Troubleshooting API Reference