Firehose (data streaming)

The firehose clients and the deployment values they carry: today’s relay host, the reconnect timeout, and the per-lexicon message parsers. The subscription machinery underneath them lives in atproto_subscription.

Four clients, sync and async, over two lexicons:

FirehoseSubscribeReposClient / AsyncFirehoseSubscribeReposClient

com.atproto.sync.subscribeRepos, every repository commit on the network.

FirehoseSubscribeLabelsClient / AsyncFirehoseSubscribeLabelsClient

com.atproto.label.subscribeLabels, labels published by a moderation service.

Tip

For how to use these, covering decoding commits, filtering, cursors, reconnects and choosing between the firehose and Jetstream, see the Firehose guide.

Note

atproto_firehose.client and atproto_firehose.models are deprecation shims. The subscription runtime moved to atproto_subscription.client and the frame models to atproto_subscription.frames; atproto.firehose_models points at the latter and is not deprecated.

class atproto_firehose.AsyncFirehoseSubscribeLabelsClient(params: dict | models.ComAtprotoLabelSubscribeLabels.Params | None = None, base_uri: str | None = 'wss://mod.bsky.app/xrpc', recv_timeout: float | None = 300.0)

Async firehose subscribe labels client.

Parameters:

  • params – Parameters model.

  • base_uri – Base websocket URI. Example: wss://bsky.social/xrpc.

  • recv_timeout – Reconnect to the server after this many seconds of inactivity. Default is 300 seconds (5 minutes).

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: ParamsModelBase | Dict[str, Any]) → None

Update params.

Warning

If you are using params arg at the client start, you must care about keeping params up to date. Otherwise, your client will be rolled back to the previous state (cursor) on reconnecting.

class atproto_firehose.AsyncFirehoseSubscribeReposClient(params: dict | models.ComAtprotoSyncSubscribeRepos.Params | None = None, base_uri: str | None = 'wss://bsky.network/xrpc', recv_timeout: float | None = 30.0)

Async firehose subscribe repos client.

Parameters:

  • params – Parameters model.

  • base_uri – Base websocket URI. Example: wss://bsky.social/xrpc.

  • recv_timeout – Reconnect to the server after this many seconds of inactivity. Default is 30 seconds.

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: ParamsModelBase | Dict[str, Any]) → None

Update params.

Warning

If you are using params arg at the client start, you must care about keeping params up to date. Otherwise, your client will be rolled back to the previous state (cursor) on reconnecting.

class atproto_firehose.FirehoseSubscribeLabelsClient(params: dict | models.ComAtprotoLabelSubscribeLabels.Params | None = None, base_uri: str | None = 'wss://mod.bsky.app/xrpc', recv_timeout: float | None = 300.0)

Firehose subscribe labels client.

Parameters:

  • params – Parameters model.

  • base_uri – Base websocket URI. Example: wss://bsky.social/xrpc.

  • recv_timeout – Reconnect to the server after this many seconds of inactivity. Default is 300 seconds (5 minutes).

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: ParamsModelBase | Dict[str, Any]) → None

Update params.

Warning

If you are using params arg at the client start, you must care about keeping params up to date. Otherwise, your client will be rolled back to the previous state (cursor) on reconnecting.

class atproto_firehose.FirehoseSubscribeReposClient(params: dict | models.ComAtprotoSyncSubscribeRepos.Params | None = None, base_uri: str | None = 'wss://bsky.network/xrpc', recv_timeout: float | None = 30.0)

Firehose subscribe repos client.

Parameters:

  • params – Parameters model.

  • base_uri – Base websocket URI. Example: wss://bsky.social/xrpc.

  • recv_timeout – Reconnect to the server after this many seconds of inactivity. Default is 30 seconds.

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: ParamsModelBase | Dict[str, Any]) → None

Update params.

Warning

If you are using params arg at the client start, you must care about keeping params up to date. Otherwise, your client will be rolled back to the previous state (cursor) on reconnecting.

atproto_firehose.parse_subscribe_labels_message(message: MessageFrame) → ComAtprotoLabelSubscribeLabelsMessage

Parse a message frame of com.atproto.label.subscribeLabels into its model.

Parameters:

message – Message frame.

Returns:

Corresponding message model.

Return type:

ComAtprotoLabelSubscribeLabelsMessage

atproto_firehose.parse_subscribe_repos_message(message: MessageFrame) → ComAtprotoSyncSubscribeReposMessage

Parse a message frame of com.atproto.sync.subscribeRepos into its model.

Parameters:

message – Message frame.

Returns:

Corresponding message model.

Return type:

ComAtprotoSyncSubscribeReposMessage

Submodules