Python Client API

pip install useceleris-client

The package supports Python 3.10 to 3.14 and runs on asyncio. The runtime dependencies are pydantic, websockets and typing-extensions. Type information ships with the package (py.typed).

create_client()

def create_client(
    *,
    credential_provider: CredentialProvider,
    base_url: str = "wss://realtime.useceleris.com",
    allow_insecure_loopback: bool = False,
    connect_timeout_ms: int = 15_000,
    reconnect_timeout_ms: int | None = None,
    presence_query_timeout_ms: int = 10_000,
    publish_queue_size: int = 64,
    deduplication_window_size: int = 1024,
    maximum_reconnect_attempts: int = 10,
) -> Client: ...

The function performs no network work. The SDK validates the options strictly. Timeouts must be int milliseconds from 1 to 900000 (15 minutes). A larger value raises ConfigurationError. Sizes must be positive int values. maximum_reconnect_attempts must be an int from 1 to 100. Any other value raises ConfigurationError. The SDK refuses a bool, float or str value and does not convert it. base_url is the realtime socket endpoint, not your HTTP credential endpoint. The value must be wss://, or ws:// to a loopback host with allow_insecure_loopback=True. The value must carry no credentials, query string or fragment. Client(...) takes the same keyword arguments.

Client

class Client:
    def channel(self, reference: str) -> Channel: ...

channel() is side-effect free and returns a new connection handle on every call. A reference is 1 to 255 characters of [A-Za-z0-9_-].

Channel

class Channel:
    @property
    def state(self) -> ChannelState: ...
    async def connect(self) -> None: ...
    async def close(self) -> None: ...
    def segment(self, segment_id: str) -> Segment: ...
    def default_segment(self) -> Segment: ...  # segment("default")
    def events(self) -> ChannelEventHandler: ...


ChannelState = Literal[
    "idle", "connecting", "connected", "reconnecting", "failed", "closing", "closed"
]

A failed channel recovers when you call connect() again. A closed channel does not recover. The SDK does not retry initial connection failures automatically. See Reconnection and Recovery for triggers, deadlines, retry budgets and restored subscriptions.

A successful connect() joins default. default_segment().on_message(listener) receives its messages without subscribe(). Read and write permissions still apply. Default membership includes no other segment and no presence subscription. The automatic join also happens on reconnect.

Use a channel from the event loop thread that connected it.

Segment

class Segment:
    @property
    def segment_id(self) -> str: ...
    def subscribe(self) -> Subscription: ...
    def on_message(self, listener: MessageListener) -> Callable[[], None]: ...
    def on_presence(
        self, listener: Callable[[PresenceEvent], None]
    ) -> Callable[[], None]: ...
    async def publish(self, payload: bytes, *, message_id: str | None = None) -> None: ...
    def subscribe_presence(self) -> Subscription: ...
    async def presence_list(self, *, page: int, per_page: int) -> PresencePage: ...

Payload helpers

def text_payload(value: str) -> bytes: ...
def json_payload(value: object) -> bytes: ...
def read_text(payload: bytes) -> str: ...
def read_json(payload: bytes) -> Any: ...  # returns the JSON value; does not validate


def create_payload_codec(
    *, encode: Callable[[T], bytes], decode: Callable[[bytes], T]
) -> PayloadCodec[T]: ...


@dataclass(frozen=True)
class PayloadCodec(Generic[T]):
    encode_payload: Callable[[T], bytes]
    read_payload: Callable[[bytes], T]

text_payload raises ConfigurationError for text with unpaired surrogates. json_payload writes compact JSON, as JSON.stringify does. It raises ConfigurationError for NaN, infinities, circular structures and values that JSON cannot represent. read_text and read_json raise ConfigurationError for invalid UTF-8 or JSON. Errors raised by your own encode or decode propagate unchanged. No size check happens here. publish() enforces the encoded-command bound.

Types

@dataclass(frozen=True)
class MessageMetadata:
    token_reference: str
    segment_id: str
    message_id: str  # always present; the SDK generates one if you omit it
    timestamp: int  # Unix milliseconds


MessageListener = Callable[[bytes, MessageMetadata], None]


@dataclass(frozen=True)
class PresenceEvent:
    segment_id: str
    token_reference: str
    connection_id: str
    joined: bool  # False = left
    timestamp: int


@dataclass(frozen=True)
class PresencePage:
    segment_id: str
    total: int
    per_page: int
    current_page: int
    from_: int  # from_ > to past the last page
    to: int
    connections: tuple[PresenceConnection, ...]


@dataclass(frozen=True)
class PresenceConnection:
    token_reference: str
    connection_id: str
    timestamp: int


@dataclass(frozen=True)
class ServerNotice:
    timestamp: int
    payload: bytes


@dataclass(frozen=True)
class RecoveryEvent:
    retry_index: int
    possible_gaps: Literal[True] = True
    possible_duplicates: Literal[True] = True


@dataclass(frozen=True)
class CredentialRequest:
    channel_reference: str
    reason: Literal["initial", "reconnect"]
    disconnected_at: int | None = None  # Unix ms at the start of the outage
    replay_lookback_ms: int | None = None  # at most 4294967295


@dataclass(frozen=True)
class Credentials:
    payload: str  # both left out of repr()
    signature: str


CredentialProvider = Callable[[CredentialRequest], Awaitable[Credentials]]


class Subscription:
    def cancel(self) -> None: ...  # idempotent

Every number is an int. Presence figures are signed 32-bit values as received. Timestamps are signed 64-bit Unix milliseconds. from_ carries a trailing underscore because from is a Python keyword. A credential provider must return a Credentials instance with non-empty, well-formed strings.

ChannelEventHandler

class ChannelEventHandler:
    def on_state_change(self, listener: Callable[[ChannelState], None]) -> Callable[[], None]: ...
    def on_recovery(self, listener: Callable[[RecoveryEvent], None]) -> Callable[[], None]: ...
    def on_notice(self, listener: Callable[[ServerNotice], None]) -> Callable[[], None]: ...
    def on_error(self, listener: Callable[[ChannelError], None]) -> Callable[[], None]: ...
    def on_message(self, listener: MessageListener) -> Callable[[], None]: ...


ChannelError = ConfigurationError | CelerisConnectionError | ProtocolError | ServerError

Each registration returns a function that removes it. Listeners run synchronously in registration order with no queue. A slow listener delays delivery and does not grow a backlog. The SDK contains the failure of a listener that raises, or of one that returns a coroutine, and reports it through on_error. The SDK swallows failures of the error listeners themselves. A recovery event follows the connected state change. It does not signal replay completion.

on_message receives all messages that the connection gets, from all segments. It receives each message after the segment listeners of that segment. It receives each message ID one time.

on_notice delivers raw server prose, such as subscription acknowledgments and refusals. It carries no segment id and no correlation id. Do not parse it.

Errors

Every error is a CelerisError with a code and a message. The SDK itself raises three kinds: ConfigurationError (code "Configuration"), CelerisConnectionError (a code from the table below) and ProtocolError (code "ProtocolError", with field and offset). Their messages say what failed and which rule or limit it broke, such as Invalid channel reference. Must not be empty.. The messages never carry your input, credentials or server text. The errors never chain an exception that holds them. The fourth kind, ServerError, carries an error that the server sent, described below. Match on code, and on type for server errors, never on message text.

If you cancel a task that you own, asyncio raises asyncio.CancelledError, not an SDK error. ProtocolError means that the SDK could not decode one message. The SDK drops that message and reports the error, and the connection stays up. The SDK silently skips a command that this SDK version does not recognize, so a newer server cannot break a deployed client.

There is no authentication code. The SDK does not expose the WebSocket handshake status. The SDK therefore reports a refused handshake as Transport and does not guess a more specific code.

ServerError

ServerErrorType = Literal[
    "ParserError",
    "SendError",
    "PermissionDeniedError",
    "RateLimitError",
    "MessageSizeLimitError",
    "InternalError",
]

ServerErrorResource = str | int | tuple["ServerErrorResource", ...] | None


class ServerError(CelerisError):  # code == "Server"
    type: str  # one of ServerErrorType, or a type a newer server adds
    sub_type: str | None  # the command it answers, e.g. "PUB"
    message: str  # the server's own text, unchanged
    resource: ServerErrorResource  # what that command names, e.g. the segment

ServerError carries an error that the server sent, with every field as sent. The connection stays up. It arrives through on_error, after the call that caused it, except for a presence query error, which raises from presence_list() instead. Keep a default branch for types that a newer server adds. Check resource's type before you use it.

Limits

The SDK never checks the size of received messages. The messages are already in memory when they arrive, so the SDK processes whatever the server sends. A publish rejected for its plan payload limit still counts toward your usage. A local encoded-command rejection sends nothing.

What this SDK will not do

• The SDK does not confirm delivery. There are no receipts anywhere in the protocol.
• The SDK does not keep a durable offline queue. While the channel is reconnecting, publishes wait in an in-memory queue. In any other state without a connection, a publish raises NotConnected. The SDK resends publishes only after a rate limit, once each, with their original ID.
• The SDK does not provide durable history or global ordering. Replay is a bounded window, not a cursor.
• The SDK does not sign anything. The signer lives in useceleris-server.