Jetstream

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. The public service is documented at bsky.network/docs/jetstream.

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 FirehoseSubscribeReposClient when verifiability matters.

Subscribing

Both clients are present in two variants: sync and async. You write a callback; the client calls it for every event.

examples/jetstream/sub_events.py

from atproto import JetstreamClient, jetstream_models, models

client = JetstreamClient()


def on_message_handler(event: jetstream_models.SubscribeEventsMessage) -> None:
    if isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Info):
        # advisory about the stream itself; it carries no seq
        print('info:', event.name, event.message)
        return

    print(event.seq, event.did, type(event).__name__)


client.start(on_message_handler)

The record arrives as JSON, so no CAR or DAG-CBOR decoding is needed. A commit’s record is already a model; a record that does not conform to its lexicon falls back to DotDict.

Filtering

Filters are applied by the server, so you receive only what you asked for:

Sync

examples/jetstream/process_posts.py

from atproto import JetstreamClient, jetstream_models, models

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

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

    if event.operation != 'create':
        return

    # already decoded into a model by the client; a non-conforming record falls back to DotDict
    print(f'[{event.seq}] {event.did}: {event.record.text}')

client.start(on_message_handler)

Async

examples/jetstream/process_posts_async.py

import asyncio

from atproto import AsyncJetstreamClient, jetstream_models, models

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

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

    if event.operation != 'create':
        return

    # already decoded into a model by the client; a non-conforming record falls back to DotDict
    print(f'[{event.seq}] {event.did}: {event.record.text}')

asyncio.run(client.start(on_message_handler))

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:

examples/jetstream/resume_from_cursor.py

"""Persist the cursor so a restart resumes where the previous run stopped."""

import os
import typing as t
from pathlib import Path

from atproto import JetstreamClient, jetstream_models, models

_CURSOR_FILE = Path('jetstream.cursor')

#: Saving on every event would hammer the disk
_SAVE_EVERY = 100


def load_cursor() -> t.Optional[int]:
    try:
        return int(_CURSOR_FILE.read_text())
    except (OSError, ValueError):
        # missing or truncated by a crash; start from the live tip
        return None


def save_cursor(cursor: int) -> None:
    # write to a temporary file and rename, so a crash cannot leave a half-written cursor
    tmp_file = _CURSOR_FILE.with_suffix('.tmp')
    tmp_file.write_text(str(cursor))
    os.replace(tmp_file, _CURSOR_FILE)


params: models.NetworkBskyJetstreamSubscribeEvents.ParamsDict = {'kinds': ['commit']}

cursor = load_cursor()
if cursor is not None:
    params['cursor'] = cursor

client = JetstreamClient(params=params)
processed = 0


def on_message_handler(event: jetstream_models.SubscribeEventsMessage) -> None:
    global processed

    if isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Info):
        return

    print(event.seq, event.did)

    processed += 1
    if processed % _SAVE_EVERY == 0 and client.cursor is not None:
        save_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:

Sync

examples/jetstream/backfill_snapshot.py

"""Replay one repository's whole history out of the Jetstream archive.

Needs an API key from https://bsky.network/account. The archive is metered in bytes, so
filter narrowly: this plan matches a single block.
"""

import os

from atproto import JetstreamClient, models

TARGET_DID = 'did:plc:kvwvcn5iqfooopmyzvb4qzba'

client = JetstreamClient(params={'dids': [TARGET_DID]}, api_key=os.environ['JETSTREAM_API_KEY'])

posts = 0
for event in client.snapshot(after_seq=0):
    if not isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Commit):
        continue

    if event.collection != models.ids.AppBskyFeedPost or event.operation != 'create':
        continue

    posts += 1
    print(f'[{event.seq}] {event.time} {event.record.text[:60]}')

print(f'\n{posts} posts, {client.bytes_downloaded:,} bytes downloaded')

Async

examples/jetstream/backfill_async.py

"""Async archive replay.

Downloads run on the event loop and decoding is offloaded, so the loop stays responsive
while blocks are being decoded.
"""

import asyncio
import os

from atproto import AsyncJetstreamClient, models

TARGET_DID = 'did:plc:kvwvcn5iqfooopmyzvb4qzba'


async def main() -> None:
    client = AsyncJetstreamClient(params={'dids': [TARGET_DID]}, api_key=os.environ['JETSTREAM_API_KEY'])

    collections: dict = {}
    async for event in client.snapshot(after_seq=0):
        if isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Commit):
            collections[event.collection] = collections.get(event.collection, 0) + 1

    for collection, count in sorted(collections.items(), key=lambda item: -item[1]):
        print(f'{count:5} {collection}')

    print(f'\n{client.bytes_downloaded:,} bytes downloaded')


asyncio.run(main())

Or replay to sweep the archive and continue into the live tail without a gap:

examples/jetstream/backfill_then_live.py

"""Catch up on the archive, then keep streaming live, without a gap or a duplicate.

`replay()` sweeps the sealed archive first and cuts over to the live tail at the seam. It
never terminates.

The archive is metered in bytes, and how much 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: `dids` prunes hard, while a popular collection appears in nearly every block
and prunes almost nothing.

Measured against the whole archive, from `after_seq=0`:

    dids=[one repo]                    ->        1 of 7,075 segments,         1 block
    collections=[app.bsky.feed.post]   ->    7,075 of 7,075 segments, 5,363,406 blocks

So this example filters to one repository. Its whole history is a single block, about
274 KB. To follow a busy collection instead, resume from a stored cursor rather than
sweeping from the beginning.
"""

import os

from atproto import JetstreamClient, models

TARGET_DID = 'did:plc:kvwvcn5iqfooopmyzvb4qzba'

client = JetstreamClient(params={'dids': [TARGET_DID]}, api_key=os.environ['JETSTREAM_API_KEY'])

# archived events arrive first, then the stream continues live from the seam
for event in client.replay(after_seq=0):
    if not isinstance(event, models.NetworkBskyJetstreamSubscribeEvents.Commit):
        continue

    print(f'{event.seq} {event.collection} {event.operation}')

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