nauta.one
Async Python Jetstream Client
pypi.org/project/atproto_jetstream/
atproto
jetstream
python
zstd
1from asyncio import run
2
3from atproto_jetstream import Jetstream, JetstreamOptions
4
5
6async def main():
7 options = JetstreamOptions(
8 endpoint="wss://jetstream1.us-east.bsky.network/subscribe",
9 compress=True,
10 )
11 async with Jetstream(options) as stream:
12 async for event in stream:
13 match event.kind:
14 case "account":
15 print(event.account)
16 case "identity":
17 print(event.identity)
18 case "commit":
19 print(event.commit)
20
21
22if __name__ == "__main__":
23 run(main())