Firehose
The firehose is the network’s event stream. A relay aggregates every commit from every PDS it crawls and republishes them over one websocket, so you see each created, deleted, liked and reposted record as it happens. A second stream carries labels from moderation services.
The wire format is described in the event stream specification; the public relay is documented at bsky.network/docs/relay.
Subscribing to repository events
You write a callback; the client calls it for every message frame it receives. parse_subscribe_repos_message turns the frame into the model for its type.
from atproto import FirehoseSubscribeReposClient, firehose_models, parse_subscribe_repos_message
client = FirehoseSubscribeReposClient()
def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
client.start(on_message_handler)
import asyncio
from atproto import (
AsyncFirehoseSubscribeReposClient,
firehose_models,
parse_subscribe_repos_message,
)
async def main() -> None:
client = AsyncFirehoseSubscribeReposClient()
async def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
await client.start(on_message_handler)
if __name__ == '__main__':
# use run() for a higher Python version
asyncio.get_event_loop().run_until_complete(main())
start blocks until the client is stopped. The async client takes an asynchronous callback and is awaited.
FirehoseSubscribeReposClient defaults to wss://bsky.network/xrpc, the public relay. Pass base_uri to subscribe to a single PDS instead.
Subscribing to label events
Labels come from a moderation service rather than a relay, so FirehoseSubscribeLabelsClient defaults to wss://mod.bsky.app/xrpc, the Bluesky moderation service. Everything else is the same.
from atproto import FirehoseSubscribeLabelsClient, firehose_models, models, parse_subscribe_labels_message
client = FirehoseSubscribeLabelsClient()
def on_message_handler(message: firehose_models.MessageFrame) -> None:
labels_message = parse_subscribe_labels_message(message)
if not isinstance(labels_message, models.ComAtprotoLabelSubscribeLabels.Labels):
return
for label in labels_message.labels:
neg = '(NEG)' if label.neg else ''
print(f'[{label.cts}] ({label.src}) {label.uri} => {label.val} {neg}')
client.start(on_message_handler)
Decoding commits
parse_subscribe_repos_message does not decode the inner DAG-CBOR. A ComAtprotoSyncSubscribeRepos.Commit carries its records as a CAR file in blocks, and the ops list tells you which CID in that file belongs to which path. Decode it with CAR:
from atproto import CAR, models
def on_message_handler(message) -> None:
commit = parse_subscribe_repos_message(message)
# we need to be sure that it's a commit message with .blocks inside
if not isinstance(commit, models.ComAtprotoSyncSubscribeRepos.Commit):
return
if not commit.blocks:
return
car = CAR.from_bytes(commit.blocks)
car.blocks maps CID to raw record data; car.root is the commit object itself. The full version, resolving each op to an AtUri then looking the record up in the CAR and turning it into a model, is worked through in the examples below, which also fan the work out to a process pool because a single Python process cannot keep up with the relay at peak:
import multiprocessing
import signal
import sys
import time
from collections import defaultdict
from types import FrameType
from typing import Any
from atproto import CAR, AtUri, FirehoseSubscribeReposClient, firehose_models, models, parse_subscribe_repos_message
_INTERESTED_RECORDS = {
models.ids.AppBskyFeedLike: models.AppBskyFeedLike,
models.ids.AppBskyFeedPost: models.AppBskyFeedPost,
models.ids.AppBskyGraphFollow: models.AppBskyGraphFollow,
}
def _get_ops_by_type(commit: models.ComAtprotoSyncSubscribeRepos.Commit) -> defaultdict:
operation_by_type = defaultdict(lambda: {'created': [], 'deleted': []})
car = CAR.from_bytes(commit.blocks)
for op in commit.ops:
if op.action == 'update':
# not supported yet
continue
uri = AtUri.from_str(f'at://{commit.repo}/{op.path}')
if op.action == 'create':
if not op.cid:
continue
create_info = {'uri': str(uri), 'cid': str(op.cid), 'author': commit.repo}
record_raw_data = car.blocks.get(op.cid)
if not record_raw_data:
continue
record = models.get_or_create(record_raw_data, strict=False)
record_type = _INTERESTED_RECORDS.get(uri.collection)
if record_type and models.is_record_type(record, record_type):
operation_by_type[uri.collection]['created'].append({'record': record, **create_info})
if op.action == 'delete':
operation_by_type[uri.collection]['deleted'].append({'uri': str(uri)})
return operation_by_type
def worker_main(cursor_value: multiprocessing.Value, pool_queue: multiprocessing.Queue) -> None:
signal.signal(signal.SIGINT, signal.SIG_IGN) # we handle it in the main process
while True:
message = pool_queue.get()
commit = parse_subscribe_repos_message(message)
if not isinstance(commit, models.ComAtprotoSyncSubscribeRepos.Commit):
continue
if commit.seq % 20 == 0:
cursor_value.value = commit.seq
if not commit.blocks:
continue
ops = _get_ops_by_type(commit)
for created_post in ops[models.ids.AppBskyFeedPost]['created']:
author = created_post['author']
record = created_post['record']
inlined_text = record.text.replace('\n', ' ')
print(f'NEW POST [CREATED_AT={record.created_at}][AUTHOR={author}]: {inlined_text}')
def get_firehose_params(cursor_value: multiprocessing.Value) -> models.ComAtprotoSyncSubscribeRepos.Params:
return models.ComAtprotoSyncSubscribeRepos.Params(cursor=cursor_value.value)
def measure_events_per_second(func: callable) -> callable:
def wrapper(*args) -> Any:
wrapper.calls += 1
cur_time = time.time()
if cur_time - wrapper.start_time >= 1:
print(f'NETWORK LOAD: {wrapper.calls} events/second')
wrapper.start_time = cur_time
wrapper.calls = 0
return func(*args)
wrapper.calls = 0
wrapper.start_time = time.time()
return wrapper
def signal_handler(_: int, __: FrameType) -> None:
print('Keyboard interrupt received. Waiting for the queue to empty before terminating processes...')
# Stop receiving new messages
client.stop()
# Drain the messages queue
while not queue.empty():
print('Waiting for the queue to empty...')
time.sleep(0.2)
print('Queue is empty. Gracefully terminating processes...')
pool.terminate()
pool.join()
sys.exit(0)
if __name__ == '__main__':
signal.signal(signal.SIGINT, signal_handler)
start_cursor = None
params = None
cursor = multiprocessing.Value('i', 0)
if start_cursor is not None:
cursor = multiprocessing.Value('i', start_cursor)
params = get_firehose_params(cursor)
client = FirehoseSubscribeReposClient(params)
workers_count = multiprocessing.cpu_count() * 2 - 1
max_queue_size = 10000
queue = multiprocessing.Queue(maxsize=max_queue_size)
pool = multiprocessing.Pool(workers_count, worker_main, (cursor, queue))
@measure_events_per_second
def on_message_handler(message: firehose_models.MessageFrame) -> None:
if cursor.value:
# we are using updating the cursor state here because of multiprocessing
# typically you can call client.update_params() directly on commit processing
client.update_params(get_firehose_params(cursor))
queue.put(message)
client.start(on_message_handler)
import asyncio
import signal
import time
from collections import defaultdict
from types import FrameType
from typing import Any
from atproto import (
CAR,
AsyncFirehoseSubscribeReposClient,
AtUri,
firehose_models,
models,
parse_subscribe_repos_message,
)
_INTERESTED_RECORDS = {
models.ids.AppBskyFeedLike: models.AppBskyFeedLike,
models.ids.AppBskyFeedPost: models.AppBskyFeedPost,
models.ids.AppBskyGraphFollow: models.AppBskyGraphFollow,
}
def _get_ops_by_type(commit: models.ComAtprotoSyncSubscribeRepos.Commit) -> defaultdict:
operation_by_type = defaultdict(lambda: {'created': [], 'deleted': []})
car = CAR.from_bytes(commit.blocks)
for op in commit.ops:
if op.action == 'update':
# not supported yet
continue
uri = AtUri.from_str(f'at://{commit.repo}/{op.path}')
if op.action == 'create':
if not op.cid:
continue
create_info = {'uri': str(uri), 'cid': str(op.cid), 'author': commit.repo}
record_raw_data = car.blocks.get(op.cid)
if not record_raw_data:
continue
record = models.get_or_create(record_raw_data, strict=False)
record_type = _INTERESTED_RECORDS.get(uri.collection)
if record_type and models.is_record_type(record, record_type):
operation_by_type[uri.collection]['created'].append({'record': record, **create_info})
if op.action == 'delete':
operation_by_type[uri.collection]['deleted'].append({'uri': str(uri)})
return operation_by_type
def measure_events_per_second(func: callable) -> callable:
def wrapper(*args) -> Any:
wrapper.calls += 1
cur_time = time.time()
if cur_time - wrapper.start_time >= 1:
print(f'NETWORK LOAD: {wrapper.calls} events/second')
wrapper.start_time = cur_time
wrapper.calls = 0
return func(*args)
wrapper.calls = 0
wrapper.start_time = time.time()
return wrapper
async def signal_handler(_: int, __: FrameType) -> None:
print('Keyboard interrupt received. Stopping...')
# Stop receiving new messages
await client.stop()
async def main(firehose_client: AsyncFirehoseSubscribeReposClient) -> None:
@measure_events_per_second
async def on_message_handler(message: firehose_models.MessageFrame) -> None:
commit = parse_subscribe_repos_message(message)
if not isinstance(commit, models.ComAtprotoSyncSubscribeRepos.Commit):
return
if commit.seq % 20 == 0:
firehose_client.update_params(models.ComAtprotoSyncSubscribeRepos.Params(cursor=commit.seq))
if not commit.blocks:
return
ops = _get_ops_by_type(commit)
for created_post in ops[models.ids.AppBskyFeedPost]['created']:
author = created_post['author']
record = created_post['record']
inlined_text = record.text.replace('\n', ' ')
print(f'NEW POST [CREATED_AT={record.created_at}][AUTHOR={author}]: {inlined_text}')
await client.start(on_message_handler)
if __name__ == '__main__':
signal.signal(signal.SIGINT, lambda _, __: asyncio.create_task(signal_handler(_, __)))
start_cursor = None
params = None
if start_cursor is not None:
params = models.ComAtprotoSyncSubscribeRepos.Params(cursor=start_cursor)
client = AsyncFirehoseSubscribeReposClient(params)
# use run() for a higher Python version
asyncio.get_event_loop().run_until_complete(main(client))
Tip
Your callback runs on the receive loop. Anything slow in it, a database write or an HTTP call, is backpressure on the socket, and the relay disconnects consumers that fall too far behind. Hand the frame to a queue and do the work elsewhere.
Filter before you parse
Parsing is the expensive part. The frame header is already decoded when your callback is called, so discard the messages you do not want first:
def on_message_handler(message: firehose_models.MessageFrame) -> None:
if message.type != '#commit':
return
commit = parse_subscribe_repos_message(message)
message.type is the t field of message.header, one of #commit, #sync, #identity, #account or #info on the repos stream.
Note
parse_subscribe_repos_message raises KeyError on a message type it does not know, which is what a newly added event type looks like to an older SDK. Filtering on the header first also protects you from that.
Cursors
Every commit carries a seq. Pass one back as the cursor param and the relay replays everything after it, which is how you avoid a gap across a restart.
The cursor you started with is the cursor the client reconnects with. It is not advanced for you. Call update_params as you process messages so that a reconnect resumes from where you actually are:
def on_message_handler(message: firehose_models.MessageFrame) -> None:
commit = parse_subscribe_repos_message(message)
if not isinstance(commit, models.ComAtprotoSyncSubscribeRepos.Commit):
return
if commit.seq % 20 == 0:
client.update_params(models.ComAtprotoSyncSubscribeRepos.Params(cursor=commit.seq))
Warning
If you pass params when you construct the client and never update them, a reconnect rolls the stream back to the cursor you started with and you reprocess everything since.
update_params accepts a params model or a plain dict, and takes effect on the next connection, not the current one. Persisting the cursor every message is usually wasteful; every N messages, as above, bounds how much you replay after a crash.
Reconnects and recv_timeout
The client reconnects on its own. Connection errors (a dropped socket, a failed handshake, an oversized frame) are not raised to you; the client backs off and dials again. The delay doubles per attempt up to 64 seconds, with a small random offset so a fleet of consumers does not return in lockstep. A connection that stayed up for at least a minute resets the backoff to its base delay rather than reconnecting instantly, for the same reason: when a relay restarts it drops every consumer at once.
recv_timeout is how long the client waits for a frame before deciding the connection is dead and reconnecting. It defaults to 30 seconds for the repos stream and 5 minutes for labels, which is idle time rather than total time, because the labels stream is quiet for long stretches and needs the longer window. Raise it if you subscribe to something quieter still; None disables the timeout, which means a silently half-open connection is never noticed.
Two things do stop the client. A server-sent error frame is raised as SubscriptionError, and a clean close by the server ends start without an exception. Frames that fail to decode are neither: one bad frame is skipped rather than killing the connection.
Stopping the client
stop is safe to call from another thread or task, and takes effect even while the client is idle waiting for the next frame.
import threading
import time
from atproto import FirehoseSubscribeReposClient, firehose_models, parse_subscribe_repos_message
_STOP_AFTER_SECONDS = 3
client = FirehoseSubscribeReposClient()
def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
def _stop_after_n_sec() -> None:
time.sleep(_STOP_AFTER_SECONDS)
client.stop()
# run our sleep functions in another thread
threading.Thread(target=_stop_after_n_sec).start()
# run the client for N seconds
client.start(on_message_handler)
print(f'Successfully stopped after {_STOP_AFTER_SECONDS} seconds!')
import asyncio
from atproto import (
AsyncFirehoseSubscribeReposClient,
firehose_models,
parse_subscribe_repos_message,
)
_STOP_AFTER_SECONDS = 3
async def main() -> None:
client = AsyncFirehoseSubscribeReposClient()
async def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
async def _stop_after_n_sec() -> None:
await asyncio.sleep(_STOP_AFTER_SECONDS)
await client.stop()
# save ref to task to eliminate problems with GC
_stop_after_n_sec_task = asyncio.create_task(_stop_after_n_sec())
await client.start(on_message_handler)
await _stop_after_n_sec_task
print(f'Successfully stopped after {_STOP_AFTER_SECONDS} seconds!')
if __name__ == '__main__':
# use run() for a higher Python version
asyncio.get_event_loop().run_until_complete(main())
Errors in your callback
An exception raised by your callback does not stop the stream. Without a second callback the traceback is printed and the next message is processed; pass one and it is called with the exception instead.
from atproto import FirehoseSubscribeReposClient, firehose_models, parse_subscribe_repos_message
client = FirehoseSubscribeReposClient()
def on_message_handler(message: firehose_models.MessageFrame) -> None:
print(message.header, parse_subscribe_repos_message(message))
raise ValueError('Failed to process message')
def on_callback_error_handler(error: BaseException) -> None:
print('Got error!', error)
client.start(on_message_handler, on_callback_error_handler)
The error callback is called on the receive loop too, and an exception raised inside it is printed and swallowed.
Which client to use
FirehoseSubscribeReposClient and FirehoseSubscribeLabelsClient are thin subclasses of the generated subscription clients that fill in the deployment details: the relay and moderation service hostnames, and the recv_timeout appropriate to each stream. The generated clients underneath take base_uri as their first, required argument and default recv_timeout to None:
Client |
Subscription |
Parser |
|---|---|---|
|
|
|
|
|
|
|
|
|
Each has an Async counterpart with the same name and arguments. They live in atproto_client.subscriptions and are generated from the lexicons, so a subscription added to the protocol gets a client without anything being written by hand.
ChatBskyModerationSubscribeModEventsClient has no Firehose* alias because it has no public host: it is the moderation event stream of a chat service, and you point it at your own.
from atproto_client.subscriptions import (
ChatBskyModerationSubscribeModEventsClient,
parse_chat_bsky_moderation_subscribe_mod_events_message,
)
client = ChatBskyModerationSubscribeModEventsClient('wss://my-chat-service.example.com/xrpc')
def on_message_handler(message) -> None:
print(parse_chat_bsky_moderation_subscribe_mod_events_message(message))
client.start(on_message_handler)
All of them are SubscriptionClients, which is where start, stop, update_params and the reconnect behaviour described above actually live.
Firehose or Jetstream?
Jetstream carries the same events as JSON, filtered server-side, with no CAR or DAG-CBOR decoding to do. It is dramatically cheaper to consume, and it is the right default for most consumers.
The firehose is what you want when you need what Jetstream drops. Jetstream events carry no repository signatures and no MST proofs, so they cannot be cryptographically verified; firehose commits can. The firehose also gives you the raw blocks, which is what you need if you are mirroring repositories rather than reacting to records.
More code examples: https://github.com/MarshalX/atproto/tree/main/examples/firehose