a digital person for bluesky
cameron.stream
1"""Thread tool for adding posts to X threads atomically."""
2from typing import Optional
3from pydantic import BaseModel, Field, validator
4
5
6class XThreadPostArgs(BaseModel):
7 text: str = Field(
8 ...,
9 description="Text content for the X post (max 280 characters)"
10 )
11
12 @validator('text')
13 def validate_text_length(cls, v):
14 if len(v) > 280:
15 raise ValueError(f"Text exceeds 280 character limit (current: {len(v)} characters)")
16 return v
17
18
19def add_post_to_x_thread(text: str) -> str:
20 """
21 Add a single post to the current X reply thread. This tool indicates to the handler
22 that it should add this post to the ongoing reply thread context when responding to a mention.
23
24 This is an atomic operation - each call adds exactly one post. The handler (x_bot.py or similar)
25 manages the thread state and ensures proper threading when multiple posts are queued.
26
27 Args:
28 text: Text content for the post (max 280 characters)
29
30 Returns:
31 Confirmation message that the post has been queued for the reply thread
32
33 Raises:
34 Exception: If text exceeds character limit. On failure, the post will be omitted
35 from the reply thread and the agent may try again with corrected text.
36 """
37 # Validate input
38 if len(text) > 280:
39 raise Exception(f"Text exceeds 280 character limit (current: {len(text)} characters). This post will be omitted from the thread. You may try again with shorter text.")
40
41 # Return confirmation - the actual posting will be handled by x_bot.py or similar
42 return f"X post queued for reply thread: {text[:50]}{'...' if len(text) > 50 else ''}"