a digital person for bluesky
cameron.stream
1"""Whitewind blog post creation tool."""
2from typing import Optional
3from pydantic import BaseModel, Field
4
5
6class WhitewindPostArgs(BaseModel):
7 title: str = Field(
8 ...,
9 description="Title of the blog post"
10 )
11 content: str = Field(
12 ...,
13 description="Main content of the blog post (Markdown supported)"
14 )
15 subtitle: Optional[str] = Field(
16 default=None,
17 description="Optional subtitle for the blog post"
18 )
19
20
21def create_whitewind_blog_post(title: str, content: str, subtitle: Optional[str] = None) -> str:
22 """
23 Create a new blog post on Whitewind.
24
25 This tool creates blog posts using the com.whtwnd.blog.entry lexicon on the ATProto network.
26 The posts are publicly visible and use the github-light theme.
27
28 Args:
29 title: Title of the blog post
30 content: Main content of the blog post (Markdown supported)
31 subtitle: Optional subtitle for the blog post
32
33 Returns:
34 Success message with the blog post URL
35
36 Raises:
37 Exception: If the post creation fails
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 handle = session.get("handle", username)
65
66 if not access_token or not user_did:
67 raise Exception("Failed to get access token or DID from session")
68
69 # Create blog post record
70 now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
71
72 blog_record = {
73 "$type": "com.whtwnd.blog.entry",
74 "theme": "github-light",
75 "title": title,
76 "content": content,
77 "createdAt": now,
78 "visibility": "public"
79 }
80
81 # Add subtitle if provided
82 if subtitle:
83 blog_record["subtitle"] = subtitle
84
85 # Create the record
86 headers = {"Authorization": f"Bearer {access_token}"}
87 create_record_url = f"{pds_host}/xrpc/com.atproto.repo.createRecord"
88
89 create_data = {
90 "repo": user_did,
91 "collection": "com.whtwnd.blog.entry",
92 "record": blog_record
93 }
94
95 post_response = requests.post(create_record_url, headers=headers, json=create_data, timeout=10)
96 post_response.raise_for_status()
97 result = post_response.json()
98
99 # Extract the record key from the URI
100 post_uri = result.get("uri")
101 if post_uri:
102 rkey = post_uri.split("/")[-1]
103 # Construct the Whitewind blog URL
104 blog_url = f"https://whtwnd.com/{handle}/{rkey}"
105 else:
106 blog_url = "URL generation failed"
107
108 # Build success message
109 success_parts = [
110 f"Successfully created Whitewind blog post!",
111 f"Title: {title}"
112 ]
113 if subtitle:
114 success_parts.append(f"Subtitle: {subtitle}")
115 success_parts.extend([
116 f"URL: {blog_url}",
117 f"Theme: github-light",
118 f"Visibility: public"
119 ])
120
121 return "\n".join(success_parts)
122
123 except Exception as e:
124 raise Exception(f"Error creating Whitewind blog post: {str(e)}")