Reading

Fetching feeds, posts, threads and profiles. Almost everything on this page is an AppView query (app.bsky.*), so it runs against a Bluesky-style AppView rather than against a bare PDS.

Timelines and feeds

get_timeline returns the home timeline of the logged-in account.

examples/home_timeline.py
from atproto import Client, models


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

    print('Home (Following):\n')

    # Get "Home" page. Use pagination (cursor + limit) to fetch all posts
    timeline = client.get_timeline(algorithm='reverse-chronological')
    for feed_view in timeline.feed:
        action = 'New Post'
        if isinstance(feed_view.reason, models.AppBskyFeedDefs.ReasonRepost):
            action_by = feed_view.reason.by.handle
            action = f'Reposted by @{action_by}'

        post = feed_view.post.record
        author = feed_view.post.author

        if models.is_record_type(post, models.AppBskyFeedPost):
            print(f'[{action}] {author.display_name}: {post.text}')


if __name__ == '__main__':
    main()

Each entry is a FeedViewPost with three parts worth knowing:

post

The post itself. post.record is the app.bsky.feed.post record (so post.record.text), post.author is the profile that wrote it, and post.uri / post.cid are what you need to like, repost or quote it.

reason

Why the post is in the feed at all. A ReasonRepost, whose by is the reposter, when someone reposted it, a ReasonPin when it is pinned, and None when it is simply the author’s own post.

reply

Present when the post is a reply; holds the root and parent posts.

algorithm selects the feed. 'reverse-chronological' is the plain Following feed.

get_author_feed returns one account’s posts. actor is a handle or a DID.

examples/profile_posts.py
from atproto import Client, models


def main(client: Client, handle: str) -> None:
    print(f'\nProfile Posts of {handle}:\n\n')

    # Get profile's posts. Use pagination (cursor + limit) to fetch all
    profile_feed = client.get_author_feed(actor=handle)
    for feed_view in profile_feed.feed:
        post = feed_view.post.record
        if models.is_record_type(post, models.AppBskyFeedPost):
            print('-', post.text)


if __name__ == '__main__':
    at_client = Client()
    at_client.login('my-handle', 'my-password')

    while True:
        input_handle = input('\nPlease, enter the handle of the user: ')
        main(at_client, input_handle)

filter narrows what comes back and defaults to 'posts_with_replies'. The other accepted values are 'posts_no_replies', 'posts_with_media', 'posts_and_author_threads' and 'posts_with_video'. include_pins defaults to False.

Single posts

Three methods, and which one you want depends on what you are holding.

get_post reads the raw record straight out of a repository with com.atproto.repo.getRecord. It takes the record key, not a URI, and the second argument is the repository (handle or DID). Omit it and it reads from your own. The response is a GetRecordResponse, so the record is under .value. This is the only one of the three that does not need an AppView, but it also returns no view data: no like counts, no author profile, no embeds resolved.

get_posts takes a list of AT-URIs and returns hydrated PostViews under .posts: counts, author, viewer state and all. Use it whenever you have URIs.

get_post_thread takes one AT-URI and returns the post with its ancestors and replies. depth controls how many levels of replies come back and parent_height how many levels of parents.

thread = client.get_post_thread(uri=post.uri, depth=2)
print(thread.thread.post.record.text)
for reply in thread.thread.replies or []:
    print('-', reply.post.record.text)

Note

thread.thread is a union: a ThreadViewPost when the post is visible, but a NotFoundPost or a BlockedPost otherwise. Check py_type, or guard on the attribute you are about to touch, before assuming there is a .post there. See Working with models.

Profiles

get_profile takes one handle or DID and returns a ProfileViewDetailed directly, not wrapped in a response model. get_profiles takes a list and returns them under .profiles.

profile = client.get_profile('atproto.blue')
print(profile.display_name, profile.followers_count, profile.posts_count)

The profile of the logged-in account is already on the client as client.me after login. See Authentication for when it is None.

Who liked or reposted a post

get_likes and get_reposted_by both take the post’s uri, an optional cid, and cursor / limit.

likes = client.get_likes(uri=post.uri, cid=post.cid)
for like in likes.likes:
    print(like.actor.handle, like.created_at)

reposts = client.get_reposted_by(uri=post.uri, cid=post.cid)
for actor in reposts.reposted_by:
    print(actor.handle)

Note the asymmetry: get_likes gives you like records (actor plus created_at), while get_reposted_by gives you profiles directly.

Pagination

Every paged method works the same way. The response carries a cursor alongside the data, and you pass that cursor back on the next call to get the next page. When the server has nothing more to give, cursor comes back None (or missing).

paginate runs that loop for you. Give it a namespace method and its params. It returns a lazy iterator over the response models, and .items() iterates over the items of every page instead:

examples/paginate_follows.py
from atproto import Client, client_utils


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

    # Fetches page after page until the end. The same loop by hand: advanced_usage/handle_cursor_pagination.py
    follows = client_utils.paginate(client.app.bsky.graph.get_follows, {'actor': 'target-handle', 'limit': 100})
    for profile in follows.items():
        print(profile.handle)


if __name__ == '__main__':
    main()
for page in client_utils.paginate(client.app.bsky.feed.get_timeline, {'limit': 100}, max_pages=5):
    print(len(page.feed), page.cursor)

It works with every query that takes and returns a cursor: get_timeline, get_author_feed, get_likes, get_reposted_by, get_follows, get_followers, list_notifications, list_convos and the rest. The types follow the method. Pages are the method’s Response model, and items are whatever its list attribute holds (ProfileView for get_follows, FeedViewPost for get_timeline). A method that does not page is a type error. With an AsyncClient method you get an async iterator, used with async for.

Pass cursor in the params to start from a saved position. Each request starts after the last page fetched, and the paginator’s cursor attribute holds the next one. If a loop stops on an exception, such as a rate limit, iterating the same paginator again continues from there.

Warning

Requests happen as you iterate, one per page, and nothing stops a list(...) over a huge feed. Cap it with max_pages. limit is capped server-side, usually at 100, and a tight loop over thousands of pages will hit rate limits.

To write the loop yourself, break on a falsy cursor, not on an empty page. A page can come back with fewer items than limit, or with none at all, and still have a cursor pointing at more data. paginate also stops if the server repeats a cursor.

examples/advanced_usage/handle_cursor_pagination.py
from atproto import Client, models


def main() -> None:
    # This is an example for get_follows method. client_utils.paginate does the same: see paginate_follows.py
    client = Client()
    client.login('my-handle', 'my-password')
    handle = 'target-handle'

    cursor = None
    follows: list[models.AppBskyActorDefs.ProfileView] = []

    while True:
        fetched = client.get_follows(actor=handle, cursor=cursor)
        follows = follows + fetched.follows

        if not fetched.cursor:
            break

        cursor = fetched.cursor

    print(follows)


if __name__ == '__main__':
    main()

Resolving a bsky.app URL

A URL like https://bsky.app/profile/test.marshal.dev/post/3laqsdrwwgc24 is not an AT-URI. It holds a handle and a record key, and an AT-URI needs a DID. So there are two steps: split the URL, then resolve the handle to a DID with an IdResolver.

examples/advanced_usage/get_bsky_post_by_url.py
from typing import Optional

from atproto import Client, IdResolver, models


def fetch_posts(client: Client, resolver: IdResolver, url: str) -> Optional[models.AppBskyFeedPost.Record]:
    """Fetch a post using its Bluesky URL.

    Args:
        client (Client): Authenticated Atproto client.
        resolver (IdResolver): Resolver instance for DID lookup.
        url (str): URL of the Bluesky post.
    Returns:
        :obj:`models.AppBskyFeedPost.Record`: Post if found, otherwise None.
    """
    try:
        # Extract the handle and post rkey from the URL
        url_parts = url.split('/')
        handle = url_parts[4]  # Username in the URL
        post_rkey = url_parts[6]  # Post Record Key in the URL

        # Resolve the DID for the username
        did = resolver.handle.resolve(handle)
        if not did:
            print(f'Could not resolve DID for handle "{handle}".')
            return None

        # Fetch the post record
        return client.get_post(post_rkey, did).value
    except (ValueError, KeyError) as e:
        print(f'Error fetching post for URL {url}: {e}')
        return None


def main() -> None:
    # Initialize a client and authenticate
    client = Client()
    client.login('my-handle', 'my-password')

    # Initialize IdResolver for DID resolution
    resolver = IdResolver()

    # Define the URL of the post to fetch
    bsky_post_url = 'https://bsky.app/profile/test.marshal.dev/post/3laqsdrwwgc24'

    # Fetch the post
    post_record = fetch_posts(client, resolver, bsky_post_url)

    # Display the post details
    if post_record:
        print(f'Post content: {post_record.text}')
    else:
        print('Post could not be fetched.')


if __name__ == '__main__':
    main()

That example goes straight to get_post(rkey, did), which returns the raw record. If you want the hydrated view instead, build the URI and call get_posts:

did = IdResolver().handle.resolve(handle)
uri = f'at://{did}/app.bsky.feed.post/{rkey}'
post = client.get_posts([uri]).posts[0]
print(post.like_count, post.author.display_name)

Going the other way, AtUri parses an AT-URI back into its parts:

from atproto import AtUri

uri = AtUri.from_str('at://did:plc:.../app.bsky.feed.post/3laqsdrwwgc24')
print(uri.host, uri.collection, uri.rkey)

Tip

client.resolve_handle(handle) resolves a handle through your PDS instead, and returns a response with a .did. IdResolver does the resolution itself over DNS and well-known HTTP, and caches it, which is the better choice when you are resolving many handles.

See also

  • Posting: creating the records this page reads back.

  • Social graph: follows, followers and lists.

  • Firehose: streaming new records instead of polling for them.