Subscription

The runtime behind every AT Protocol subscription: a reconnecting websocket client, the XRPC frame format, and the callback loop that hands decoded frames to your code. It is lexicon-agnostic: it knows how a subscription behaves, not what any particular one carries.

Three things sit on top of it. Firehose and Jetstream are the two you use directly, and every generated subscription client in atproto_client.subscriptions, including the ones with no hand-written wrapper such as ChatBskyModerationSubscribeModEventsClient, is a SubscriptionClient with a method name and a parser bound to it.

You reach for this package directly when you are subscribing to a lexicon the SDK has no convenience client for, and when you want the semantics: what recv_timeout does, when the client reconnects, and how update_params survives one. Those are covered in Firehose, and they apply to every subscription.

from atproto_subscription import SubscriptionClient

client = SubscriptionClient(
    method='com.example.subscribeThings',
    base_uri='wss://example.com/xrpc',
    params={'cursor': 42},
)


def on_message_handler(message) -> None:
    print(message.header, message.body)


client.start(on_message_handler)

start blocks until stop is called, the server closes cleanly, or the server sends an error frame. AsyncSubscriptionClient is the same client with awaited callbacks.

Note

This package was factored out of atproto_firehose, which is why the frame models are re-exported as atproto.firehose_models. atproto_firehose.client and atproto_firehose.models are deprecation shims that forward here and warn: FirehoseClient and AsyncFirehoseClient are now SubscriptionClient and AsyncSubscriptionClient, and the frame models moved to atproto_subscription.frames. The FirehoseSubscribeReposClient and FirehoseSubscribeLabelsClient you actually use are unaffected.

atproto_subscription.AsyncSubscriptionClient: alias of _AsyncWebsocketClient

class atproto_subscription.AsyncWebsocketClient(*args: Any, **kwargs: Any)

Reconnecting asynchronous websocket client.

async start(on_message_callback: Callable[[Any], Coroutine[Any, Any, None]], on_callback_error_callback: Callable[[BaseException], Coroutine[Any, Any, None]] | None = None) → None

Subscribe and start the client.

Parameters:

on_message_callback – Callback that will be called on the new message.
on_callback_error_callback – Callback that will be called if the on_message_callback raised an exception.

Returns:

None

async stop() → None

Unsubscribe and stop the client.

Safe to call from another task. The client stops even if it is currently waiting for the next frame on an idle connection.

Returns:: None

update_params(params: Dict[str, Any]) → None

Update params.

Parameters:: params – Query params.
Returns:: None

class atproto_subscription.ErrorFrame(header: ErrorFrameHeader, body: ErrorFrameBody)

Subscription error frame.

static from_bytes(data: bytes | bytearray) → MessageFrame | ErrorFrame

Decode frame from bytes of stream of bytes.

Parameters:: data – Bytes or stream of bytes of frame.
Returns:: MessageFrame or ErrorFrame
Raises:: atproto.exceptions.SubscriptionError – Invalid data frame.

property is_error: bool: bool: Is frame the ErrorFrame.

property is_message: bool: bool: Is frame the MessageFrame.

property operation: FrameType: FrameType: Frame operation (frame type).

header: ErrorFrameHeader: Header.

body: ErrorFrameBody: Body.

class atproto_subscription.ErrorFrameBody(error: str, message: str | None = None)

Body of error frame.

message: str | None = None: Description of the error.

error: str: Code of the error.

class atproto_subscription.ErrorFrameHeader(op: FrameType = FrameType.ERROR)

Header of the error frame.

op: FrameType = -1: Operation. For Error header is FrameType.ERROR always.

class atproto_subscription.Frame(header: MessageFrameHeader | ErrorFrameHeader, body: ErrorFrameBody | dict)

Base subscription frame.

static from_bytes(data: bytes | bytearray) → MessageFrame | ErrorFrame

Decode frame from bytes of stream of bytes.

Parameters:: data – Bytes or stream of bytes of frame.
Returns:: MessageFrame or ErrorFrame
Raises:: atproto.exceptions.SubscriptionError – Invalid data frame.

property is_error: bool: bool: Is frame the ErrorFrame.

property is_message: bool: bool: Is frame the MessageFrame.

property operation: FrameType: FrameType: Frame operation (frame type).

header: MessageFrameHeader | ErrorFrameHeader: Header.

body: ErrorFrameBody | dict: Body

exception atproto_subscription.FrameDecodingError

Frame could not be decoded.

add_note(object, /): Exception.add_note(note) – add a note to the exception

args

with_traceback(object, /): Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class atproto_subscription.FrameType(*values)

Type of frame.

classmethod has_value(value: int) → bool

MESSAGE = 1

ERROR = -1

class atproto_subscription.MessageFrame(header: MessageFrameHeader, body: dict)

Subscription message frame.

static from_bytes(data: bytes | bytearray) → MessageFrame | ErrorFrame

Decode frame from bytes of stream of bytes.

Parameters:: data – Bytes or stream of bytes of frame.
Returns:: MessageFrame or ErrorFrame
Raises:: atproto.exceptions.SubscriptionError – Invalid data frame.

property is_error: bool: bool: Is frame the ErrorFrame.

property is_message: bool: bool: Is frame the MessageFrame.

property operation: FrameType: FrameType: Frame operation (frame type).

property type: str: str: Type of body.

header: MessageFrameHeader: Header.

body: dict: Body.

class atproto_subscription.MessageFrameHeader(op: FrameType = FrameType.MESSAGE, t: str | None = None)

Header of the message frame.

op: FrameType = 1: Operation. For Message header is FrameType.MESSAGE always.

t: str | None = None: Type of body content.

atproto_subscription.SubscriptionClient: alias of _WebsocketClient

exception atproto_subscription.SubscriptionError

Base exception of the subscription runtime.

add_note(object, /): Exception.add_note(note) – add a note to the exception

args

with_traceback(object, /): Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

class atproto_subscription.WebsocketClient(*args: Any, **kwargs: Any)

Reconnecting synchronous websocket client.

start(on_message_callback: Callable[[Any], None], on_callback_error_callback: Callable[[BaseException], None] | None = None) → None

Subscribe and start the client.

Parameters:

on_message_callback – Callback that will be called on the new message.
on_callback_error_callback – Callback that will be called if the on_message_callback raised an exception.

Returns:

None

stop() → None

Unsubscribe and stop the client.

Safe to call from another thread. The client stops even if it is currently waiting for the next frame on an idle connection.

Returns:: None

update_params(params: Dict[str, Any]) → None

Update params.

Parameters:: params – Query params.
Returns:: None

class atproto_subscription.WebsocketClientBase(method: str, base_uri: str, params: Dict[str, Any] | None = None, recv_timeout: float | None = None, max_message_size_bytes: int | None = None, subprotocols: Sequence[str] | None = None)

Base of the reconnecting websocket clients.

Subclasses own frame decoding; this class owns the connection lifecycle.

update_params(params: Dict[str, Any]) → None

Update params.

Parameters:: params – Query params.
Returns:: None

atproto_subscription.build_websocket_uri(method: str, base_uri: str, params: Dict[str, Any] | None = None) → str

Build an XRPC subscription URI.

Parameters:

method – NSID of the subscription.
base_uri – Base websocket URI.
params – Query params.

Returns:

Websocket URI.

Return type:

str

atproto_subscription.parse_frame(header: MessageFrameHeader | ErrorFrameHeader, raw_body: dict) → ErrorFrame | MessageFrame

atproto_subscription.parse_frame_header(raw_header: dict) → MessageFrameHeader | ErrorFrameHeader

Subscription

Submodules