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