0
# ATProto Python SDK
1

2
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.
3

4
## Package Information
5

6
- **Package Name**: atproto
7
- **Language**: Python
8
- **Installation**: `pip install atproto`
9
- **Documentation**: https://atproto.blue
10

11
## Core Imports
12

13
```python
14
import atproto
15
```
16

17
Common patterns for client operations:
18

19
```python
20
from atproto import Client, AsyncClient, models, Session
21
```
22

23
Specific functionality imports:
24

25
```python
26
from atproto import (
27
    # Client and session management
28
    Client, AsyncClient, Session, SessionEvent,
29
    # Core AT Protocol types
30
    CID, CAR, AtUri, NSID, DidDocument,
31
    # Cryptographic operations
32
    verify_signature, get_did_key, Multikey,
33
    # Real-time streaming
34
    FirehoseSubscribeReposClient, AsyncFirehoseSubscribeReposClient,
35
    # Identity resolution
36
    IdResolver, AsyncIdResolver, AtprotoData,
37
    # JWT operations
38
    verify_jwt, parse_jwt, JwtPayload
39
)
40
```
41

42
## Basic Usage
43

44
```python
45
import atproto
46

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

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

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

58
# Get the user's timeline
59
timeline = client.get_timeline()
60
for post in timeline.feed:
61
    print(f"{post.post.author.handle}: {post.post.record.text}")
62
```
63

64
Async usage:
65

66
```python
67
import asyncio
68
import atproto
69

70
async def main():
71
    client = atproto.AsyncClient()
72
    await client.login('your-handle.bsky.social', 'your-password')
73
    
74
    # Create a post asynchronously
75
    await client.send_post(text="Hello from async ATProto!")
76
    
77
    # Close the client
78
    await client.close()
79

80
asyncio.run(main())
81
```
82

83
## Architecture
84

85
The ATProto SDK is organized into six main functional areas:
86

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

94
## Capabilities
95

96
### Client Operations
97

98
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.
99

100
```python { .api }
101
class Client:
102
    def __init__(self, base_url: Optional[str] = None, *args, **kwargs): ...
103
    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: ...
104
    def send_post(self, text: Union[str, 'client_utils.TextBuilder'], **kwargs) -> models.AppBskyFeedPost.CreateRecordResponse: ...
105

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

111
class Session:
112
    handle: str
113
    did: str
114
    access_jwt: str
115
    refresh_jwt: str
116
    pds_endpoint: Optional[str]
117
```
118

119
[Client Operations](./client-operations.md)
120

121
### Core Functionality
122

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

125
```python { .api }
126
class CID:
127
    version: int
128
    codec: int
129
    hash: Multihash
130
    @classmethod
131
    def decode(cls, value: Union[str, bytes]) -> 'CID': ...
132

133
class CAR:
134
    root: CID
135
    blocks: Dict[CID, bytes]
136
    @classmethod
137
    def from_bytes(cls, data: bytes) -> 'CAR': ...
138

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

142
class NSID:
143
    segments: List[str]
144
    @classmethod
145
    def from_str(cls, nsid: str) -> 'NSID': ...
146
```
147

148
[Core Functionality](./core-functionality.md)
149

150
### Cryptographic Operations
151

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

154
```python { .api }
155
class Multikey:
156
    jwt_alg: str
157
    key_bytes: bytes
158
    @classmethod
159
    def from_str(cls, multikey: str) -> 'Multikey': ...
160

161
def get_did_key(key: Any) -> str: ...
162
def verify_signature(did_key: str, signing_input: bytes, signature: bytes) -> bool: ...
163
def bytes_to_multibase(encoding: str, data: bytes) -> str: ...
164
def multibase_to_bytes(data: str) -> bytes: ...
165
```
166

167
[Cryptographic Operations](./cryptographic-operations.md)
168

169
### Real-time Streaming
170

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

173
```python { .api }
174
class FirehoseSubscribeReposClient:
175
    def __init__(self, base_url: str = 'wss://bsky.network', *args, **kwargs): ...
176

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

180
def parse_subscribe_repos_message(message: MessageFrame) -> SubscribeReposMessage: ...
181
def parse_subscribe_labels_message(message: MessageFrame) -> SubscribeLabelsMessage: ...
182
```
183

184
[Real-time Streaming](./real-time-streaming.md)
185

186
### Identity Resolution
187

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

190
```python { .api }
191
class IdResolver:
192
    def __init__(self, plc_url: Optional[str] = None, timeout: Optional[float] = None, cache: Optional[DidBaseCache] = None): ...
193

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

197
class AtprotoData:
198
    did: str
199
    signing_key: Optional[str]
200
    handle: Optional[str]
201
    pds: Optional[str]
202
    @classmethod
203
    def from_did_doc(cls, did_doc: DidDocument) -> 'AtprotoData': ...
204
```
205

206
[Identity Resolution](./identity-resolution.md)
207

208
### JWT Operations
209

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

212
```python { .api }
213
class JwtPayload:
214
    iss: Optional[str]  # Issuer (DID)
215
    sub: Optional[str]  # Subject (DID)
216
    aud: Optional[Union[str, List[str]]]  # Audience (DID)
217
    exp: Optional[int]  # Expiration Time
218
    iat: Optional[int]  # Issued At
219
    scope: Optional[str]  # ATProto-specific scope
220

221
def parse_jwt(jwt: Union[str, bytes]) -> Tuple[bytes, bytes, Dict[str, Any], bytes]: ...
222
def get_jwt_payload(jwt: Union[str, bytes]) -> JwtPayload: ...
223
def verify_jwt(jwt: Union[str, bytes], signing_key: str) -> bool: ...
224
def verify_jwt_async(jwt: Union[str, bytes], signing_key: str) -> bool: ...
225
```
226

227
[JWT Operations](./jwt-operations.md)
228

229
## Types
230

231
### Common Types
232

233
```python { .api }
234
from typing import Union, Optional, List, Dict, Any, Tuple
235

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

240
# Enums
241
class SessionEvent(Enum):
242
    IMPORT = 'import'
243
    CREATE = 'create' 
244
    REFRESH = 'refresh'
245
```
246

247
### Model Types
248

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

251
```python { .api }
252
from atproto import models
253

254
# Actor models
255
models.AppBskyActorDefs.ProfileViewDetailed
256
models.AppBskyActorProfile.Record
257

258
# Feed models  
259
models.AppBskyFeedPost.Record
260
models.AppBskyFeedLike.Record
261

262
# Graph models
263
models.AppBskyGraphFollow.Record
264
models.AppBskyGraphBlock.Record
265

266
# Server models
267
models.ComAtprotoServerCreateSession.Response
268
models.ComAtprotoRepoCreateRecord.Response
269
```