Python Server API

pip install useceleris-server

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

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. A Signer 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, and a tuple is not a list. The signer refuses text with unpaired surrogates. The signer ignores keys beyond these, 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, for example a worker that publishes events, this function 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, and then signs. If you cancel the attempt while the provider awaits claims, the provider cancels that call and signs nothing. The synchronous signer itself is not interruptible. The server package re-exports CredentialRequest and Credentials. 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 again on every attempt and is the sole authority over scope. The provider copies nothing from the request into the claims automatically. Trusted worker code chooses this example's channel. In an HTTP credential endpoint, authorize the requested channel against the verified user's access before you sign it. The 30-second replay cap is an example policy, not a retention promise.

The client reports invalid claims from the provider as its generic Transport error. That error never repeats what went wrong. To check a claims function, sign its result directly with signer.sign(claims_for(request)). An invalid result raises a ConfigurationError that names the claim and the rule.

A fresh signature does not guarantee a particular credential lifetime, and it does not 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.. The message never repeats your input, secrets or claims. An unknown keyword argument raises Python's own TypeError.

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