Client

Client is the entry point to the SDK. It wraps the XRPC endpoints of a PDS in methods that take and return models, and it keeps the session for you: log in once and every later call carries the credentials, refreshing them when they expire.

Sync and async

Every method exists in two clients with the same names, arguments and return types. Pick the one that matches the program you are writing:

Client runs each call to completion before returning. Use it in scripts, one-off tasks and anything already written in blocking style.
AsyncClient returns coroutines, so calls have to be awaited. Use it inside an event loop, and whenever you want many requests in flight at once.

Switching between them is an import, an await and nothing else:

from atproto import Client

client = Client()
profile = client.login('my-handle', 'my-password')
print('Welcome,', profile.display_name)

import asyncio

from atproto import AsyncClient

async def main() -> None:
    client = AsyncClient()
    profile = await client.login('my-handle', 'my-password')
    print('Welcome,', profile.display_name)

asyncio.run(main())

Both clients talk to https://bsky.social unless you pass another PDS as the first argument, and both expose the generated lexicon namespaces directly, so anything missing from the high-level methods is still reachable: client.app.bsky.feed.get_timeline(...).

Client

class atproto_client.client.client.Client(base_url: str | None = None, *args: Any, **kwargs: Any)

Bases: SessionDispatchMixin, SessionMethodsMixin, TimeMethodsMixin, HeadersConfigurationMethodsMixin, ClientRaw

High-level client for XRPC of ATProto.

me: t.Optional[models.AppBskyActorDefs.ProfileViewDetailed]: Profile of the logged-in account, when it was fetched at login.

Authorize a client and get profile info.

Parameters:

login – Handle/username of the account.
password – Main or app-specific password of the account.
session_string – Session string (use export_session_string() to get it).
auth_factor_token – Auth factor token (for Email 2FA).
fetch_bsky_profile – Look up the Bluesky profile of the account after authorizing.

Note

Either session_string or login and password should be provided.

Note

Authorization itself uses only com.atproto.server. The profile lookup is app.bsky.actor.getProfile, which not every PDS serves. A PDS that does not serve it does not fail the login: me is None and a warning is emitted. Any other failure of the lookup, such as an expired session, propagates. Pass fetch_bsky_profile=False on a PDS without app.bsky to skip the request and the warning.

Returns:: Profile information, or None when it was not fetched.
Return type:: models.AppBskyActorDefs.ProfileViewDetailed
Raises:: atproto.exceptions.AtProtocolError – Base exception.

Send post.

Note

If profile_identify is not provided will be sent to the current profile.

The default language is en. Available languages are defined in atproto.xrpc_client.models.languages.

Parameters:

text – Text of the post.
profile_identify – Handle or DID. Where to send post.
reply_to – Root and parent of the post to reply to.
embed – Embed models that should be attached to the post.
langs – List of used languages in the post.
facets – List of facets (rich text items).

Returns:

Reference to the created record.

Return type:

models.AppBskyFeedPost.CreateRecordResponse

Raises:

atproto.exceptions.AtProtocolError – Base exception.

delete_post(post_uri: str) → bool

Delete post.

Parameters:: post_uri – AT URI of the post.
Returns:: Success status.
Return type:: bool
Raises:: atproto.exceptions.AtProtocolError – Base exception.

Send post with multiple attached images (up to 4 images).

Note

If profile_identify is not provided will be sent to the current profile.

Parameters:

text – Text of the post.
images – List of binary images to attach. The length must be less than or equal to 4.
image_alts – List of text version of the images. The length should be shorter than or equal to the length of images.
profile_identify – Handle or DID. Where to send post.
reply_to – Root and parent of the post to reply to.
langs – List of used languages in the post.
facets – List of facets (rich text items).
image_aspect_ratios – List of aspect ratios of the images. The length should be shorter than or equal to the length of images.

Returns:

Reference to the created record.

Return type:

models.AppBskyFeedPost.CreateRecordResponse

Raises:

atproto.exceptions.AtProtocolError – Base exception.

Send post with attached image.

Note

If profile_identify is not provided will be sent to the current profile.

Parameters:

text – Text of the post.
image – Binary image to attach.
image_alt – Text version of the image.
profile_identify – Handle or DID. Where to send post.
reply_to – Root and parent of the post to reply to.
langs – List of used languages in the post.
facets – List of facets (rich text items).
image_aspect_ratio – Aspect ratio of the image.

Returns:

Reference to the created record.

Return type:

models.AppBskyFeedPost.CreateRecordResponse

Raises:

atproto.exceptions.AtProtocolError – Base exception.

Send post with attached video.

Note

If profile_identify is not provided will be sent to the current profile.

Parameters:

text – Text of the post.
video – Binary video to attach.
video_alt – Text version of the video.
profile_identify – Handle or DID. Where to send post.
reply_to – Root and parent of the post to reply to.
langs – List of used languages in the post.
facets – List of facets (rich text items).
video_aspect_ratio – Aspect ratio of the video.

Returns:

Reference to the created record.

Return type:

models.AppBskyFeedPost.CreateRecordResponse

Raises:

atproto.exceptions.AtProtocolError – Base exception.

get_post(post_rkey: str, profile_identify: str | None = None, cid: str | None = None) → GetRecordResponse

Get post.

Parameters:

post_rkey – ID (slug) of the post.
profile_identify – Handler or DID. Who created the post.
cid – The CID of the version of the post.

Returns:

Post.

Return type:

models.AppBskyFeedPost.GetRecordResponse

Raises:

atproto.exceptions.AtProtocolError – Base exception.

get_posts(uris: List[str]) → Response

Get posts.

Parameters:: uris – Uris (AT URI).

Example

client.get_posts(['at://did:plc:kvwvcn5iqfooopmyzvb4qzba/app.bsky.feed.post/3k2yihcrp6f2c'])

Returns:: Posts.
Return type:: models.AppBskyFeedGetPosts.Response
Raises:: atproto.exceptions.AtProtocolError – Base exception.

get_post_thread(uri: str, depth: int | None = None, parent_height: int | None = None) → Response

Get post thread.

Parameters:

uri – AT URI.
depth – Depth of the thread.
parent_height – Height of the parent post.

Returns:

Post thread.

Return type:

models.AppBskyFeedGetPostThread.Response

Raises:

atproto.exceptions.AtProtocolError – Base exception.

get_likes(uri: str, cid: str | None = None, cursor: str | None = None, limit: int | None = None) → Response

Get likes.

Parameters:

uri – AT URI.
cid – CID.
cursor – Cursor of the last like in the previous page.
limit – Limit count of likes to return.

Returns:

Likes.

Return type:

models.AppBskyFeedGetLikes.Response

Raises:

atproto.exceptions.AtProtocolError – Base exception.

get_reposted_by(uri: str, cid: str | None = None, cursor: str | None = None, limit: int | None = None) → Response

Get reposted by (reposts).

Parameters:

uri – AT URI.
cid – CID.
cursor – Cursor of the last like in the previous page.
limit – Limit count of likes to return.

Returns:

Reposts.

Return type:

models.AppBskyFeedGetRepostedBy.Response

Raises:

atproto.exceptions.AtProtocolError – Base exception.

get_timeline(algorithm: str | None = None, cursor: str | None = None, limit: int | None = None) → Response

Get home timeline.

Parameters:

algorithm – Algorithm.
cursor – Cursor of the last like in the previous page.
limit – Limit count of likes to return.

Returns:

Home timeline.

Return type:

models.AppBskyFeedGetTimeline.Response

Raises:

atproto.exceptions.AtProtocolError – Base exception.

get_author_feed(actor: str, cursor: str | None = None, filter: str | None = None, limit: int | None = None, include_pins: bool = False) → Response

Get author (profile) feed.

Parameters:

actor – Actor (handle or DID).
cursor – Cursor of the last like in the previous page.
filter – Filter.
limit – Limit count of likes to return.
include_pins – Include pins.

Returns:

Feed.

Return type:

models.AppBskyFeedGetAuthorFeed.Response

Raises:

atproto.exceptions.AtProtocolError – Base exception.

like(uri: str, cid: str) → CreateRecordResponse

Like the record.

Parameters:

cid – The CID of the record.
uri – The URI of the record.

Note

Record could be post, custom feed, etc.

Returns:: Reference to the created record.
Return type:: models.AppBskyFeedLike.CreateRecordResponse
Raises:: atproto.exceptions.AtProtocolError – Base exception.

unlike(like_uri: str) → bool

Unlike the post.

Parameters:: like_uri – AT URI of the like.
Returns:: Success status.
Return type:: bool
Raises:: atproto.exceptions.AtProtocolError – Base exception.

repost(uri: str, cid: str) → CreateRecordResponse

Repost post.

Parameters:

cid – The CID of the post.
uri – The URI of the post.

Returns:

Reference to the reposted record.

Return type:

models.AppBskyFeedRepost.CreateRecordResponse

Raises:

atproto.exceptions.AtProtocolError – Base exception.

unrepost(repost_uri: str) → bool

Unrepost the post (delete repost).

Parameters:: repost_uri – AT URI of the repost.
Returns:: Success status.
Return type:: bool
Raises:: atproto.exceptions.AtProtocolError – Base exception.

follow(subject: str) → CreateRecordResponse

Follow the profile.

Parameters:: subject – DID of the profile.
Returns:: Reference to the created record.
Return type:: models.AppBskyGraphFollow.CreateRecordResponse
Raises:: atproto.exceptions.AtProtocolError – Base exception.

unfollow(follow_uri: str) → bool

Unfollow the profile.

Parameters:: follow_uri – AT URI of the follow.
Returns:: Success status.
Return type:: bool
Raises:: atproto.exceptions.AtProtocolError – Base exception.

get_follows(actor: str, cursor: str | None = None, limit: int | None = None) → Response

Get follows of the profile.

Parameters:

actor – Actor (handle or DID).
cursor – Cursor of the next page.
limit – Limit count of follows to return.

Returns:

Follows.

Return type:

models.AppBskyGraphGetFollows.Response

Raises:

atproto.exceptions.AtProtocolError – Base exception.

get_followers(actor: str, cursor: str | None = None, limit: int | None = None) → Response

Get followers of the profile.

Parameters:

actor – Actor (handle or DID).
cursor – Cursor of the next page.
limit – Limit count of followers to return.

Returns:

Followers.

Return type:

models.AppBskyGraphGetFollowers.Response

Raises:

atproto.exceptions.AtProtocolError – Base exception.

get_profile(actor: str) → ProfileViewDetailed

Get profile.

Parameters:: actor – Actor (handle or DID).
Returns:: Profile.
Return type:: models.AppBskyActorDefs.ProfileViewDetailed
Raises:: atproto.exceptions.AtProtocolError – Base exception.

get_profiles(actors: List[str]) → Response

Get profiles.

Parameters:: actors – List of actors (handles or DIDs).
Returns:: Profiles.
Return type:: models.AppBskyActorGetProfiles.Response
Raises:: atproto.exceptions.AtProtocolError – Base exception.

mute(actor: str) → bool

Mute actor (profile).

Parameters:: actor – Actor (handle or DID).
Returns:: Success status.
Return type:: bool
Raises:: atproto.exceptions.AtProtocolError – Base exception.

unmute(actor: str) → bool

Unmute actor (profile).

Parameters:: actor – Actor (handle or DID).
Returns:: Success status.
Return type:: bool
Raises:: atproto.exceptions.AtProtocolError – Base exception.

resolve_handle(handle: str) → Response

Resolve the handle.

Parameters:: handle – Handle.
Returns:: Resolved handle (DID).
Return type:: models.ComAtprotoIdentityResolveHandle.Response
Raises:: atproto.exceptions.AtProtocolError – Base exception.

update_handle(handle: str) → bool

Update the handle.

Parameters:: handle – New handle.
Returns:: Success status.
Return type:: bool
Raises:: atproto.exceptions.AtProtocolError – Base exception.

upload_blob(data: bytes) → Response

Upload blob.

Parameters:: data – Binary data.
Returns:: Uploaded blob reference.
Return type:: models.ComAtprotoRepoUploadBlob.Response
Raises:: atproto.exceptions.AtProtocolError – Base exception.

delete_follow(follow_uri: str) → bool: Alias for unfollow

delete_like(like_uri: str) → bool: Alias for unlike

delete_repost(repost_uri: str) → bool: Alias for unrepost

post(text: str | TextBuilder, profile_identify: str | None = None, reply_to: ReplyRef | None = None, embed: Main | Main | Main | Main | Main | None = None, langs: List[str] | None = None, facets: List[Main] | None = None) → CreateRecordResponse: Alias for send_post

unsend(post_uri: str) → bool: Alias for delete_post

class AtprotoServiceType(*values)

Bases: Enum

The type of atproto service.

ATPROTO_LABELER = 'atproto_labeler'

BSKY_CHAT = 'bsky_chat'

BSKY_CHAT_DID: t.ClassVar[t.Literal['did:web:api.bsky.chat']] = 'did:web:api.bsky.chat'

BSKY_LABELER_DID: t.ClassVar[t.Literal['did:plc:ar7c4by46qjdydhdevvrndac']] = 'did:plc:ar7c4by46qjdydhdevvrndac'

clone() → Self

Clone the client instance.

Used to customize atproto proxy and set of labeler services.

Note

The clone shares the session and its dispatcher with the original, so it stays authenticated even when it is created before the first login.

Returns:: Cloned client instance.

configure_labelers_header(labeler_dids: List[str]) → None

Configure the atproto-labelers header to be applied on requests.

Parameters:: labeler_dids – The DIDs of the labelers.

configure_proxy_header(service_type: AtprotoServiceType | str, did: str) → None

Configure the atproto-proxy header to be applied on requests.

Parameters:

service_type – The type of service.
did – The DID of the proxy.

export_session_string() → str

Export session string.

Note

This method is useful for storing the session and reusing it later.

Warning

You should use it if you create the client instance often. Because of server rate limits for createSession. Rate limited by handle. 30/5 min, 300/day.

Attention

You must export session at the end of the Client`s life cycle! Alternatively, you can subscribe to the session change event. Use on_session_change() to register handler.

Example

>>> from atproto import Client
>>> # the first time login with login and password
>>> client = Client()
>>> client.login('login', 'password')
>>> session_string = client.export_session_string()
>>> # store session_string somewhere.
>>> # for example, in env and next time use it for login
>>> client2 = Client()
>>> client2.login(session_string=session_string)

Returns:: Session string.
Return type:: str

get_current_time() → datetime: Get current time in Server Timezone (UTC).

get_current_time_iso() → str: Get current time in Server Timezone (UTC) and ISO format.

get_time_from_timestamp(timestamp: int) → datetime: Get datetime from timestamp in Server Timezone (UTC).

invoke_procedure(nsid: str, params: ParamsModelBase | None = None, data: DataModelBase | bytes | None = None, **kwargs: Any) → Response

invoke_query(nsid: str, params: ParamsModelBase | None = None, data: DataModelBase | bytes | None = None, **kwargs: Any) → Response

on_session_change(callback: Callable[[SessionEvent, Session], None]) → None

Register a callback for session change event.

Parameters:: callback – A callback to be called when the session changes. The callback must accept two arguments: event and session.

Note

Possible events: SessionEvent.IMPORT, SessionEvent.CREATE, SessionEvent.REFRESH.

Tip

You should save the session string to persistent storage on SessionEvent.CREATE and SessionEvent.REFRESH event.

Example

>>> from atproto import Client, SessionEvent, Session
>>>
>>> client = Client()
>>>
>>> @client.on_session_change
>>> def on_session_change(event: SessionEvent, session: Session):
>>>     print(event, session)
>>>
>>> # or you can use this syntax:
>>> # client.on_session_change(on_session_change)

Returns:: None

property request: Request

update_base_url(base_url: str | None = None) → None

Update XRPC base URL.

Typically used for switching between PDSs.

Parameters:: base_url – New base URL. Defaults to bsky.social.

with_bsky_chat_proxy() → Self

Get a new client instance with the atproto-proxy header configured for bsky.chat.

Returns:: Configured client instance.
Return type:: self

with_bsky_labeler() → Self

Get a new client instance with the atproto-accept-labelers header configured for Bluesky Labeler.

Returns:: Configured client instance.
Return type:: self

with_labelers(labeler_dids: List[str]) → Self

Get a new client instance with the atproto-accept-labelers header configured.

Parameters:: labeler_dids – The DIDs of the labelers.
Returns:: Configured client instance.
Return type:: self

with_proxy(service_type: AtprotoServiceType | str, did: str) → Self

Get a new client instance with the atproto-proxy header configured.

Parameters:

service_type – The type of service.
did – The DID of the proxy.

Returns:

Configured client instance.

Return type:

self

app: sync_ns.AppNamespace

chat: sync_ns.ChatNamespace

com: sync_ns.ComNamespace

internal: sync_ns.InternalNamespace

network: sync_ns.NetworkNamespace

site: sync_ns.SiteNamespace

tools: sync_ns.ToolsNamespace

AsyncClient

The method list is not repeated here: every method above exists on AsyncClient under the same name, with the same arguments and the same return type, and returns a coroutine.

class atproto_client.client.async_client.AsyncClient(base_url: str | None = None, *args: Any, **kwargs: Any): High-level client for XRPC of ATProto.

Session

A session is what login returns you in exchange for credentials, and what every later call authenticates with. Export it with export_session_string, keep it somewhere durable, and hand it back to login instead of the password.

The client refreshes the session on its own, so a stored string goes stale. Subscribe to the change event and write the new one out each time:

from atproto import Client, Session, SessionEvent


client = Client()


@client.on_session_change
def on_session_change(event: SessionEvent, session: Session) -> None:
    if event in (SessionEvent.CREATE, SessionEvent.REFRESH):
        save_somewhere(session.export())

class atproto_client.client.session.Session(handle: str, did: str, access_jwt: str, refresh_jwt: str, pds_endpoint: str | None = 'https://bsky.social')

handle: str

did: str

access_jwt: str

refresh_jwt: str

pds_endpoint: str | None = 'https://bsky.social'

property access_jwt_payload: JwtPayload

property refresh_jwt_payload: JwtPayload

encode() → str

classmethod decode(session_string: str) → Session

copy() → Session

export() → str: Alias for encode

class atproto_client.client.session.SessionEvent(*values)

IMPORT = 'import'

CREATE = 'create'

REFRESH = 'refresh'

class atproto_client.client.session.SessionDispatcher(session: Session | None = None)

property session: Session | None: Session the dispatcher holds. Shared by every client cloned from the same original.

set_session(session: Session) → None

get_auth_headers() → Dict[str, str]: Build the Authorization header from the access token. Empty when there is no session.

get_refresh_auth_headers() → Dict[str, str]: Build the Authorization header from the refresh token. Empty when there is no session.

on_session_change(callback: Callable[[SessionEvent, Session], Coroutine[Any, Any, None]] | Callable[[SessionEvent, Session], None]) → None

dispatch_session_change(event: SessionEvent) → None

async dispatch_session_change_async(event: SessionEvent) → None