a digital person for bluesky
cameron.stream
1"""Post tool for creating Bluesky posts."""
2from pydantic import BaseModel, Field
3
4
5class PostArgs(BaseModel):
6 text: str = Field(..., description="The text content to post (max 300 characters)")
7
8
9def create_new_bluesky_post(text: str) -> str:
10 """
11 Create a NEW standalone post on Bluesky. This tool creates independent posts that
12 start new conversations.
13
14 IMPORTANT: This tool is ONLY for creating new posts. To reply to an existing post,
15 use reply_to_bluesky_post instead.
16
17 Args:
18 text: The post content (max 300 characters)
19
20 Returns:
21 Success message with post URL
22
23 Raises:
24 Exception: If the post fails
25 """
26 import os
27 import requests
28 from datetime import datetime, timezone
29
30 try:
31 # Validate character limit
32 if len(text) > 300:
33 raise Exception(f"Post exceeds 300 character limit (current: {len(text)} characters)")
34
35 # Get credentials from environment
36 username = os.getenv("BSKY_USERNAME")
37 password = os.getenv("BSKY_PASSWORD")
38 pds_host = os.getenv("PDS_URI", "https://bsky.social")
39
40 if not username or not password:
41 raise Exception("BSKY_USERNAME and BSKY_PASSWORD environment variables must be set")
42
43 # Create session
44 session_url = f"{pds_host}/xrpc/com.atproto.server.createSession"
45 session_data = {
46 "identifier": username,
47 "password": password
48 }
49
50 session_response = requests.post(session_url, json=session_data, timeout=10)
51 session_response.raise_for_status()
52 session = session_response.json()
53 access_token = session.get("accessJwt")
54 user_did = session.get("did")
55
56 if not access_token or not user_did:
57 raise Exception("Failed to get access token or DID from session")
58
59 # Build post record with facets for mentions and URLs
60 now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
61
62 post_record = {
63 "$type": "app.bsky.feed.post",
64 "text": text,
65 "createdAt": now,
66 }
67
68 # Add facets for mentions and URLs
69 import re
70 facets = []
71
72 # Parse mentions
73 mention_regex = rb"[$|\W](@([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)"
74 text_bytes = text.encode("UTF-8")
75
76 for m in re.finditer(mention_regex, text_bytes):
77 handle = m.group(1)[1:].decode("UTF-8") # Remove @ prefix
78 try:
79 resolve_resp = requests.get(
80 f"{pds_host}/xrpc/com.atproto.identity.resolveHandle",
81 params={"handle": handle},
82 timeout=5
83 )
84 if resolve_resp.status_code == 200:
85 did = resolve_resp.json()["did"]
86 facets.append({
87 "index": {
88 "byteStart": m.start(1),
89 "byteEnd": m.end(1),
90 },
91 "features": [{"$type": "app.bsky.richtext.facet#mention", "did": did}],
92 })
93 except:
94 continue
95
96 # Parse URLs
97 url_regex = rb"[$|\W](https?:\/\/(www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b([-a-zA-Z0-9()@:%_\+.~#?&//=]*[-a-zA-Z0-9@%_\+~#//=])?)"
98
99 for m in re.finditer(url_regex, text_bytes):
100 url = m.group(1).decode("UTF-8")
101 facets.append({
102 "index": {
103 "byteStart": m.start(1),
104 "byteEnd": m.end(1),
105 },
106 "features": [{"$type": "app.bsky.richtext.facet#link", "uri": url}],
107 })
108
109 if facets:
110 post_record["facets"] = facets
111
112 # Create the post
113 create_record_url = f"{pds_host}/xrpc/com.atproto.repo.createRecord"
114 headers = {"Authorization": f"Bearer {access_token}"}
115
116 create_data = {
117 "repo": user_did,
118 "collection": "app.bsky.feed.post",
119 "record": post_record
120 }
121
122 post_response = requests.post(create_record_url, headers=headers, json=create_data, timeout=10)
123 post_response.raise_for_status()
124 result = post_response.json()
125
126 post_uri = result.get("uri")
127 handle = session.get("handle", username)
128 rkey = post_uri.split("/")[-1] if post_uri else ""
129 post_url = f"https://bsky.app/profile/{handle}/post/{rkey}"
130
131 return f"Successfully posted to Bluesky!\nPost URL: {post_url}\nText: {text}"
132
133 except Exception as e:
134 raise Exception(f"Error posting to Bluesky: {str(e)}")