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. Keep your Celeris client ID and signing secret in that server's configuration. Never put them in an application that you ship to users.

1. Understand the two responsibilities

useceleris-server signs access. It does not provide a separate publish transport. A backend that publishes or receives uses useceleris-client, as any other application does. That backend can sign its own credentials through create_credential_provider().

Applications that 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 on the server, and signs fresh credentials 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). Call it only after your application decides that the user can access the requested channel. Never sign permissions only because the caller asked for them. Never cache or backdate credentials. sign() is synchronous and validates the claims. For an invalid claim, it raises a ConfigurationError that names the claim and the rule that failed.

Scope every credential to what the user needs. Restricted channels and segments are explicit. "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)). That call raises a ConfigurationError that names them.

This worker explicitly restricts its channel and segment and caps the requested replay at 30 seconds. The claims function can also be an async def, for claims that need a database lookup. The provider calls it on every connection attempt and signs freshly. When you cancel the attempt while the claims function runs, the SDK cancels the function too, and nothing is signed.

A publish has the same local-acceptance semantics on a server as anywhere else. A completed publish() means that the socket accepted the bytes, not that delivery is confirmed.

4. Apply this to your service

Sign from server-owned configuration. For multi-tenant services, authorize each channel. Do not replace the restriction with 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 messages.

Next: Python server API and Python client API.

Errors and Troubleshooting Changelog