or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

client-operations.mdcore-functionality.mdcryptographic-operations.mdidentity-resolution.mdindex.mdjwt-operations.mdreal-time-streaming.md
tile.json

tessl/pypi-atproto

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/atproto@0.0.x

To install, run

npx @tessl/cli install tessl/pypi-atproto@0.0.0

index.mddocs/

ATProto Python SDK

A comprehensive Python SDK for the AT Protocol, providing client interfaces, authentication, cryptographic operations, and real-time streaming capabilities for building decentralized social network applications. The SDK offers both synchronous and asynchronous APIs with automatic session management, extensive type hints, and robust error handling.

Package Information

  • Package Name: atproto
  • Language: Python
  • Installation: pip install atproto
  • Documentation: https://atproto.blue

Core Imports

import atproto

Common patterns for client operations:

from atproto import Client, AsyncClient, models, Session

Specific functionality imports:

from atproto import (
    # Client and session management
    Client, AsyncClient, Session, SessionEvent,
    # Core AT Protocol types
    CID, CAR, AtUri, NSID, DidDocument,
    # Cryptographic operations
    verify_signature, get_did_key, Multikey,
    # Real-time streaming
    FirehoseSubscribeReposClient, AsyncFirehoseSubscribeReposClient,
    # Identity resolution
    IdResolver, AsyncIdResolver, AtprotoData,
    # JWT operations
    verify_jwt, parse_jwt, JwtPayload
)

Basic Usage

import atproto

# Create a client and log in
client = atproto.Client()
client.login('your-handle.bsky.social', 'your-password')

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

# Create a post
text = "Hello from the ATProto Python SDK!"
client.send_post(text=text)

# Get the user's timeline
timeline = client.get_timeline()
for post in timeline.feed:
    print(f"{post.post.author.handle}: {post.post.record.text}")

Async usage:

import asyncio
import atproto

async def main():
    client = atproto.AsyncClient()
    await client.login('your-handle.bsky.social', 'your-password')
    
    # Create a post asynchronously
    await client.send_post(text="Hello from async ATProto!")
    
    # Close the client
    await client.close()

asyncio.run(main())

Architecture

The ATProto SDK is organized into six main functional areas:

  • Client Operations: High-level client interfaces for XRPC communication with automatic session management
  • Core Functionality: Fundamental AT Protocol data structures and utilities for content addressing and URI handling
  • Cryptographic Operations: Key management, signature verification, and multibase encoding for secure communications
  • Real-time Streaming: Firehose clients for consuming live AT Protocol data streams
  • Identity Resolution: DID document resolution and caching for decentralized identity management
  • JWT Operations: JSON Web Token parsing, validation, and verification for authentication

Capabilities

Client Operations

High-level client interfaces providing synchronous and asynchronous access to AT Protocol services. Includes automatic session management, JWT refresh, and comprehensive model support for all AT Protocol operations.

class Client:
    def __init__(self, base_url: Optional[str] = None, *args, **kwargs): ...
    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: ...
    def send_post(self, text: Union[str, 'client_utils.TextBuilder'], **kwargs) -> models.AppBskyFeedPost.CreateRecordResponse: ...

class AsyncClient:
    def __init__(self, base_url: Optional[str] = None, *args, **kwargs): ...
    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: ...
    async def send_post(self, text: Union[str, 'client_utils.TextBuilder'], **kwargs) -> models.AppBskyFeedPost.CreateRecordResponse: ...

class Session:
    handle: str
    did: str
    access_jwt: str
    refresh_jwt: str
    pds_endpoint: Optional[str]

Client Operations

Core Functionality

Fundamental AT Protocol data structures including Content Identifiers (CID), Content Addressable Archives (CAR), AT Protocol URIs, Namespaced IDs (NSID), and DID document handling.

class CID:
    version: int
    codec: int
    hash: Multihash
    @classmethod
    def decode(cls, value: Union[str, bytes]) -> 'CID': ...

class CAR:
    root: CID
    blocks: Dict[CID, bytes]
    @classmethod
    def from_bytes(cls, data: bytes) -> 'CAR': ...

class AtUri:
    def __init__(self, host: str, pathname: str = '', hash_: str = '', search_params: Optional[List[Tuple[str, Any]]] = None): ...

class NSID:
    segments: List[str]
    @classmethod
    def from_str(cls, nsid: str) -> 'NSID': ...

Core Functionality

Cryptographic Operations

Cryptographic key management, signature verification, multibase encoding/decoding, and DID key generation for secure AT Protocol communications.

class Multikey:
    jwt_alg: str
    key_bytes: bytes
    @classmethod
    def from_str(cls, multikey: str) -> 'Multikey': ...

def get_did_key(key: Any) -> str: ...
def verify_signature(did_key: str, signing_input: bytes, signature: bytes) -> bool: ...
def bytes_to_multibase(encoding: str, data: bytes) -> str: ...
def multibase_to_bytes(data: str) -> bytes: ...

Cryptographic Operations

Real-time Streaming

Firehose clients for consuming live AT Protocol data streams including repository updates and labeling events. Provides both synchronous and asynchronous streaming capabilities.

class FirehoseSubscribeReposClient:
    def __init__(self, base_url: str = 'wss://bsky.network', *args, **kwargs): ...

class AsyncFirehoseSubscribeReposClient:
    def __init__(self, base_url: str = 'wss://bsky.network', *args, **kwargs): ...

def parse_subscribe_repos_message(message: MessageFrame) -> SubscribeReposMessage: ...
def parse_subscribe_labels_message(message: MessageFrame) -> SubscribeLabelsMessage: ...

Real-time Streaming

Identity Resolution

DID document resolution, caching mechanisms, and AT Protocol-specific identity data extraction for decentralized identity management.

class IdResolver:
    def __init__(self, plc_url: Optional[str] = None, timeout: Optional[float] = None, cache: Optional[DidBaseCache] = None): ...

class AsyncIdResolver:
    def __init__(self, plc_url: Optional[str] = None, timeout: Optional[float] = None, cache: Optional[DidBaseCache] = None): ...

class AtprotoData:
    did: str
    signing_key: Optional[str]
    handle: Optional[str]
    pds: Optional[str]
    @classmethod
    def from_did_doc(cls, did_doc: DidDocument) -> 'AtprotoData': ...

Identity Resolution

JWT Operations

JSON Web Token parsing, payload extraction, validation, and signature verification for AT Protocol authentication and authorization.

class JwtPayload:
    iss: Optional[str]  # Issuer (DID)
    sub: Optional[str]  # Subject (DID)
    aud: Optional[Union[str, List[str]]]  # Audience (DID)
    exp: Optional[int]  # Expiration Time
    iat: Optional[int]  # Issued At
    scope: Optional[str]  # ATProto-specific scope

def parse_jwt(jwt: Union[str, bytes]) -> Tuple[bytes, bytes, Dict[str, Any], bytes]: ...
def get_jwt_payload(jwt: Union[str, bytes]) -> JwtPayload: ...
def verify_jwt(jwt: Union[str, bytes], signing_key: str) -> bool: ...
def verify_jwt_async(jwt: Union[str, bytes], signing_key: str) -> bool: ...

JWT Operations

Types

Common Types

from typing import Union, Optional, List, Dict, Any, Tuple

# Type aliases
CIDType = Union[str, CID]
SessionResponse = Union[models.ComAtprotoServerCreateSession.Response, models.ComAtprotoServerRefreshSession.Response]

# Enums
class SessionEvent(Enum):
    IMPORT = 'import'
    CREATE = 'create' 
    REFRESH = 'refresh'

Model Types

The models module contains 600+ generated classes representing the complete AT Protocol lexicon:

from atproto import models

# Actor models
models.AppBskyActorDefs.ProfileViewDetailed
models.AppBskyActorProfile.Record

# Feed models  
models.AppBskyFeedPost.Record
models.AppBskyFeedLike.Record

# Graph models
models.AppBskyGraphFollow.Record
models.AppBskyGraphBlock.Record

# Server models
models.ComAtprotoServerCreateSession.Response
models.ComAtprotoRepoCreateRecord.Response