diff --git a/tools/blocks.py b/tools/blocks.py index 0c6dbd6..e3c243d 100644 --- a/tools/blocks.py +++ b/tools/blocks.py @@ -45,6 +45,35 @@ class UserNoteViewArgs(BaseModel): handle: str = Field(..., description="User Bluesky handle (e.g., 'cameron.pfiffer.org')") +# X (Twitter) User Block Management +class AttachXUserBlocksArgs(BaseModel): + user_ids: List[str] = Field(..., description="List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592'])") + + +class DetachXUserBlocksArgs(BaseModel): + user_ids: List[str] = Field(..., description="List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592'])") + + +class XUserNoteAppendArgs(BaseModel): + user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')") + note: str = Field(..., description="Note to append to the user's memory block (e.g., '\\\\n- Cameron is a person')") + + +class XUserNoteReplaceArgs(BaseModel): + user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')") + old_text: str = Field(..., description="Text to find and replace in the user's memory block") + new_text: str = Field(..., description="Text to replace the old_text with") + + +class XUserNoteSetArgs(BaseModel): + user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')") + content: str = Field(..., description="Complete content to set for the user's memory block") + + +class XUserNoteViewArgs(BaseModel): + user_id: str = Field(..., description="X user ID (e.g., '1232326955652931584')") + + def attach_user_blocks(handles: list, agent_state: "AgentState") -> str: """ @@ -386,3 +415,334 @@ def user_note_view(handle: str, agent_state: "AgentState") -> str: raise Exception(f"Error viewing user block: {str(e)}") +# X (Twitter) User Block Management Functions + +def attach_x_user_blocks(user_ids: list, agent_state: "AgentState") -> str: + """ + Attach X user-specific memory blocks to the agent. Creates blocks if they don't exist. + + Args: + user_ids: List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592']) + agent_state: The agent state object containing agent information + + Returns: + String with attachment results for each user ID + """ + logger = logging.getLogger(__name__) + + user_ids = list(set(user_ids)) + + try: + client = get_letta_client() + results = [] + + # Get current blocks using the API + current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) + current_block_labels = set() + + for block in current_blocks: + current_block_labels.add(block.label) + + for user_id in user_ids: + # Create block label with x_user_ prefix + block_label = f"x_user_{user_id}" + + # Skip if already attached + if block_label in current_block_labels: + results.append(f"✓ {user_id}: Already attached") + continue + + # Check if block exists or create new one + try: + blocks = client.blocks.list(label=block_label) + if blocks and len(blocks) > 0: + block = blocks[0] + logger.debug(f"Found existing block: {block_label}") + else: + block = client.blocks.create( + label=block_label, + value=f"# X User: {user_id}\n\nNo information about this user yet.", + limit=5000 + ) + logger.info(f"Created new block: {block_label}") + + # Attach block atomically + client.agents.blocks.attach( + agent_id=str(agent_state.id), + block_id=str(block.id) + ) + + results.append(f"✓ {user_id}: Block attached") + logger.debug(f"Successfully attached block {block_label} to agent") + + except Exception as e: + results.append(f"✗ {user_id}: Error - {str(e)}") + logger.error(f"Error processing block for {user_id}: {e}") + + return f"X user attachment results:\n" + "\n".join(results) + + except Exception as e: + logger.error(f"Error attaching X user blocks: {e}") + raise Exception(f"Error attaching X user blocks: {str(e)}") + + +def detach_x_user_blocks(user_ids: list, agent_state: "AgentState") -> str: + """ + Detach X user-specific memory blocks from the agent. Blocks are preserved for later use. + + Args: + user_ids: List of X user IDs (e.g., ['1232326955652931584', '1950680610282094592']) + agent_state: The agent state object containing agent information + + Returns: + String with detachment results for each user ID + """ + logger = logging.getLogger(__name__) + + try: + client = get_letta_client() + results = [] + + # Build mapping of block labels to IDs using the API + current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) + block_label_to_id = {} + + for block in current_blocks: + block_label_to_id[block.label] = str(block.id) + + # Process each user ID and detach atomically + for user_id in user_ids: + block_label = f"x_user_{user_id}" + + if block_label in block_label_to_id: + try: + # Detach block atomically + client.agents.blocks.detach( + agent_id=str(agent_state.id), + block_id=block_label_to_id[block_label] + ) + results.append(f"✓ {user_id}: Detached") + logger.debug(f"Successfully detached block {block_label} from agent") + except Exception as e: + results.append(f"✗ {user_id}: Error during detachment - {str(e)}") + logger.error(f"Error detaching block {block_label}: {e}") + else: + results.append(f"✗ {user_id}: Not attached") + + return f"X user detachment results:\n" + "\n".join(results) + + except Exception as e: + logger.error(f"Error detaching X user blocks: {e}") + raise Exception(f"Error detaching X user blocks: {str(e)}") + + +def x_user_note_append(user_id: str, note: str, agent_state: "AgentState") -> str: + """ + Append a note to an X user's memory block. Creates the block if it doesn't exist. + + Args: + user_id: X user ID (e.g., '1232326955652931584') + note: Note to append to the user's memory block + agent_state: The agent state object containing agent information + + Returns: + String confirming the note was appended + """ + logger = logging.getLogger(__name__) + + try: + client = get_letta_client() + + block_label = f"x_user_{user_id}" + + # Check if block exists + blocks = client.blocks.list(label=block_label) + + if blocks and len(blocks) > 0: + # Block exists, append to it + block = blocks[0] + current_value = block.value + new_value = current_value + note + + # Update the block + client.blocks.modify( + block_id=str(block.id), + value=new_value + ) + logger.info(f"Appended note to existing block: {block_label}") + return f"✓ Appended note to X user {user_id}'s memory block" + + else: + # Block doesn't exist, create it with the note + initial_value = f"# X User: {user_id}\n\n{note}" + block = client.blocks.create( + label=block_label, + value=initial_value, + limit=5000 + ) + logger.info(f"Created new block with note: {block_label}") + + # Check if block needs to be attached to agent + current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) + current_block_labels = {block.label for block in current_blocks} + + if block_label not in current_block_labels: + # Attach the new block to the agent + client.agents.blocks.attach( + agent_id=str(agent_state.id), + block_id=str(block.id) + ) + logger.info(f"Attached new block to agent: {block_label}") + return f"✓ Created and attached X user {user_id}'s memory block with note" + else: + return f"✓ Created X user {user_id}'s memory block with note" + + except Exception as e: + logger.error(f"Error appending note to X user block: {e}") + raise Exception(f"Error appending note to X user block: {str(e)}") + + +def x_user_note_replace(user_id: str, old_text: str, new_text: str, agent_state: "AgentState") -> str: + """ + Replace text in an X user's memory block. + + Args: + user_id: X user ID (e.g., '1232326955652931584') + old_text: Text to find and replace + new_text: Text to replace the old_text with + agent_state: The agent state object containing agent information + + Returns: + String confirming the text was replaced + """ + logger = logging.getLogger(__name__) + + try: + client = get_letta_client() + + block_label = f"x_user_{user_id}" + + # Check if block exists + blocks = client.blocks.list(label=block_label) + + if not blocks or len(blocks) == 0: + raise Exception(f"No memory block found for X user: {user_id}") + + block = blocks[0] + current_value = block.value + + # Check if old_text exists in the block + if old_text not in current_value: + raise Exception(f"Text '{old_text}' not found in X user {user_id}'s memory block") + + # Replace the text + new_value = current_value.replace(old_text, new_text) + + # Update the block + client.blocks.modify( + block_id=str(block.id), + value=new_value + ) + logger.info(f"Replaced text in block: {block_label}") + return f"✓ Replaced text in X user {user_id}'s memory block" + + except Exception as e: + logger.error(f"Error replacing text in X user block: {e}") + raise Exception(f"Error replacing text in X user block: {str(e)}") + + +def x_user_note_set(user_id: str, content: str, agent_state: "AgentState") -> str: + """ + Set the complete content of an X user's memory block. + + Args: + user_id: X user ID (e.g., '1232326955652931584') + content: Complete content to set for the memory block + agent_state: The agent state object containing agent information + + Returns: + String confirming the content was set + """ + logger = logging.getLogger(__name__) + + try: + client = get_letta_client() + + block_label = f"x_user_{user_id}" + + # Check if block exists + blocks = client.blocks.list(label=block_label) + + if blocks and len(blocks) > 0: + # Block exists, update it + block = blocks[0] + client.blocks.modify( + block_id=str(block.id), + value=content + ) + logger.info(f"Set content for existing block: {block_label}") + return f"✓ Set content for X user {user_id}'s memory block" + + else: + # Block doesn't exist, create it + block = client.blocks.create( + label=block_label, + value=content, + limit=5000 + ) + logger.info(f"Created new block with content: {block_label}") + + # Check if block needs to be attached to agent + current_blocks = client.agents.blocks.list(agent_id=str(agent_state.id)) + current_block_labels = {block.label for block in current_blocks} + + if block_label not in current_block_labels: + # Attach the new block to the agent + client.agents.blocks.attach( + agent_id=str(agent_state.id), + block_id=str(block.id) + ) + logger.info(f"Attached new block to agent: {block_label}") + return f"✓ Created and attached X user {user_id}'s memory block" + else: + return f"✓ Created X user {user_id}'s memory block" + + except Exception as e: + logger.error(f"Error setting X user block content: {e}") + raise Exception(f"Error setting X user block content: {str(e)}") + + +def x_user_note_view(user_id: str, agent_state: "AgentState") -> str: + """ + View the content of an X user's memory block. + + Args: + user_id: X user ID (e.g., '1232326955652931584') + agent_state: The agent state object containing agent information + + Returns: + String containing the user's memory block content + """ + logger = logging.getLogger(__name__) + + try: + client = get_letta_client() + + block_label = f"x_user_{user_id}" + + # Check if block exists + blocks = client.blocks.list(label=block_label) + + if not blocks or len(blocks) == 0: + return f"No memory block found for X user: {user_id}" + + block = blocks[0] + logger.info(f"Retrieved content for block: {block_label}") + + return f"Memory block for X user {user_id}:\n\n{block.value}" + + except Exception as e: + logger.error(f"Error viewing X user block: {e}") + raise Exception(f"Error viewing X user block: {str(e)}") + + diff --git a/x.py b/x.py index 683da7c..0f1007f 100644 --- a/x.py +++ b/x.py @@ -395,13 +395,84 @@ def thread_to_yaml_string(thread_data: Dict) -> str: tweet_obj = { 'text': tweet.get('text'), 'created_at': tweet.get('created_at'), - 'author': author_info + 'author': author_info, + 'author_id': author_id # Include user ID for block management } simplified_thread["conversation"].append(tweet_obj) return yaml.dump(simplified_thread, default_flow_style=False, sort_keys=False) + +def ensure_x_user_blocks_attached(thread_data: Dict, agent_id: str) -> None: + """ + Ensure all users in the thread have their X user blocks attached. + Creates blocks with initial content including their handle if they don't exist. + + Args: + thread_data: Dict with 'tweets' and 'users' keys from get_thread_context() + agent_id: The Letta agent ID to attach blocks to + """ + if not thread_data or "users" not in thread_data: + return + + try: + from tools.blocks import attach_x_user_blocks, x_user_note_set + from config_loader import get_letta_config + from letta_client import Letta + + # Get Letta client + config = get_letta_config() + client = Letta(token=config['api_key'], timeout=config['timeout']) + + # Get agent info to create a mock agent_state for the functions + class MockAgentState: + def __init__(self, agent_id): + self.id = agent_id + + agent_state = MockAgentState(agent_id) + + users_data = thread_data["users"] + user_ids = list(users_data.keys()) + + if not user_ids: + return + + logger.info(f"Ensuring X user blocks for {len(user_ids)} users: {user_ids}") + + # Get current blocks to check which users already have blocks with content + current_blocks = client.agents.blocks.list(agent_id=agent_id) + existing_user_blocks = {} + + for block in current_blocks: + if block.label.startswith("x_user_"): + user_id = block.label.replace("x_user_", "") + existing_user_blocks[user_id] = block + + # Attach all user blocks (this will create missing ones with basic content) + attach_result = attach_x_user_blocks(user_ids, agent_state) + logger.info(f"X user block attachment result: {attach_result}") + + # For newly created blocks, update with user handle information + for user_id in user_ids: + if user_id not in existing_user_blocks: + user_info = users_data[user_id] + username = user_info.get('username', 'unknown') + name = user_info.get('name', 'Unknown') + + # Set initial content with handle information + initial_content = f"# X User: {user_id}\n\n**Handle:** @{username}\n**Name:** {name}\n\nNo additional information about this user yet." + + try: + x_user_note_set(user_id, initial_content, agent_state) + logger.info(f"Set initial content for X user {user_id} (@{username})") + except Exception as e: + logger.error(f"Failed to set initial content for X user {user_id}: {e}") + + except Exception as e: + logger.error(f"Error ensuring X user blocks: {e}") + + # X Caching and Queue System Functions def load_last_seen_id() -> Optional[str]: