Notifications

Likes, replies, follows and mentions arrive as notifications from the AppView. The SDK exposes the app.bsky.notification namespace; there is no higher-level wrapper on the client, so you call the namespace methods directly.

Listing notifications

list_notifications returns the most recent notifications for the logged-in account.

response = client.app.bsky.notification.list_notifications()
for notification in response.notifications:
    print(notification.reason, notification.author.handle, notification.is_read)

Each Notification carries:

reason: Why you were notified. See Reasons.
author: The ProfileView of whoever caused it.
uri and cid: The record that caused it: the like record, the reply post, the follow record.
reason_subject: The AT-URI of your record it happened to. On a like or reply, this is the post of yours that was liked or replied to. Absent for a follow, which has no subject.
record: The raw record behind uri, as an unknown type. See Working with models for how to narrow it.
is_read: Whether it is below the seen_at mark. See Marking as seen.
indexed_at: When the AppView indexed it. Notifications come back newest first.

The parameters: cursor and limit for paging (see Pagination), reasons to filter to a subset of reason strings, priority for the priority-only view, and seen_at to compute is_read against a time other than your stored mark.

Reasons

reason is an open string. The server can add new values, so treat anything you do not recognise as a notification you skip rather than a crash. The values the SDK currently knows are:

like: Someone liked your post.
repost: Someone reposted your post.
follow: Someone followed you.
mention: Someone mentioned you in a post, with a mention facet.
reply: Someone replied to your post.
quote: Someone quoted your post.
starterpack-joined: Someone signed up through your starter pack. The pack is on starter_pack.
verified / unverified: Your verification status changed.
like-via-repost / repost-via-repost: Someone liked or reposted your post from somebody else’s repost of it.
subscribed-post: An account you subscribed to activity notifications for has posted.
contact-match: A contact of yours joined.

The unread count

get_unread_count returns just a number, which is much cheaper than listing when all you want is a badge.

print(client.app.bsky.notification.get_unread_count().count)

It takes the same priority and seen_at parameters as list_notifications.

Marking as seen

update_seen takes a single seen_at timestamp and moves the server-side mark to it. Everything indexed before that point becomes is_read=True, and the unread count drops accordingly.

client.app.bsky.notification.update_seen({'seen_at': client.get_current_time_iso()})

get_current_time_iso gives you a correctly formatted UTC timestamp.

Warning

Take the timestamp before you fetch, not after you finish processing. Anything that arrives while your loop is running is indexed after the mark you captured, so it stays unread and you pick it up on the next pass. Stamping the time at the end silently drops those.

Polling

There is no push transport. The SDK has no websocket, no long-poll and no callback registration for notifications. You poll list_notifications on a timer, or you watch the firehose.

The pattern is: capture the time, fetch, act on everything with is_read=False, mark seen with the time you captured, sleep.

examples/process_notifications.py

from time import sleep

from atproto import Client

# how often we should check for new notifications
FETCH_NOTIFICATIONS_DELAY_SEC = 3


def main() -> None:
    client = Client()
    client.login('my-handle', 'my-password')

    # fetch new notifications
    while True:
        # save the time in UTC when we fetch notifications
        last_seen_at = client.get_current_time_iso()

        response = client.app.bsky.notification.list_notifications()
        for notification in response.notifications:
            if not notification.is_read:
                print(f'Got new notification! Type: {notification.reason}; from: {notification.author.did}')
                # example: "Got new notification! Type: like; from: did:plc:hlorqa2iqfooopmyzvb4byaz"

        # mark notifications as processed (isRead=True)
        client.app.bsky.notification.update_seen({'seen_at': last_seen_at})
        print('Successfully process notification. Last seen at:', last_seen_at)

        sleep(FETCH_NOTIFICATIONS_DELAY_SEC)


if __name__ == '__main__':
    main()

The async version does the same thing, but hands each notification to a callback and runs the callbacks concurrently with asyncio.gather:

examples/advanced_usage/notifications_callback.py

import asyncio
import typing as t

from atproto import AsyncClient, models

# how often we should check for new notifications
FETCH_NOTIFICATIONS_DELAY_SEC = 3.0

Notification = models.AppBskyNotificationListNotifications.Notification


async def main() -> None:
    async_client = AsyncClient()
    await async_client.login('my-handle', 'my-password')

    async def on_notification_callback(notification: Notification) -> None:
        print(f'Got new notification! Type: {notification.reason}; from: {notification.author.did}')
        # example: "Got new notification! Type: like; from: did:plc:hlorqa2iqfooopmyzvb4byaz"

    async def listen_for_notifications(
        on_notification: t.Callable[[Notification], t.Coroutine[t.Any, t.Any, None]],
    ) -> None:
        print('Start listening for notifications...')
        while True:
            # save the time in UTC when we fetch notifications
            last_seen_at = async_client.get_current_time_iso()

            # fetch new notifications
            response = await async_client.app.bsky.notification.list_notifications()

            # create a task list to run callbacks concurrently
            on_notification_tasks = []
            for notification in response.notifications:
                if not notification.is_read:
                    on_notification_tasks.append(on_notification(notification))

            # run callback on each notification
            await asyncio.gather(*on_notification_tasks)

            # mark notifications as processed (isRead=True)
            await async_client.app.bsky.notification.update_seen({'seen_at': last_seen_at})
            print('Successfully process notification. Last seen at:', last_seen_at)

            await asyncio.sleep(FETCH_NOTIFICATIONS_DELAY_SEC)

    # run our notification listener and register the callback on notification
    await asyncio.ensure_future(listen_for_notifications(on_notification_callback))


if __name__ == '__main__':
    # use run() for a higher Python version
    asyncio.get_event_loop().run_until_complete(main())

Tip

Three seconds is fine for one account. Polling is rate limited like any other request, so if you are running a bot across many accounts, back off. The current limits are at bsky.network/docs/rate-limits.

Neither example pages. list_notifications returns one page, so a backlog longer than the page size needs the cursor loop from Pagination. Otherwise update_seen marks notifications read that you never looked at.

The firehose instead

Polling costs you latency and a request per tick. If you want the events as they happen, or you are watching more than your own account, subscribe to the firehose and filter the record stream yourself: a like on your post is an app.bsky.feed.like commit whose subject.uri is in your repository.

That is more work, because you get every record on the network rather than a list addressed to you, with no is_read or seen mark to lean on. But it scales in a way polling does not. See Firehose.