a digital person for bluesky
cameron.stream
1"""Tool for creating standalone posts on X (Twitter)."""
2import os
3import json
4import requests
5from typing import Optional
6from pydantic import BaseModel, Field, validator
7from requests_oauthlib import OAuth1
8
9
10class PostToXArgs(BaseModel):
11 text: str = Field(
12 ...,
13 description="Text content for the X post (max 280 characters)"
14 )
15
16 @validator('text')
17 def validate_text_length(cls, v):
18 if len(v) > 280:
19 raise ValueError(f"Text exceeds 280 character limit (current: {len(v)} characters)")
20 return v
21
22
23def post_to_x(text: str) -> str:
24 """
25 Create a new standalone post on X (Twitter). This is not a reply to another post.
26
27 Use this tool when you want to share a thought, observation, or announcement
28 that isn't in response to anyone else's post.
29
30 Args:
31 text: Text content for the post (max 280 characters)
32
33 Returns:
34 Success message with post ID if successful
35
36 Raises:
37 Exception: If text exceeds character limit or posting fails
38 """
39 import requests
40
41 # Validate input
42 if len(text) > 280:
43 raise Exception(f"Text exceeds 280 character limit (current: {len(text)} characters)")
44
45 try:
46 # Get credentials from environment variables
47 consumer_key = os.environ.get("X_CONSUMER_KEY")
48 consumer_secret = os.environ.get("X_CONSUMER_SECRET")
49 access_token = os.environ.get("X_ACCESS_TOKEN")
50 access_token_secret = os.environ.get("X_ACCESS_TOKEN_SECRET")
51
52 if not all([consumer_key, consumer_secret, access_token, access_token_secret]):
53 raise Exception("Missing X API credentials in environment variables")
54
55 # Create OAuth 1.0a authentication
56 auth = OAuth1(
57 consumer_key,
58 client_secret=consumer_secret,
59 resource_owner_key=access_token,
60 resource_owner_secret=access_token_secret
61 )
62
63 # Prepare the request
64 url = "https://api.x.com/2/tweets"
65 headers = {"Content-Type": "application/json"}
66 payload = {"text": text}
67
68 # Make the POST request
69 response = requests.post(
70 url,
71 headers=headers,
72 json=payload,
73 auth=auth
74 )
75
76 # Check response
77 if response.status_code == 201:
78 result = response.json()
79 if 'data' in result:
80 tweet_id = result['data'].get('id', 'unknown')
81 tweet_url = f"https://x.com/i/status/{tweet_id}"
82 return f"Successfully posted to X. Tweet ID: {tweet_id}. URL: {tweet_url}"
83 else:
84 raise Exception(f"Unexpected response format: {result}")
85 else:
86 error_msg = f"X API error: {response.status_code} - {response.text}"
87 raise Exception(error_msg)
88
89 except requests.exceptions.RequestException as e:
90 error_msg = f"Network error posting to X: {str(e)}"
91 raise Exception(error_msg)
92 except Exception as e:
93 if "Missing X API credentials" in str(e) or "X API error" in str(e):
94 raise
95 error_msg = f"Unexpected error posting to X: {str(e)}"
96 raise Exception(error_msg)