Python Server API

pip install --pre useceleris-server

Python 3.10 to 3.14, on trusted servers only. This package holds your signing secret and must never ship in a browser, mobile or desktop application. It installs the matching useceleris-client.

create_signer()

def create_signer(
    *,
    client_id: str,  # no colon, CR or LF
    signing_secret: str,
    clock: Callable[[], int] | None = None,  # Unix ms; a test seam
) -> Signer: ...


class Signer:
    def sign(self, claims: SigningClaims) -> Credentials: ...

sign() is synchronous and deterministic for fixed inputs. It validates the claims, stamps the current time and returns useceleris_client.Credentials. A Signer's repr() never shows the secret, and it holds no resources. Signer(...) takes the same keyword arguments as create_signer.

import os

from useceleris_server import create_signer

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

credentials = signer.sign(
    {
        "channels": {"kind": "restricted", "references": ["room-42"]},
        "permissions": {
            "kind": "restricted",
            "segments": [{"segment_id": "chat", "read": True, "write": True}],
        },
        "reference": "user-8317",
        "replay": {"lookback_ms": 30_000},
    }
)

Return credentials.payload and credentials.signature to the application unchanged. They are opaque: never parse, re-encode or cache them.

The claims surface

class SigningClaims(TypedDict):
    channels: ChannelScope
    permissions: SegmentPermissions
    reference: NotRequired[str]
    replay: NotRequired[bool | ReplayLookback]
    allow_echo: NotRequired[bool]


ChannelScope = AllChannels | RestrictedChannels  # {"kind": "all"} | {"kind": "restricted", "references": [...]}
SegmentPermissions = AllSegments | RestrictedSegments


class SegmentClaim(TypedDict):
    segment_id: str
    read: bool
    write: bool


class ReplayLookback(TypedDict):
    lookback_ms: int

Validation is strict: 1 is not a bool, a tuple is not a list, and text with unpaired surrogates is refused. Keys beyond these are ignored, so never rely on an unknown key to grant access or enforce a restriction.

Unrestricted access is always an explicit opt-in:

signer.sign(
    {
        "channels": {"kind": "all"},
        "permissions": {"kind": "all", "read": True, "write": False},
    }
)

create_credential_provider()

When a backend is itself a realtime client, a worker publishing events for example, this bridges the signer to the client's provider.

def create_credential_provider(
    *, signer: SupportsSign, claims: ClaimsCallback
) -> CredentialProvider: ...


ClaimsCallback = Callable[[CredentialRequest], SigningClaims | Awaitable[SigningClaims]]


class SupportsSign(Protocol):
    def sign(self, claims: SigningClaims) -> Credentials: ...

The provider calls claims(request) on every attempt, awaits it if it returns an awaitable, then signs. Cancelling the attempt while claims is awaited cancels it, so nothing is signed; the synchronous signer itself is not interruptible. CredentialRequest and Credentials are re-exported by the server package; see the client types for their fields.

from useceleris_client import create_client

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:
    return {
        "channels": {"kind": "restricted", "references": [request.channel_reference]},
        "permissions": {
            "kind": "restricted",
            "segments": [{"segment_id": "chat", "read": True, "write": True}],
        },
        "reference": "worker-1",
        "replay": (
            {"lookback_ms": min(request.replay_lookback_ms, 30_000)}
            if request.replay_lookback_ms is not None
            else False
        ),
    }


client = create_client(
    credential_provider=create_credential_provider(signer=signer, claims=claims_for)
)
channel = client.channel("room-42")
await channel.connect()

Your claims function runs freshly on every attempt and is the sole authority over scope: nothing from the request is copied into the claims automatically. This example's channel is chosen by trusted worker code; in an HTTP credential endpoint, authorize the requested channel against the verified user's access before signing it. The 30-second replay cap is an example policy, not a retention promise.

Invalid claims returned through the provider surface on the client as its generic Transport error, which never repeats what went wrong. Check a claims function by signing its result directly, signer.sign(claims_for(request)), which raises a ConfigurationError naming the claim and the rule.

Fresh signing does not guarantee a particular credential lifetime or revoke an existing connection. Keep signed credentials confidential and pass them unchanged.

Errors

Invalid options and claims raise useceleris_client.ConfigurationError (code "Configuration"), a CelerisError like every client error. Its message names each option or claim that failed and the rule it broke, such as Invalid claims. channels.references[0]: Must not be empty., and never repeats your input, secrets or claims. An unknown keyword argument is Python's own TypeError.

An error raised by your own claims function propagates unchanged from the provider; through connect() it surfaces as Transport, like any provider failure.