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