cameron.stream
a digital person for bluesky
Add X reply functionality for specific post interaction
- Implement post_reply method in XClient class
- Add reply_to_cameron_post function for testing specific post
- Support command line args: loop, reply
- Handle API permissions and rate limiting gracefully
- Ready for elevated X API access when available
🤖 Generated with [Claude Code](https://claude.ai/code)
Co-Authored-By: Claude <noreply@anthropic.com>
+78
-2
x.py
+78
-2
x.py
···
92
92
93
93
response = self._make_request(endpoint, params)
94
94
return response.get("data") if response else None
95
+
96
+
def post_reply(self, reply_text: str, in_reply_to_tweet_id: str) -> Optional[Dict]:
97
+
"""
98
+
Post a reply to a specific tweet.
99
+
100
+
Args:
101
+
reply_text: The text content of the reply
102
+
in_reply_to_tweet_id: The ID of the tweet to reply to
103
+
104
+
Returns:
105
+
Response data if successful, None if failed
106
+
"""
107
+
endpoint = "/tweets"
108
+
109
+
payload = {
110
+
"text": reply_text,
111
+
"reply": {
112
+
"in_reply_to_tweet_id": in_reply_to_tweet_id
113
+
}
114
+
}
115
+
116
+
try:
117
+
url = f"{self.base_url}{endpoint}"
118
+
response = requests.post(url, headers=self.headers, json=payload)
119
+
response.raise_for_status()
120
+
121
+
result = response.json()
122
+
logger.info(f"Successfully posted reply to tweet {in_reply_to_tweet_id}")
123
+
return result
124
+
125
+
except requests.exceptions.HTTPError as e:
126
+
if response.status_code == 401:
127
+
logger.error("X API authentication failed for posting - check your bearer token permissions")
128
+
elif response.status_code == 403:
129
+
logger.error("X API posting forbidden - may need elevated access or different permissions")
130
+
elif response.status_code == 429:
131
+
logger.error("X API rate limit exceeded for posting")
132
+
else:
133
+
logger.error(f"X API post request failed: {e}")
134
+
logger.error(f"Response content: {response.text}")
135
+
return None
136
+
except Exception as e:
137
+
logger.error(f"Unexpected error posting to X: {e}")
138
+
return None
95
139
96
140
def load_x_config(config_path: str = "config.yaml") -> Dict[str, str]:
97
141
"""Load X configuration from config file."""
···
154
198
except Exception as e:
155
199
print(f"Test failed: {e}")
156
200
201
+
def reply_to_cameron_post():
202
+
"""Reply to Cameron's specific X post."""
203
+
try:
204
+
client = create_x_client()
205
+
206
+
# Cameron's post ID from the URL: https://x.com/cameron_pfiffer/status/1950690566909710618
207
+
cameron_post_id = "1950690566909710618"
208
+
209
+
# Simple reply message
210
+
reply_text = "Hello from void! 🤖 Testing X integration."
211
+
212
+
print(f"Attempting to reply to post {cameron_post_id}")
213
+
print(f"Reply text: {reply_text}")
214
+
215
+
result = client.post_reply(reply_text, cameron_post_id)
216
+
217
+
if result:
218
+
print(f"✅ Successfully posted reply!")
219
+
print(f"Reply ID: {result.get('data', {}).get('id', 'Unknown')}")
220
+
else:
221
+
print("❌ Failed to post reply")
222
+
223
+
except Exception as e:
224
+
print(f"Reply failed: {e}")
225
+
157
226
def x_notification_loop():
158
227
"""
159
228
Simple X notification loop that fetches mentions and logs them.
···
229
298
230
299
if __name__ == "__main__":
231
300
import sys
232
-
if len(sys.argv) > 1 and sys.argv[1] == "loop":
233
-
x_notification_loop()
301
+
if len(sys.argv) > 1:
302
+
if sys.argv[1] == "loop":
303
+
x_notification_loop()
304
+
elif sys.argv[1] == "reply":
305
+
reply_to_cameron_post()
306
+
else:
307
+
print("Usage: python x.py [loop|reply]")
308
+
print(" loop - Run the notification monitoring loop")
309
+
print(" reply - Reply to Cameron's specific post")
234
310
else:
235
311
test_x_client()