Jetstream (data streaming)

Jetstream is a streaming service for the AT Protocol network. Unlike the Firehose, it delivers records as plain JSON, filters server-side, and needs no CAR or DAG-CBOR decoding.

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.

Note

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

Both clients are present in two variants: sync and async. Filters are applied by the server, so you receive only what you asked for:

from atproto import JetstreamClient, models

client = JetstreamClient(params={'collections': [models.ids.AppBskyFeedPost], 'kinds': ['commit']})


def on_message_handler(event) -> None:
    if not isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Commit):
        return

    if event.operation == 'create':
        # already decoded into a model; a non-conforming record falls back to DotDict
        print(event.seq, event.record.text)


client.start(on_message_handler)

The record arrives as JSON, so no CAR or DAG-CBOR decoding is needed.

Filtering

The three filters are independent and combined with AND. Each matches everything when omitted:

  • kinds: commit, identity, account, sync.

  • dids: repositories to receive events for. Applies to every kind.

  • collections: NSIDs or <prefix>.* patterns.

Warning

collections constrains commit events only. Identity, account, and sync events are delivered regardless of it, because they are the only signals telling you an account was deactivated or deleted. Pass kinds=['commit'] to get a commits-only stream.

Cursor and reconnects

The cursor is tracked for you. Reconnects resume from the last delivered event, and events the server replays are dropped before reaching your callback, so you never see a gap or a duplicate.

Persist cursor to resume across restarts:

from atproto import JetstreamClient

client = JetstreamClient(params={'cursor': load_my_cursor()})


def on_message_handler(event) -> None:
    ...
    save_my_cursor(client.cursor)


client.start(on_message_handler)

Note

Cursors are instance-local and are not portable between servers or between Jetstream versions.

Compression

Frames are compressed by default using Jetstream’s dict-zstd scheme, which cuts bandwidth by roughly 60%. The client fetches the server’s dictionary over HTTPS once at startup, negotiates it on the websocket, and decompresses each frame transparently. Your callback sees the same models either way.

client = JetstreamClient()
print(client.compressed)  # False until the first connection negotiates it

Compression is best-effort and never fatal. If the dictionary cannot be fetched, or the server rotates it and the new one cannot be obtained, the client falls back to an uncompressed stream and keeps running. Check compressed to see what the current connection negotiated.

Pass compress=False to disable it:

client = JetstreamClient(compress=False)

Note

Decompression costs roughly 2 microseconds per frame, about 12% of the time spent turning a frame into a model.

Archive replay

Jetstream keeps the whole network’s history and can replay it. Pass an api_key and use snapshot for the sealed archive, or replay to sweep the archive and continue into the live tail without a gap:

from atproto import JetstreamClient

client = JetstreamClient(params={'dids': ['did:plc:...']}, api_key='...')

# the archive, then stop
for event in client.snapshot(after_seq=0):
    print(event.seq, event.did)

# the archive, then the live tail, seamlessly
for event in client.replay(after_seq=0):
    print(event.seq, event.did)

Both yield the same models the live tail delivers, so a consumer cannot tell whether an event came from a segment or the socket. The async client mirrors this with async for.

Record CIDs are not stored in the archive; the client derives each one from the record’s CBOR, matching what the PDS reports.

Note

Get a key at bsky.network/account. It is not an AT Protocol credential: a PDS session token and a com.atproto.server.getServiceAuth token are both rejected. The key is used only for the archive, never on the websocket, and a self-hosted Jetstream needs none.

Metering

Warning

The archive is metered in bytes downloaded, not requests. The whole network is roughly 1.85 TB. Check bytes_downloaded to see what a sweep cost.

What a filter costs depends on how selective it is, not on whether one is set. Every filter is sent to the planner, but segments carry per-DID bloom filters, so dids prunes hard while a popular collection appears in nearly every block and prunes almost nothing. Planning the whole archive:

filter

segments matched

blocks

dids=['did:plc:...']

1 of 7,075

1

collections=['app.bsky.feed.post']

7,075 of 7,075

5,363,406

To follow a busy collection, resume from a stored cursor rather than sweeping from after_seq=0.

The client honours the plan’s download mode, so a sparse filter fetches individual blocks rather than whole 261 MB segments, and whole segments are read in HTTP Range slices so they never land in memory at once. If the quota is exhausted the server replies 429 with Retry-After, and the client waits it out rather than retrying blindly.

More code examples: https://github.com/MarshalX/atproto/tree/main/examples/jetstream

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

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

Type:

int

property compressed: bool

Whether frames are being received compressed.

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

Type:

bool

property cursor: int | None

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.

Type:

int

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

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

Type:

int

property compressed: bool

Whether frames are being received compressed.

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

Type:

bool

property cursor: int | None

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.

Type:

int

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) → Commit | Identity | Account | Sync | 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