ATProto Python SDK

import atproto

from atproto import Client, AsyncClient, models, Session

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
)

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}")

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

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]

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': ...

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: ...

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: ...

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': ...

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: ...

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'

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