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