a digital person for bluesky
cameron.stream
1"""Reply to Bluesky posts tool."""
2
3from typing import Optional
4from pydantic import BaseModel, Field
5
6class ReplyArgs(BaseModel):
7 """Arguments for replying to a Bluesky post."""
8 text: str = Field(..., description="The reply text content (max 300 characters)")
9 reply_to_uri: str = Field(..., description="The AT URI of the post being replied to (required)")
10 reply_to_cid: str = Field(..., description="The CID of the post being replied to (required)")
11
12def reply_to_bluesky_post(text: str, reply_to_uri: str, reply_to_cid: str) -> str:
13 """
14 Reply to an existing Bluesky post. This tool creates a threaded reply to a specific post.
15
16 IMPORTANT: This tool is ONLY for replying to existing posts. To create a new standalone post,
17 use create_new_bluesky_post instead.
18
19 Args:
20 text: The reply text content (max 300 characters)
21 reply_to_uri: The AT URI of the post being replied to (required)
22 reply_to_cid: The CID of the post being replied to (required)
23
24 Returns:
25 Success message with post URI or error message
26
27 Raises:
28 Exception: If the reply fails
29 """
30 if len(text) > 300:
31 raise ValueError(f"Reply text too long: {len(text)} characters (max 300)")
32
33 if not reply_to_uri:
34 raise ValueError("reply_to_uri is required for replies")
35
36 if not reply_to_cid:
37 raise ValueError("reply_to_cid is required for replies")
38
39 import os
40 import requests
41 from datetime import datetime, timezone
42
43 try:
44 # Get credentials from environment
45 username = os.getenv("BSKY_USERNAME")
46 password = os.getenv("BSKY_PASSWORD")
47 pds_host = os.getenv("PDS_URI", "https://bsky.social")
48
49 if not username or not password:
50 raise Exception("BSKY_USERNAME and BSKY_PASSWORD environment variables must be set")
51
52 # Create session
53 session_url = f"{pds_host}/xrpc/com.atproto.server.createSession"
54 session_data = {
55 "identifier": username,
56 "password": password
57 }
58
59 session_response = requests.post(session_url, json=session_data, timeout=10)
60 session_response.raise_for_status()
61 session = session_response.json()
62 access_token = session.get("accessJwt")
63 user_did = session.get("did")
64
65 if not access_token or not user_did:
66 raise Exception("Failed to get access token or DID from session")
67
68 # Build reply record
69 now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
70
71 # Create the reply reference
72 reply_ref = {
73 "root": {"uri": reply_to_uri, "cid": reply_to_cid},
74 "parent": {"uri": reply_to_uri, "cid": reply_to_cid}
75 }
76
77 post_record = {
78 "$type": "app.bsky.feed.post",
79 "text": text,
80 "createdAt": now,
81 "reply": reply_ref
82 }
83
84 # Create the post
85 create_record_url = f"{pds_host}/xrpc/com.atproto.repo.createRecord"
86 headers = {"Authorization": f"Bearer {access_token}"}
87
88 create_data = {
89 "repo": user_did,
90 "collection": "app.bsky.feed.post",
91 "record": post_record
92 }
93
94 post_response = requests.post(create_record_url, headers=headers, json=create_data, timeout=10)
95 post_response.raise_for_status()
96 result = post_response.json()
97
98 post_uri = result.get("uri")
99 handle = session.get("handle", username)
100 rkey = post_uri.split("/")[-1] if post_uri else ""
101 post_url = f"https://bsky.app/profile/{handle}/post/{rkey}"
102
103 return f"Successfully posted reply to Bluesky!\nReply URL: {post_url}\nText: {text}"
104
105 except Exception as e:
106 raise Exception(f"Error replying to Bluesky post: {str(e)}")