Jetstream (data streaming)

Jetstream delivers the same events as the firehose as plain JSON, filtered server-side, with no CAR or DAG-CBOR decoding. This package holds the client, its models, and the archive replay.

JetstreamClient / AsyncJetstreamClient

The live tail, plus snapshot() and replay() over the archive.

Tip

For how to use these, covering filters, cursors, compression, archive replay and what the archive costs, see the Jetstream guide.

Note

Only the Jetstream v2 wire is supported. The legacy v1 hosts (jetstream1.*, jetstream2.*) speak a different, frozen protocol and will not work with this client.

Warning

Jetstream carries no repository signatures or MST proofs, so its data cannot be cryptographically verified. Use FirehoseSubscribeReposClient when verifiability matters.

class atproto_jetstream.AsyncJetstreamClient(params: Params | ParamsDict | None = None, base_uri: str = 'wss://jetstream.us-east.bsky.network/xrpc', recv_timeout: float | None = 60.0, compress: bool = True, api_key: str | None = None)

Async jetstream v2 client.

Note

Only the v2 wire is supported. Legacy v1 hosts (jetstream1.*, jetstream2.*) will not work.

Note

The cursor is tracked for you. Reconnects resume from the last delivered event and redelivered events are dropped. Persist cursor to resume across restarts.

Parameters:

  • params – Parameters model.

  • base_uri – Base websocket URI. Example: wss://jetstream.us-east.bsky.network/xrpc.

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

  • compress – Receive compressed frames. Falls back to an uncompressed stream if the server does not cooperate. Default is True.

  • api_key – Archive credential, obtainable at https://bsky.network/account. Used only by snapshot and replay; never sent on the websocket. The live tail needs no key.

property bytes_downloaded: int

int: Archive bytes downloaded. Jetstream meters usage in bytes, not requests.

property compressed: bool

bool: Whether frames are being received compressed.

Compression is best-effort: it degrades to an uncompressed stream rather than failing.

property cursor: int | None

int: Seq of the last delivered event, or None if nothing was delivered yet.

Persist it to resume the stream later. Reconnects resume from it automatically.

async replay(after_seq: int = 0, *, with_cid: bool = True) → AsyncIterator[Commit | Identity | Account | Sync | Info]

Replay the archive, then cut over to the live tail without a gap.

Note

Never terminates: it becomes the live tail once the archive is consumed.

Parameters:

  • after_seq – Exclusive lower bound. 0 means the whole archive.

  • with_cid – Derive each record’s CID for archived events.

Yields:

SubscribeEventsMessage – The same models the live tail delivers.

async snapshot(after_seq: int = 0, before_seq: int | None = None, *, with_cid: bool = True) → AsyncIterator[Commit | Identity | Account | Sync | Info]

Replay the sealed archive, then stop.

Note

A point-in-time view: rows still in the unsealed active segment are not included. Use replay to continue into the live tail instead.

Parameters:

  • after_seq – Exclusive lower bound. 0 means the whole archive.

  • before_seq – Inclusive upper bound.

  • with_cid – Derive each record’s CID. Skipping it saves a hash per record.

Yields:

SubscribeEventsMessage – The same models the live tail delivers.

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_jetstream.JetstreamClient(params: Params | ParamsDict | None = None, base_uri: str = 'wss://jetstream.us-east.bsky.network/xrpc', recv_timeout: float | None = 60.0, compress: bool = True, api_key: str | None = None)

Jetstream v2 client.

Note

Only the v2 wire is supported. Legacy v1 hosts (jetstream1.*, jetstream2.*) will not work.

Note

The cursor is tracked for you. Reconnects resume from the last delivered event and redelivered events are dropped. Persist cursor to resume across restarts.

Parameters:

  • params – Parameters model.

  • base_uri – Base websocket URI. Example: wss://jetstream.us-east.bsky.network/xrpc.

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

  • compress – Receive compressed frames. Falls back to an uncompressed stream if the server does not cooperate. Default is True.

  • api_key – Archive credential, obtainable at https://bsky.network/account. Used only by snapshot and replay; never sent on the websocket. The live tail needs no key.

property bytes_downloaded: int

int: Archive bytes downloaded. Jetstream meters usage in bytes, not requests.

property compressed: bool

bool: Whether frames are being received compressed.

Compression is best-effort: it degrades to an uncompressed stream rather than failing.

property cursor: int | None

int: Seq of the last delivered event, or None if nothing was delivered yet.

Persist it to resume the stream later. Reconnects resume from it automatically.

replay(after_seq: int = 0, *, with_cid: bool = True) → Iterator[Commit | Identity | Account | Sync | Info]

Replay the archive, then cut over to the live tail without a gap.

Note

Never terminates: it becomes the live tail once the archive is consumed.

Parameters:

  • after_seq – Exclusive lower bound. 0 means the whole archive.

  • with_cid – Derive each record’s CID for archived events.

Yields:

SubscribeEventsMessage – The same models the live tail delivers.

snapshot(after_seq: int = 0, before_seq: int | None = None, *, with_cid: bool = True) → Iterator[Commit | Identity | Account | Sync | Info]

Replay the sealed archive, then stop.

Note

A point-in-time view: rows still in the unsealed active segment are not included. Use replay to continue into the live tail instead.

Parameters:

  • after_seq – Exclusive lower bound. 0 means the whole archive.

  • before_seq – Inclusive upper bound.

  • with_cid – Derive each record’s CID. Skipping it saves a hash per record.

Yields:

SubscribeEventsMessage – The same models the live tail delivers.

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

atproto_jetstream.parse_subscribe_events_message(message: MessageFrame) → models.NetworkBskyJetstreamSubscribeEvents.Commit | models.NetworkBskyJetstreamSubscribeEvents.Identity | models.NetworkBskyJetstreamSubscribeEvents.Account | models.NetworkBskyJetstreamSubscribeEvents.Sync | models.NetworkBskyJetstreamSubscribeEvents.Info

Parse Jetstream message to the corresponding model.

Parameters:

message – Message frame.

Returns:

Corresponding message model.

Return type:

SubscribeEventsMessage

Raises:

atproto.exceptions.JetstreamDecodingError – Unknown message type.

Submodules