Websocket client

The connection layer under every subscription client, and under Jetstream. It owns the lifecycle (dial, receive loop, callback dispatch, reconnect) and leaves frame decoding to its subclasses, which implement _decode_frame and _handle_frame_decoding_error.

Reconnection is automatic and exponential: the delay doubles per attempt up to 64 seconds, with a random offset of up to half a second so that a fleet of consumers does not reconnect in lockstep. A connection that stayed up for at least a minute resets the backoff to its base delay instead of retrying instantly, because a server restart drops every consumer at once. A clean close by the server ends the loop; a transport failure does not.

Exceptions raised by your callback never reach the loop: they go to the error callback if you passed one, and are printed otherwise.

atproto_subscription.websocket.Frame

A decoded frame. Its type is defined by the subclass.

atproto_subscription.websocket.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

class atproto_subscription.websocket.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)

Bases: object

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

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

Bases: WebsocketClientBase

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

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

Bases: WebsocketClientBase

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