CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-atproto

Comprehensive Python SDK for the AT Protocol, providing client interfaces, authentication, and real-time streaming for decentralized social networks.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

client-operations.mddocs/

Client Operations

High-level client interfaces providing synchronous and asynchronous access to AT Protocol services. The client classes handle authentication, session management, automatic JWT refresh, and provide access to all AT Protocol operations through a comprehensive set of generated methods.

Capabilities

Client Classes

Synchronous Client

The main synchronous client for AT Protocol operations with automatic session management and thread-safe JWT refresh.

class Client:
    """
    High-level synchronous client for XRPC and ATProto operations.
    
    Attributes:
        me (Optional[models.AppBskyActorDefs.ProfileViewDetailed]): Current user profile
    """
    def __init__(self, base_url: Optional[str] = None, *args, **kwargs):
        """
        Initialize the client.
        
        Args:
            base_url (str, optional): Custom base URL for AT Protocol server
        """
    
    def login(
        self, 
        login: Optional[str] = None,
        password: Optional[str] = None,
        session_string: Optional[str] = None,
        auth_factor_token: Optional[str] = None
    ) -> models.AppBskyActorDefs.ProfileViewDetailed:
        """
        Authenticate with AT Protocol server.
        
        Args:
            login (str, optional): Handle or email of the account
            password (str, optional): Main or app-specific password
            session_string (str, optional): Session string for re-authentication
            auth_factor_token (str, optional): Auth factor token for Email 2FA
            
        Note:
            Either session_string or login and password should be provided.
            
        Returns:
            AppBskyActorDefs.ProfileViewDetailed: Profile information
        """
    
    def send_post(
        self, 
        text: Union[str, 'client_utils.TextBuilder'],
        profile_identify: Optional[str] = None,
        reply_to: Optional['models.AppBskyFeedPost.ReplyRef'] = None,
        embed: Optional[Union[
            'models.AppBskyEmbedImages.Main',
            'models.AppBskyEmbedExternal.Main',
            'models.AppBskyEmbedRecord.Main',
            'models.AppBskyEmbedRecordWithMedia.Main',
            'models.AppBskyEmbedVideo.Main'
        ]] = None,
        langs: Optional[List[str]] = None,
        facets: Optional[List['models.AppBskyRichtextFacet.Main']] = None
    ) -> models.AppBskyFeedPost.CreateRecordResponse:
        """
        Create a post record.
        
        Args:
            text (str or TextBuilder): Post content
            profile_identify (str, optional): Handle or DID where to send post
            reply_to (AppBskyFeedPost.ReplyRef, optional): Root and parent of post to reply to
            embed (optional): Embed models attached to the post
            langs (List[str], optional): List of languages used in the post
            facets (List[AppBskyRichtextFacet.Main], optional): Rich text facets
            
        Returns:
            AppBskyFeedPost.CreateRecordResponse: Reference to created record
        """
    
    def get_timeline(self, **kwargs) -> models.AppBskyFeedGetTimeline.Response:
        """
        Get the authenticated user's timeline.
        
        Args:
            **kwargs: Timeline parameters (limit, cursor, etc.)
            
        Returns:
            AppBskyFeedGetTimeline.Response: Timeline data
        """
    
    def get_profile(self, actor: str) -> models.AppBskyActorGetProfile.Response:
        """
        Get profile information for an actor.
        
        Args:
            actor (str): Handle or DID of the actor
            
        Returns:
            AppBskyActorGetProfile.Response: Profile information
        """
    
    def delete_post(self, post_uri: str) -> bool:
        """
        Delete a post record.
        
        Args:
            post_uri (str): URI of the post to delete
            
        Returns:
            bool: True if deletion was successful
        """
    
    def send_image(self, text: str, image: bytes, image_alt: str, **kwargs) -> models.AppBskyFeedPost.CreateRecordResponse:
        """
        Send a post with an image.
        
        Args:
            text (str): Post text content
            image (bytes): Image data
            image_alt (str): Alt text for the image
            **kwargs: Additional post parameters
            
        Returns:
            AppBskyFeedPost.CreateRecordResponse: Created post record
        """
    
    def get_post(self, post_rkey: str, profile_identify: str, cid: Optional[str] = None) -> models.ComAtprotoRepoGetRecord.Response:
        """
        Get a single post record.
        
        Args:
            post_rkey (str): Record key of the post
            profile_identify (str): Handle or DID of the post author
            cid (str, optional): Specific version CID
            
        Returns:
            ComAtprotoRepoGetRecord.Response: Post record data
        """
    
    def like(self, uri: str, cid: str) -> models.AppBskyFeedLike.CreateRecordResponse:
        """
        Like a post.
        
        Args:
            uri (str): AT-URI of the post to like
            cid (str): CID of the post to like
            
        Returns:
            AppBskyFeedLike.CreateRecordResponse: Like record reference
        """
    
    def unlike(self, like_uri: str) -> bool:
        """
        Remove a like from a post.
        
        Args:
            like_uri (str): URI of the like record to delete
            
        Returns:
            bool: True if unlike was successful
        """
    
    def repost(self, uri: str, cid: str) -> models.AppBskyFeedRepost.CreateRecordResponse:
        """
        Repost a post.
        
        Args:
            uri (str): AT-URI of the post to repost
            cid (str): CID of the post to repost
            
        Returns:
            AppBskyFeedRepost.CreateRecordResponse: Repost record reference
        """
    
    def follow(self, subject: str) -> models.AppBskyGraphFollow.CreateRecordResponse:
        """
        Follow a user.
        
        Args:
            subject (str): Handle or DID of the user to follow
            
        Returns:
            AppBskyGraphFollow.CreateRecordResponse: Follow record reference
        """
    
    def unfollow(self, follow_uri: str) -> bool:
        """
        Unfollow a user.
        
        Args:
            follow_uri (str): URI of the follow record to delete
            
        Returns:
            bool: True if unfollow was successful
        """

Usage example:

from atproto import Client

# Initialize and authenticate
client = Client()
session = client.login('alice.bsky.social', 'password')

# Access user profile
print(f"Logged in as: {client.me.handle}")
print(f"DID: {client.me.did}")

# Create a post
response = client.send_post(text="Hello AT Protocol!")
print(f"Post created: {response.uri}")

# Get timeline
timeline = client.get_timeline(limit=10)
for item in timeline.feed:
    author = item.post.author
    text = item.post.record.text
    print(f"{author.handle}: {text}")

Asynchronous Client

The asynchronous version of the client with the same interface but async methods.

class AsyncClient:
    """
    High-level asynchronous client for XRPC and ATProto operations.
    
    Attributes:
        me (Optional[models.AppBskyActorDefs.ProfileViewDetailed]): Current user profile
    """
    def __init__(self, base_url: Optional[str] = None, *args, **kwargs):
        """
        Initialize the async client.
        
        Args:
            base_url (str, optional): Custom base URL for AT Protocol server
        """
    
    async def login(
        self, 
        login: Optional[str] = None,
        password: Optional[str] = None,
        session_string: Optional[str] = None,
        auth_factor_token: Optional[str] = None
    ) -> models.AppBskyActorDefs.ProfileViewDetailed:
        """
        Authenticate with AT Protocol server asynchronously.
        
        Args:
            login (str, optional): Handle or email of the account
            password (str, optional): Main or app-specific password
            session_string (str, optional): Session string for re-authentication
            auth_factor_token (str, optional): Auth factor token for Email 2FA
            
        Note:
            Either session_string or login and password should be provided.
            
        Returns:
            AppBskyActorDefs.ProfileViewDetailed: Profile information
        """
    
    async def send_post(
        self, 
        text: Union[str, 'client_utils.TextBuilder'],
        profile_identify: Optional[str] = None,
        reply_to: Optional['models.AppBskyFeedPost.ReplyRef'] = None,
        embed: Optional[Union[
            'models.AppBskyEmbedImages.Main',
            'models.AppBskyEmbedExternal.Main',
            'models.AppBskyEmbedRecord.Main',
            'models.AppBskyEmbedRecordWithMedia.Main',
            'models.AppBskyEmbedVideo.Main'
        ]] = None,
        langs: Optional[List[str]] = None,
        facets: Optional[List['models.AppBskyRichtextFacet.Main']] = None
    ) -> models.AppBskyFeedPost.CreateRecordResponse:
        """
        Create a post record asynchronously.
        
        Args:
            text (str or TextBuilder): Post content
            profile_identify (str, optional): Handle or DID where to send post
            reply_to (AppBskyFeedPost.ReplyRef, optional): Root and parent of post to reply to
            embed (optional): Embed models attached to the post
            langs (List[str], optional): List of languages used in the post
            facets (List[AppBskyRichtextFacet.Main], optional): Rich text facets
            
        Returns:
            AppBskyFeedPost.CreateRecordResponse: Reference to created record
        """
    
    async def delete_post(self, post_uri: str) -> bool:
        """Delete a post record asynchronously."""
    
    async def send_image(self, text: str, image: bytes, image_alt: str, **kwargs) -> models.AppBskyFeedPost.CreateRecordResponse:
        """Send a post with an image asynchronously."""
    
    async def like(self, uri: str, cid: str) -> models.AppBskyFeedLike.CreateRecordResponse:
        """Like a post asynchronously."""
    
    async def follow(self, subject: str) -> models.AppBskyGraphFollow.CreateRecordResponse:
        """Follow a user asynchronously."""
    
    async def close(self):
        """Close the async HTTP client connection."""

Usage example:

import asyncio
from atproto import AsyncClient

async def main():
    client = AsyncClient()
    
    # Authenticate
    await client.login('alice.bsky.social', 'password')
    
    # Create a post
    await client.send_post(text="Hello from async!")
    
    # Always close the client
    await client.close()

asyncio.run(main())

Session Management

Session Class

Represents an authenticated session with AT Protocol, including tokens and user information.

class Session:
    """
    Authenticated session with AT Protocol.
    
    Attributes:
        handle (str): User handle
        did (str): Decentralized identifier
        access_jwt (str): Access token for API calls
        refresh_jwt (str): Refresh token for renewing access
        pds_endpoint (Optional[str]): Personal Data Server endpoint
    """
    handle: str
    did: str
    access_jwt: str
    refresh_jwt: str
    pds_endpoint: Optional[str]
    
    def encode(self) -> str:
        """
        Serialize session to string for storage.
        
        Returns:
            str: Encoded session string
        """
    
    @classmethod
    def decode(cls, session_string: str) -> 'Session':
        """
        Deserialize session from string.
        
        Args:
            session_string (str): Encoded session string
            
        Returns:
            Session: Decoded session object
        """
    
    def copy(self) -> 'Session':
        """
        Create a copy of the session.
        
        Returns:
            Session: Copied session object
        """
    
    @property
    def access_jwt_payload(self) -> 'JwtPayload':
        """Get decoded access token payload."""
    
    @property
    def refresh_jwt_payload(self) -> 'JwtPayload':
        """Get decoded refresh token payload."""

Usage example:

from atproto import Client, Session

# Login and get session
client = Client()
session = client.login('alice.bsky.social', 'password')

# Save session for later use
session_string = session.encode()
# Store session_string securely...

# Restore session later
restored_session = Session.decode(session_string)
client = Client()
client.session = restored_session

Session Events

Events that occur during session lifecycle for monitoring and debugging.

class SessionEvent(Enum):
    """Events during session lifecycle."""
    IMPORT = 'import'    # Session imported from storage
    CREATE = 'create'    # New session created via login
    REFRESH = 'refresh'  # Session refreshed with new tokens

Request Handling

Synchronous Request Handler

Low-level HTTP request handler using HTTPX for synchronous operations.

class Request:
    """
    Synchronous HTTP request handler with HTTPX.
    """
    def get(self, *args, **kwargs) -> 'Response':
        """
        Make a GET request.
        
        Returns:
            Response: HTTP response wrapper
        """
    
    def post(self, *args, **kwargs) -> 'Response':
        """
        Make a POST request.
        
        Returns:
            Response: HTTP response wrapper
        """
    
    def set_additional_headers(self, headers: Dict[str, str]):
        """
        Set additional headers for all requests.
        
        Args:
            headers (Dict[str, str]): Headers to add
        """
    
    def clone(self) -> 'Request':
        """
        Create a cloned request handler.
        
        Returns:
            Request: Cloned request instance
        """

Asynchronous Request Handler

Low-level HTTP request handler for asynchronous operations.

class AsyncRequest:
    """
    Asynchronous HTTP request handler with HTTPX.
    """
    async def get(self, *args, **kwargs) -> 'Response':
        """
        Make an async GET request.
        
        Returns:
            Response: HTTP response wrapper
        """
    
    async def post(self, *args, **kwargs) -> 'Response':
        """
        Make an async POST request.
        
        Returns:
            Response: HTTP response wrapper
        """
    
    async def close(self):
        """Close the async HTTP client."""

Response Wrapper

HTTP response wrapper providing unified access to response data.

class Response:
    """
    HTTP response wrapper.
    
    Attributes:
        success (bool): Whether the request was successful
        status_code (int): HTTP status code
        content (Optional[Union[Dict[str, Any], bytes, XrpcError]]): Response content
        headers (Dict[str, Any]): Response headers
    """
    success: bool
    status_code: int
    content: Optional[Union[Dict[str, Any], bytes, XrpcError]]
    headers: Dict[str, Any]

Utilities

Text Builder

Helper for constructing rich text with facets (mentions, links, tags).

class TextBuilder:
    """
    Helper for constructing rich text with facets.
    """
    def __init__(self):
        """
        Initialize text builder.
        """
    
    def text(self, text: str) -> 'TextBuilder':
        """
        Add plain text to the builder.
        
        Args:
            text (str): Text to add
            
        Returns:
            TextBuilder: Self for chaining
        """
    
    def mention(self, text: str, did: str) -> 'TextBuilder':
        """
        Add a mention facet.
        
        Args:
            text (str): Text of the mention (e.g., '@alice')
            did (str): DID of the mentioned user
            
        Returns:
            TextBuilder: Self for chaining
        """
    
    def link(self, text: str, url: str) -> 'TextBuilder':
        """
        Add a link facet.
        
        Args:
            text (str): Link text
            url (str): Target URL
            
        Returns:
            TextBuilder: Self for chaining
        """
    
    def tag(self, text: str, tag: str) -> 'TextBuilder':
        """
        Add a hashtag facet.
        
        Args:
            text (str): Text of the tag (e.g., '#atproto')
            tag (str): Tag name (without #)
            
        Returns:
            TextBuilder: Self for chaining
        """
    
    def build_text(self) -> str:
        """
        Build the text from current state.
        
        Returns:
            str: Built text string
        """
    
    def build_facets(self) -> List['models.AppBskyRichtextFacet.Main']:
        """
        Build the facets from current state.
        
        Returns:
            List[models.AppBskyRichtextFacet.Main]: Built facets list
        """

Usage example:

from atproto import client_utils

# Build rich text with mentions and links
builder = client_utils.TextBuilder()
rich_text = (builder
    .text("Hello ")
    .mention("@alice", "did:plc:alice123")
    .text("! Check out ")
    .link("this article", "https://example.com")
    .text(" about ")
    .tag("#atproto", "atproto"))

# Use in a post (TextBuilder can be passed directly)
client.send_post(rich_text)

Exception Handling

class ModelError(Exception):
    """Base exception for model-related errors."""

class NetworkError(Exception):
    """Network-related request errors."""

class UnauthorizedError(Exception):
    """Authentication/authorization errors."""

class LoginRequiredError(Exception):
    """Thrown when login is required for an operation."""

Common error handling patterns:

from atproto import Client, LoginRequiredError, NetworkError

client = Client()

try:
    client.login('user@example.com', 'wrong-password')
except UnauthorizedError:
    print("Invalid credentials")

try:
    # This requires authentication
    client.send_post(text="Hello!")
except LoginRequiredError:
    print("Please log in first")
except NetworkError as e:
    print(f"Network error: {e}")

Install with Tessl CLI

npx tessl i tessl/pypi-atproto

docs

client-operations.md

core-functionality.md

cryptographic-operations.md

identity-resolution.md

index.md

jwt-operations.md

real-time-streaming.md

tile.json