Advanced usage

Longer examples that go past the high-level client: session handling, pagination, the embed models, transport configuration, and error handling.

Reuse a session across runs

createSession is rate limited by handle. If your program starts and exits repeatedly, export the session string and log in with that instead of the password. See Authentication.

examples/advanced_usage/session_reuse.py
from typing import Optional

from atproto_client import Client, Session, SessionEvent


def get_session() -> Optional[str]:
    try:
        with open('session.txt', encoding='UTF-8') as f:
            return f.read()
    except FileNotFoundError:
        return None


def save_session(session_string: str) -> None:
    with open('session.txt', 'w', encoding='UTF-8') as f:
        f.write(session_string)


def on_session_change(event: SessionEvent, session: Session) -> None:
    print('Session changed:', event, repr(session))
    if event in (SessionEvent.CREATE, SessionEvent.REFRESH):
        print('Saving changed session')
        save_session(session.export())


def init_client() -> Client:
    client = Client()
    client.on_session_change(on_session_change)

    session_string = get_session()
    if session_string:
        print('Reusing session')
        client.login(session_string=session_string)
    else:
        print('Creating new session')
        client.login('username', 'password')

    return client


if __name__ == '__main__':
    client = init_client()
    # do something with the client
    print('Client is ready to use!')

Page through a cursor

There is no pagination helper. Responses carry a cursor you feed back in until the server stops returning one.

examples/advanced_usage/handle_cursor_pagination.py
from atproto import Client


def main() -> None:
    # This is an example for get_follows method.
    client = Client()
    client.login('my-handle', 'my-password')
    handle = 'target-handle'

    cursor = None
    follows = []

    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()

Rich text by hand

TextBuilder covers most cases. This is what it builds underneath, if you need the facets themselves.

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

# To send links as "link card" or "quote post" look at the send_embed.py example.
# There is a helper class TextBuilder
# that helps construct rich text: https://atproto.blue/en/latest/atproto_client/utils/text_builder.html


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

    text = 'example link'
    url = 'https://google.com'

    facets = [
        models.AppBskyRichtextFacet.Main(
            features=[models.AppBskyRichtextFacet.Link(uri=url)],
            # we should pass when our link starts and ends in the text
            # the example below selects all the text
            index=models.AppBskyRichtextFacet.ByteSlice(byte_start=0, byte_end=len(text.encode('UTF-8'))),
        )
    ]

    client.send_post(text=text, facets=facets)


if __name__ == '__main__':
    main()

Resolve a bsky.app URL to a post

A web URL carries a handle and a record key. Getting the record means resolving the handle to a DID first.

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()

Update your profile

The profile is a record like any other, at rkey self. Read it, change it, put it back. Do not create a new one, or you will drop the fields you did not set.

examples/advanced_usage/update_profile.py
import os

from atproto import Client, models
from atproto.exceptions import BadRequestError


def main() -> None:
    client = Client()
    client.login(os.environ['USERNAME'], os.environ['PASSWORD'])

    try:
        current_profile_record = client.app.bsky.actor.profile.get(client.me.did, 'self')
        current_profile = current_profile_record.value
        swap_record_cid = current_profile_record.cid
    except BadRequestError:
        current_profile = swap_record_cid = None

    old_description = old_display_name = None
    if current_profile:
        old_description = current_profile.description
        old_display_name = current_profile.display_name

    # set new values to update
    new_description = None
    new_display_name = None

    client.com.atproto.repo.put_record(
        models.ComAtprotoRepoPutRecord.Data(
            collection=models.ids.AppBskyActorProfile,
            repo=client.me.did,
            rkey='self',
            swap_record=swap_record_cid,
            record=models.AppBskyActorProfile.Record(
                avatar=current_profile.avatar,  # keep old avatar. to set a new one, you should upload blob first
                banner=current_profile.banner,  # keep old banner. to set a new one, you should upload blob first
                description=new_description or old_description,
                display_name=new_display_name or old_display_name,
            ),
        )
    )


if __name__ == '__main__':
    main()

Add someone to a list

examples/advanced_usage/add_user_to_list.py
from time import sleep

from atproto_client import Client, models
from atproto_core.uri import AtUri
from atproto_identity.resolver import IdResolver


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

    # https://bsky.app/profile/test.marshal.dev/lists/3k5z5k4k6qw2r
    mod_list_uri = 'at://did:plc:kvwvcn5iqfooopmyzvb4qzba/app.bsky.graph.list/3k5z5k4k6qw2r'
    user_handle_to_add = 'test.marshal.dev'

    mod_list_owner = AtUri.from_str(mod_list_uri).host
    user_to_add = IdResolver().handle.resolve(user_handle_to_add)

    print(f'Adding {user_to_add} to the list {mod_list_uri} (owned by {mod_list_owner})')

    created_list_item = client.app.bsky.graph.listitem.create(
        mod_list_owner,
        models.AppBskyGraphListitem.Record(
            list=mod_list_uri,
            subject=user_to_add,
            created_at=client.get_current_time_iso(),
        ),
    )

    print(f'Created list item: CID={created_list_item.cid}; URI={created_list_item.uri}')

    sleep(3)  # sleep for 3 sec because it takes some time to update the list for the backend

    mod_list = client.app.bsky.graph.get_list(models.AppBskyGraphGetList.Params(list=mod_list_uri))
    mod_list_users = [item.subject.did for item in mod_list.items]
    print(f'List users: {mod_list_users}')
    assert user_to_add in mod_list_users, f'User {user_to_add} not found in the list {mod_list_uri}'

    deleted_success = client.app.bsky.graph.listitem.delete(mod_list_owner, AtUri.from_str(created_list_item.uri).rkey)
    print(f'Deleted list item: {deleted_success}')


if __name__ == '__main__':
    main()

Poll for notifications

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())

Direct messages

Chat lives on a separate service, so the client has to be proxied to it. The app password needs the direct-messages grant. See Direct messages.

examples/advanced_usage/direct_messages.py
from atproto import Client, IdResolver, models

USERNAME = 'example.com'
PASSWORD = 'hunter2'  # noqa: S105 never hardcode your password in a real application


def main() -> None:
    # create client instance and login
    client = Client()
    client.login(USERNAME, PASSWORD)  # use App Password with access to Direct Messages!

    # create client proxied to Bluesky Chat service
    dm_client = client.with_bsky_chat_proxy()
    # create shortcut to convo methods
    dm = dm_client.chat.bsky.convo

    convo_list = dm.list_convos()  # use limit and cursor to paginate
    print(f'Your conversations ({len(convo_list.convos)}):')
    for convo in convo_list.convos:
        members = ', '.join(member.display_name for member in convo.members)
        print(f'- ID: {convo.id} ({members})')

    # create resolver instance with in-memory cache
    id_resolver = IdResolver()
    # resolve DID
    chat_to = id_resolver.handle.resolve('test.marshal.dev')

    # create or get conversation with chat_to
    convo = dm.get_convo_for_members(
        models.ChatBskyConvoGetConvoForMembers.Params(members=[chat_to]),
    ).convo

    print(f'\nConvo ID: {convo.id}')
    print('Convo members:')
    for member in convo.members:
        print(f'- {member.display_name} ({member.did})')

    # send a message to the conversation
    message = dm.send_message(
        models.ChatBskyConvoSendMessage.Data(
            convo_id=convo.id,
            message=models.ChatBskyConvoDefs.MessageInput(
                text='Hello from Python SDK!',
            ),
        )
    )

    # add a reaction to the message
    dm.add_reaction(
        models.ChatBskyConvoAddReaction.Data(
            convo_id=convo.id,
            message_id=message.id,
            value='👍',
        )
    )

    print('\nMessage sent!')


if __name__ == '__main__':
    main()

Proxies and labelers

examples/advanced_usage/proxy_and_labelers.py
from atproto import Client

USERNAME = 'example.com'
PASSWORD = 'hunter2'  # noqa: S105 never hardcode your password in a real application


def main() -> None:
    client = Client()
    client.login(USERNAME, PASSWORD)  # use App Password with access to Direct Messages!

    # `with_*` returns a configured clone; the original client keeps its own headers
    dm_client = client.with_bsky_chat_proxy()
    print('Proxy header:', dm_client.request.get_headers()['atproto-proxy'])
    print('Set on the original client:', 'atproto-proxy' in client.request.get_headers())

    # spelled out, this is what the convenience wrapper above does
    dm_client = client.with_proxy(Client.AtprotoServiceType.BSKY_CHAT, Client.BSKY_CHAT_DID)

    convos = dm_client.chat.bsky.convo.list_convos()
    print(f'You have {len(convos.convos)} conversations.')

    # ask the AppView to apply the labels of the Bluesky moderation service
    labeled_client = client.with_bsky_labeler()
    print('Labelers header:', labeled_client.request.get_headers()['atproto-accept-labelers'])

    # any set of labeler DIDs works
    labeled_client = client.with_labelers([Client.BSKY_LABELER_DID])

    profile = labeled_client.get_profile(USERNAME)
    for label in profile.labels or []:
        print(f'- {label.val} (from {label.src})')

    # clones share the session, so a token refresh on one is visible to all of them
    print('Same session:', client.export_session_string() == dm_client.export_session_string())


if __name__ == '__main__':
    main()

Configure the transport

Timeouts, retries, and anything else httpx exposes. See HTTP and transport.

examples/advanced_usage/custom_request.py
import httpx
from atproto import Client, Request, models

USERNAME = 'example.com'
PASSWORD = 'hunter2'  # noqa: S105 never hardcode your password in a real application


def main() -> None:
    # retry connection failures, and give slow uploads more than the default 5 seconds
    transport = httpx.HTTPTransport(retries=3)
    request = Request(timeout=httpx.Timeout(30.0), transport=transport)

    client = Client(base_url='https://bsky.social', request=request)
    client.login(USERNAME, PASSWORD)

    # low-level invoke: returns the raw Response dataclass instead of a parsed model
    response = client.invoke_query(
        'com.atproto.identity.resolveHandle',
        params=models.ComAtprotoIdentityResolveHandle.Params(handle='marshal.dev'),
        output_encoding='application/json',
    )

    print('Success:', response.success)
    print('Status code:', response.status_code)
    print('Content type:', response.headers.get('content-type'))
    print('Content:', response.content)

    # point the client at another PDS; "/xrpc" is appended for you
    client.update_base_url('https://pds.example.com')

    client.request.close()


if __name__ == '__main__':
    main()

Handle errors

See Errors and timeouts.

examples/advanced_usage/error_handling.py
from atproto import Client
from atproto.exceptions import (
    BadRequestError,
    InvokeTimeoutError,
    RateLimitExceededError,
    RequestErrorBase,
    UnauthorizedError,
)
from atproto_client.models.common import XrpcError

USERNAME = 'example.com'
PASSWORD = 'hunter2'  # noqa: S105 never hardcode your password in a real application


def describe(error: RequestErrorBase) -> str:
    """Summarize what the server said about a failed request."""
    content = error.response.content if error.response else None
    if isinstance(content, XrpcError):
        return f'{content.error}: {content.message}'

    # a non-JSON body (an HTML error page from a proxy, for example) arrives as raw bytes
    return repr(content)


def main() -> None:
    client = Client()

    try:
        client.login(USERNAME, PASSWORD)
    except UnauthorizedError as e:
        print('Login rejected:', describe(e))
        return
    except RateLimitExceededError as e:
        # createSession is rate limited by handle: 30/5 min, 300/day
        print('Too many logins. Retry at:', e.reset_at)
        return
    except InvokeTimeoutError:
        print('The PDS did not answer in time.')
        return

    try:
        client.com.atproto.identity.resolve_handle({'handle': 'not a handle'})
    except BadRequestError as e:
        print('Status code:', e.response.status_code)
        print('Server said:', describe(e))


if __name__ == '__main__':
    main()

Validate string formats

Handles, DIDs, NSIDs, AT-URIs and the rest are validated only when you opt in. See String formats.

examples/advanced_usage/validate_string_formats.py
from atproto_client.models import string_formats
from pydantic import TypeAdapter, ValidationError

some_good_handle = 'test.bsky.social'
some_bad_handle = 'invalid@ @handle'

strict_validation_context = {'strict_string_format': True}
HandleTypeAdapter = TypeAdapter(string_formats.Handle)

assert string_formats._OPT_IN_KEY == 'strict_string_format'

# values will not be validated if not opting in
sneaky_bad_handle = HandleTypeAdapter.validate_python(some_bad_handle)

assert sneaky_bad_handle == some_bad_handle

print(f'{sneaky_bad_handle=}\n\n')

# values will be validated if opting in
validated_good_handle = HandleTypeAdapter.validate_python(some_good_handle, context=strict_validation_context)

assert validated_good_handle == some_good_handle

print(f'{validated_good_handle=}\n\n')

try:
    print('Trying to validate a bad handle with strict validation...')
    HandleTypeAdapter.validate_python(some_bad_handle, context=strict_validation_context)
except ValidationError as e:
    print(e)