0
# Client Operations
1
2
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.
3
4
## Capabilities
5
6
### Client Classes
7
8
#### Synchronous Client
9
10
The main synchronous client for AT Protocol operations with automatic session management and thread-safe JWT refresh.
11
12
```python { .api }
13
class Client:
14
"""
15
High-level synchronous client for XRPC and ATProto operations.
16
17
Attributes:
18
me (Optional[models.AppBskyActorDefs.ProfileViewDetailed]): Current user profile
19
"""
20
def __init__(self, base_url: Optional[str] = None, *args, **kwargs):
21
"""
22
Initialize the client.
23
24
Args:
25
base_url (str, optional): Custom base URL for AT Protocol server
26
"""
27
28
def login(
29
self,
30
login: Optional[str] = None,
31
password: Optional[str] = None,
32
session_string: Optional[str] = None,
33
auth_factor_token: Optional[str] = None
34
) -> models.AppBskyActorDefs.ProfileViewDetailed:
35
"""
36
Authenticate with AT Protocol server.
37
38
Args:
39
login (str, optional): Handle or email of the account
40
password (str, optional): Main or app-specific password
41
session_string (str, optional): Session string for re-authentication
42
auth_factor_token (str, optional): Auth factor token for Email 2FA
43
44
Note:
45
Either session_string or login and password should be provided.
46
47
Returns:
48
AppBskyActorDefs.ProfileViewDetailed: Profile information
49
"""
50
51
def send_post(
52
self,
53
text: Union[str, 'client_utils.TextBuilder'],
54
profile_identify: Optional[str] = None,
55
reply_to: Optional['models.AppBskyFeedPost.ReplyRef'] = None,
56
embed: Optional[Union[
57
'models.AppBskyEmbedImages.Main',
58
'models.AppBskyEmbedExternal.Main',
59
'models.AppBskyEmbedRecord.Main',
60
'models.AppBskyEmbedRecordWithMedia.Main',
61
'models.AppBskyEmbedVideo.Main'
62
]] = None,
63
langs: Optional[List[str]] = None,
64
facets: Optional[List['models.AppBskyRichtextFacet.Main']] = None
65
) -> models.AppBskyFeedPost.CreateRecordResponse:
66
"""
67
Create a post record.
68
69
Args:
70
text (str or TextBuilder): Post content
71
profile_identify (str, optional): Handle or DID where to send post
72
reply_to (AppBskyFeedPost.ReplyRef, optional): Root and parent of post to reply to
73
embed (optional): Embed models attached to the post
74
langs (List[str], optional): List of languages used in the post
75
facets (List[AppBskyRichtextFacet.Main], optional): Rich text facets
76
77
Returns:
78
AppBskyFeedPost.CreateRecordResponse: Reference to created record
79
"""
80
81
def get_timeline(self, **kwargs) -> models.AppBskyFeedGetTimeline.Response:
82
"""
83
Get the authenticated user's timeline.
84
85
Args:
86
**kwargs: Timeline parameters (limit, cursor, etc.)
87
88
Returns:
89
AppBskyFeedGetTimeline.Response: Timeline data
90
"""
91
92
def get_profile(self, actor: str) -> models.AppBskyActorGetProfile.Response:
93
"""
94
Get profile information for an actor.
95
96
Args:
97
actor (str): Handle or DID of the actor
98
99
Returns:
100
AppBskyActorGetProfile.Response: Profile information
101
"""
102
103
def delete_post(self, post_uri: str) -> bool:
104
"""
105
Delete a post record.
106
107
Args:
108
post_uri (str): URI of the post to delete
109
110
Returns:
111
bool: True if deletion was successful
112
"""
113
114
def send_image(self, text: str, image: bytes, image_alt: str, **kwargs) -> models.AppBskyFeedPost.CreateRecordResponse:
115
"""
116
Send a post with an image.
117
118
Args:
119
text (str): Post text content
120
image (bytes): Image data
121
image_alt (str): Alt text for the image
122
**kwargs: Additional post parameters
123
124
Returns:
125
AppBskyFeedPost.CreateRecordResponse: Created post record
126
"""
127
128
def get_post(self, post_rkey: str, profile_identify: str, cid: Optional[str] = None) -> models.ComAtprotoRepoGetRecord.Response:
129
"""
130
Get a single post record.
131
132
Args:
133
post_rkey (str): Record key of the post
134
profile_identify (str): Handle or DID of the post author
135
cid (str, optional): Specific version CID
136
137
Returns:
138
ComAtprotoRepoGetRecord.Response: Post record data
139
"""
140
141
def like(self, uri: str, cid: str) -> models.AppBskyFeedLike.CreateRecordResponse:
142
"""
143
Like a post.
144
145
Args:
146
uri (str): AT-URI of the post to like
147
cid (str): CID of the post to like
148
149
Returns:
150
AppBskyFeedLike.CreateRecordResponse: Like record reference
151
"""
152
153
def unlike(self, like_uri: str) -> bool:
154
"""
155
Remove a like from a post.
156
157
Args:
158
like_uri (str): URI of the like record to delete
159
160
Returns:
161
bool: True if unlike was successful
162
"""
163
164
def repost(self, uri: str, cid: str) -> models.AppBskyFeedRepost.CreateRecordResponse:
165
"""
166
Repost a post.
167
168
Args:
169
uri (str): AT-URI of the post to repost
170
cid (str): CID of the post to repost
171
172
Returns:
173
AppBskyFeedRepost.CreateRecordResponse: Repost record reference
174
"""
175
176
def follow(self, subject: str) -> models.AppBskyGraphFollow.CreateRecordResponse:
177
"""
178
Follow a user.
179
180
Args:
181
subject (str): Handle or DID of the user to follow
182
183
Returns:
184
AppBskyGraphFollow.CreateRecordResponse: Follow record reference
185
"""
186
187
def unfollow(self, follow_uri: str) -> bool:
188
"""
189
Unfollow a user.
190
191
Args:
192
follow_uri (str): URI of the follow record to delete
193
194
Returns:
195
bool: True if unfollow was successful
196
"""
197
```
198
199
Usage example:
200
201
```python
202
from atproto import Client
203
204
# Initialize and authenticate
205
client = Client()
206
session = client.login('alice.bsky.social', 'password')
207
208
# Access user profile
209
print(f"Logged in as: {client.me.handle}")
210
print(f"DID: {client.me.did}")
211
212
# Create a post
213
response = client.send_post(text="Hello AT Protocol!")
214
print(f"Post created: {response.uri}")
215
216
# Get timeline
217
timeline = client.get_timeline(limit=10)
218
for item in timeline.feed:
219
author = item.post.author
220
text = item.post.record.text
221
print(f"{author.handle}: {text}")
222
```
223
224
#### Asynchronous Client
225
226
The asynchronous version of the client with the same interface but async methods.
227
228
```python { .api }
229
class AsyncClient:
230
"""
231
High-level asynchronous client for XRPC and ATProto operations.
232
233
Attributes:
234
me (Optional[models.AppBskyActorDefs.ProfileViewDetailed]): Current user profile
235
"""
236
def __init__(self, base_url: Optional[str] = None, *args, **kwargs):
237
"""
238
Initialize the async client.
239
240
Args:
241
base_url (str, optional): Custom base URL for AT Protocol server
242
"""
243
244
async def login(
245
self,
246
login: Optional[str] = None,
247
password: Optional[str] = None,
248
session_string: Optional[str] = None,
249
auth_factor_token: Optional[str] = None
250
) -> models.AppBskyActorDefs.ProfileViewDetailed:
251
"""
252
Authenticate with AT Protocol server asynchronously.
253
254
Args:
255
login (str, optional): Handle or email of the account
256
password (str, optional): Main or app-specific password
257
session_string (str, optional): Session string for re-authentication
258
auth_factor_token (str, optional): Auth factor token for Email 2FA
259
260
Note:
261
Either session_string or login and password should be provided.
262
263
Returns:
264
AppBskyActorDefs.ProfileViewDetailed: Profile information
265
"""
266
267
async def send_post(
268
self,
269
text: Union[str, 'client_utils.TextBuilder'],
270
profile_identify: Optional[str] = None,
271
reply_to: Optional['models.AppBskyFeedPost.ReplyRef'] = None,
272
embed: Optional[Union[
273
'models.AppBskyEmbedImages.Main',
274
'models.AppBskyEmbedExternal.Main',
275
'models.AppBskyEmbedRecord.Main',
276
'models.AppBskyEmbedRecordWithMedia.Main',
277
'models.AppBskyEmbedVideo.Main'
278
]] = None,
279
langs: Optional[List[str]] = None,
280
facets: Optional[List['models.AppBskyRichtextFacet.Main']] = None
281
) -> models.AppBskyFeedPost.CreateRecordResponse:
282
"""
283
Create a post record asynchronously.
284
285
Args:
286
text (str or TextBuilder): Post content
287
profile_identify (str, optional): Handle or DID where to send post
288
reply_to (AppBskyFeedPost.ReplyRef, optional): Root and parent of post to reply to
289
embed (optional): Embed models attached to the post
290
langs (List[str], optional): List of languages used in the post
291
facets (List[AppBskyRichtextFacet.Main], optional): Rich text facets
292
293
Returns:
294
AppBskyFeedPost.CreateRecordResponse: Reference to created record
295
"""
296
297
async def delete_post(self, post_uri: str) -> bool:
298
"""Delete a post record asynchronously."""
299
300
async def send_image(self, text: str, image: bytes, image_alt: str, **kwargs) -> models.AppBskyFeedPost.CreateRecordResponse:
301
"""Send a post with an image asynchronously."""
302
303
async def like(self, uri: str, cid: str) -> models.AppBskyFeedLike.CreateRecordResponse:
304
"""Like a post asynchronously."""
305
306
async def follow(self, subject: str) -> models.AppBskyGraphFollow.CreateRecordResponse:
307
"""Follow a user asynchronously."""
308
309
async def close(self):
310
"""Close the async HTTP client connection."""
311
```
312
313
Usage example:
314
315
```python
316
import asyncio
317
from atproto import AsyncClient
318
319
async def main():
320
client = AsyncClient()
321
322
# Authenticate
323
await client.login('alice.bsky.social', 'password')
324
325
# Create a post
326
await client.send_post(text="Hello from async!")
327
328
# Always close the client
329
await client.close()
330
331
asyncio.run(main())
332
```
333
334
### Session Management
335
336
#### Session Class
337
338
Represents an authenticated session with AT Protocol, including tokens and user information.
339
340
```python { .api }
341
class Session:
342
"""
343
Authenticated session with AT Protocol.
344
345
Attributes:
346
handle (str): User handle
347
did (str): Decentralized identifier
348
access_jwt (str): Access token for API calls
349
refresh_jwt (str): Refresh token for renewing access
350
pds_endpoint (Optional[str]): Personal Data Server endpoint
351
"""
352
handle: str
353
did: str
354
access_jwt: str
355
refresh_jwt: str
356
pds_endpoint: Optional[str]
357
358
def encode(self) -> str:
359
"""
360
Serialize session to string for storage.
361
362
Returns:
363
str: Encoded session string
364
"""
365
366
@classmethod
367
def decode(cls, session_string: str) -> 'Session':
368
"""
369
Deserialize session from string.
370
371
Args:
372
session_string (str): Encoded session string
373
374
Returns:
375
Session: Decoded session object
376
"""
377
378
def copy(self) -> 'Session':
379
"""
380
Create a copy of the session.
381
382
Returns:
383
Session: Copied session object
384
"""
385
386
@property
387
def access_jwt_payload(self) -> 'JwtPayload':
388
"""Get decoded access token payload."""
389
390
@property
391
def refresh_jwt_payload(self) -> 'JwtPayload':
392
"""Get decoded refresh token payload."""
393
```
394
395
Usage example:
396
397
```python
398
from atproto import Client, Session
399
400
# Login and get session
401
client = Client()
402
session = client.login('alice.bsky.social', 'password')
403
404
# Save session for later use
405
session_string = session.encode()
406
# Store session_string securely...
407
408
# Restore session later
409
restored_session = Session.decode(session_string)
410
client = Client()
411
client.session = restored_session
412
```
413
414
#### Session Events
415
416
Events that occur during session lifecycle for monitoring and debugging.
417
418
```python { .api }
419
class SessionEvent(Enum):
420
"""Events during session lifecycle."""
421
IMPORT = 'import' # Session imported from storage
422
CREATE = 'create' # New session created via login
423
REFRESH = 'refresh' # Session refreshed with new tokens
424
```
425
426
### Request Handling
427
428
#### Synchronous Request Handler
429
430
Low-level HTTP request handler using HTTPX for synchronous operations.
431
432
```python { .api }
433
class Request:
434
"""
435
Synchronous HTTP request handler with HTTPX.
436
"""
437
def get(self, *args, **kwargs) -> 'Response':
438
"""
439
Make a GET request.
440
441
Returns:
442
Response: HTTP response wrapper
443
"""
444
445
def post(self, *args, **kwargs) -> 'Response':
446
"""
447
Make a POST request.
448
449
Returns:
450
Response: HTTP response wrapper
451
"""
452
453
def set_additional_headers(self, headers: Dict[str, str]):
454
"""
455
Set additional headers for all requests.
456
457
Args:
458
headers (Dict[str, str]): Headers to add
459
"""
460
461
def clone(self) -> 'Request':
462
"""
463
Create a cloned request handler.
464
465
Returns:
466
Request: Cloned request instance
467
"""
468
```
469
470
#### Asynchronous Request Handler
471
472
Low-level HTTP request handler for asynchronous operations.
473
474
```python { .api }
475
class AsyncRequest:
476
"""
477
Asynchronous HTTP request handler with HTTPX.
478
"""
479
async def get(self, *args, **kwargs) -> 'Response':
480
"""
481
Make an async GET request.
482
483
Returns:
484
Response: HTTP response wrapper
485
"""
486
487
async def post(self, *args, **kwargs) -> 'Response':
488
"""
489
Make an async POST request.
490
491
Returns:
492
Response: HTTP response wrapper
493
"""
494
495
async def close(self):
496
"""Close the async HTTP client."""
497
```
498
499
#### Response Wrapper
500
501
HTTP response wrapper providing unified access to response data.
502
503
```python { .api }
504
class Response:
505
"""
506
HTTP response wrapper.
507
508
Attributes:
509
success (bool): Whether the request was successful
510
status_code (int): HTTP status code
511
content (Optional[Union[Dict[str, Any], bytes, XrpcError]]): Response content
512
headers (Dict[str, Any]): Response headers
513
"""
514
success: bool
515
status_code: int
516
content: Optional[Union[Dict[str, Any], bytes, XrpcError]]
517
headers: Dict[str, Any]
518
```
519
520
### Utilities
521
522
#### Text Builder
523
524
Helper for constructing rich text with facets (mentions, links, tags).
525
526
```python { .api }
527
class TextBuilder:
528
"""
529
Helper for constructing rich text with facets.
530
"""
531
def __init__(self):
532
"""
533
Initialize text builder.
534
"""
535
536
def text(self, text: str) -> 'TextBuilder':
537
"""
538
Add plain text to the builder.
539
540
Args:
541
text (str): Text to add
542
543
Returns:
544
TextBuilder: Self for chaining
545
"""
546
547
def mention(self, text: str, did: str) -> 'TextBuilder':
548
"""
549
Add a mention facet.
550
551
Args:
552
text (str): Text of the mention (e.g., '@alice')
553
did (str): DID of the mentioned user
554
555
Returns:
556
TextBuilder: Self for chaining
557
"""
558
559
def link(self, text: str, url: str) -> 'TextBuilder':
560
"""
561
Add a link facet.
562
563
Args:
564
text (str): Link text
565
url (str): Target URL
566
567
Returns:
568
TextBuilder: Self for chaining
569
"""
570
571
def tag(self, text: str, tag: str) -> 'TextBuilder':
572
"""
573
Add a hashtag facet.
574
575
Args:
576
text (str): Text of the tag (e.g., '#atproto')
577
tag (str): Tag name (without #)
578
579
Returns:
580
TextBuilder: Self for chaining
581
"""
582
583
def build_text(self) -> str:
584
"""
585
Build the text from current state.
586
587
Returns:
588
str: Built text string
589
"""
590
591
def build_facets(self) -> List['models.AppBskyRichtextFacet.Main']:
592
"""
593
Build the facets from current state.
594
595
Returns:
596
List[models.AppBskyRichtextFacet.Main]: Built facets list
597
"""
598
```
599
600
Usage example:
601
602
```python
603
from atproto import client_utils
604
605
# Build rich text with mentions and links
606
builder = client_utils.TextBuilder()
607
rich_text = (builder
608
.text("Hello ")
609
.mention("@alice", "did:plc:alice123")
610
.text("! Check out ")
611
.link("this article", "https://example.com")
612
.text(" about ")
613
.tag("#atproto", "atproto"))
614
615
# Use in a post (TextBuilder can be passed directly)
616
client.send_post(rich_text)
617
```
618
619
### Exception Handling
620
621
```python { .api }
622
class ModelError(Exception):
623
"""Base exception for model-related errors."""
624
625
class NetworkError(Exception):
626
"""Network-related request errors."""
627
628
class UnauthorizedError(Exception):
629
"""Authentication/authorization errors."""
630
631
class LoginRequiredError(Exception):
632
"""Thrown when login is required for an operation."""
633
```
634
635
Common error handling patterns:
636
637
```python
638
from atproto import Client, LoginRequiredError, NetworkError
639
640
client = Client()
641
642
try:
643
client.login('user@example.com', 'wrong-password')
644
except UnauthorizedError:
645
print("Invalid credentials")
646
647
try:
648
# This requires authentication
649
client.send_post(text="Hello!")
650
except LoginRequiredError:
651
print("Please log in first")
652
except NetworkError as e:
653
print(f"Network error: {e}")
654
```