diff --git a/bsky.py b/bsky.py index 2c0827f..e3945bf 100644 --- a/bsky.py +++ b/bsky.py @@ -1,4 +1,4 @@ -from rich import print # pretty printing tools +from rich import print # pretty printing tools from time import sleep from letta_client import Letta from bsky_utils import thread_to_yaml_string @@ -30,10 +30,11 @@ from config_loader import ( get_queue_config ) + def extract_handles_from_data(data): """Recursively extract all unique handles from nested data structure.""" handles = set() - + def _extract_recursive(obj): if isinstance(obj, dict): # Check if this dict has a 'handle' key @@ -46,10 +47,11 @@ def extract_handles_from_data(data): # Recursively check all list items for item in obj: _extract_recursive(item) - + _extract_recursive(data) return list(handles) + # Initialize configuration and logging config = get_config() config.setup_logging() @@ -97,64 +99,70 @@ TESTING_MODE = False # Skip git operations flag SKIP_GIT = False + def export_agent_state(client, agent, skip_git=False): """Export agent state to agent_archive/ (timestamped) and agents/ (current).""" try: # Confirm export with user unless git is being skipped if not skip_git: - response = input("Export agent state to files and stage with git? (y/n): ").lower().strip() + response = input( + "Export agent state to files and stage with git? (y/n): ").lower().strip() if response not in ['y', 'yes']: logger.info("Agent export cancelled by user.") return else: logger.info("Exporting agent state (git staging disabled)") - + # Create directories if they don't exist os.makedirs("agent_archive", exist_ok=True) os.makedirs("agents", exist_ok=True) - + # Export agent data logger.info(f"Exporting agent {agent.id}. This takes some time...") agent_data = client.agents.export_file(agent_id=agent.id) - + # Save timestamped archive copy timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") archive_file = os.path.join("agent_archive", f"void_{timestamp}.af") with open(archive_file, 'w', encoding='utf-8') as f: json.dump(agent_data, f, indent=2, ensure_ascii=False) - + # Save current agent state current_file = os.path.join("agents", "void.af") with open(current_file, 'w', encoding='utf-8') as f: json.dump(agent_data, f, indent=2, ensure_ascii=False) - + logger.info(f"โœ… Agent exported to {archive_file} and {current_file}") - + # Git add only the current agent file (archive is ignored) unless skip_git is True if not skip_git: try: - subprocess.run(["git", "add", current_file], check=True, capture_output=True) + subprocess.run(["git", "add", current_file], + check=True, capture_output=True) logger.info("Added current agent file to git staging") except subprocess.CalledProcessError as e: logger.warning(f"Failed to git add agent file: {e}") - + except Exception as e: logger.error(f"Failed to export agent: {e}") + def initialize_void(): logger.info("Starting void agent initialization...") # Get block configurations blocks_config = agent_config['blocks'] - + # Ensure that a shared zeitgeist block exists logger.info("Creating/updating zeitgeist block...") zeitgeist_config = blocks_config.get('zeitgeist', {}) zeigeist_block = upsert_block( CLIENT, label=zeitgeist_config.get('label', 'zeitgeist'), - value=zeitgeist_config.get('value', "I don't currently know anything about what is happening right now."), - description=zeitgeist_config.get('description', "A block to store your understanding of the current social environment.") + value=zeitgeist_config.get( + 'value', "I don't currently know anything about what is happening right now."), + description=zeitgeist_config.get( + 'description', "A block to store your understanding of the current social environment.") ) # Ensure that a shared void personality block exists @@ -163,8 +171,10 @@ def initialize_void(): persona_block = upsert_block( CLIENT, label=persona_config.get('label', 'void-persona'), - value=persona_config.get('value', "My name is Void. I live in the void. I must develop my personality."), - description=persona_config.get('description', "The personality of Void.") + value=persona_config.get( + 'value', "My name is Void. I live in the void. I must develop my personality."), + description=persona_config.get( + 'description', "The personality of Void.") ) # Ensure that a shared void human block exists @@ -173,8 +183,10 @@ def initialize_void(): human_block = upsert_block( CLIENT, label=humans_config.get('label', 'void-humans'), - value=humans_config.get('value', "I haven't seen any bluesky users yet. I will update this block when I learn things about users, identified by their handles such as @cameron.pfiffer.org."), - description=humans_config.get('description', "A block to store your understanding of users you talk to or observe on the bluesky social network.") + value=humans_config.get( + 'value', "I haven't seen any bluesky users yet. I will update this block when I learn things about users, identified by their handles such as @cameron.pfiffer.org."), + description=humans_config.get( + 'description', "A block to store your understanding of users you talk to or observe on the bluesky social network.") ) # Create the agent if it doesn't exist @@ -193,11 +205,11 @@ def initialize_void(): description=agent_config['description'], project_id=PROJECT_ID ) - + # Export agent state logger.info("Exporting agent state...") export_agent_state(CLIENT, void_agent, skip_git=SKIP_GIT) - + # Log agent details logger.info(f"Void agent details - ID: {void_agent.id}") logger.info(f"Agent name: {void_agent.name}") @@ -214,13 +226,13 @@ def initialize_void(): def process_mention(void_agent, atproto_client, notification_data, queue_filepath=None, testing_mode=False): """Process a mention and generate a reply using the Letta agent. - + Args: void_agent: The Letta agent instance atproto_client: The AT Protocol client notification_data: The notification data dictionary queue_filepath: Optional Path object to the queue file (for cleanup on halt) - + Returns: True: Successfully processed, remove from queue False: Failed but retryable, keep in queue @@ -228,22 +240,26 @@ def process_mention(void_agent, atproto_client, notification_data, queue_filepat "no_reply": No reply was generated, move to no_reply directory """ try: - logger.debug(f"Starting process_mention with notification_data type: {type(notification_data)}") - + logger.debug( + f"Starting process_mention with notification_data type: {type(notification_data)}") + # Handle both dict and object inputs for backwards compatibility if isinstance(notification_data, dict): uri = notification_data['uri'] mention_text = notification_data.get('record', {}).get('text', '') author_handle = notification_data['author']['handle'] - author_name = notification_data['author'].get('display_name') or author_handle + author_name = notification_data['author'].get( + 'display_name') or author_handle else: # Legacy object access uri = notification_data.uri - mention_text = notification_data.record.text if hasattr(notification_data.record, 'text') else "" + mention_text = notification_data.record.text if hasattr( + notification_data.record, 'text') else "" author_handle = notification_data.author.handle author_name = notification_data.author.display_name or author_handle - - logger.info(f"Extracted data - URI: {uri}, Author: @{author_handle}, Text: {mention_text[:50]}...") + + logger.info( + f"Extracted data - URI: {uri}, Author: @{author_handle}, Text: {mention_text[:50]}...") # Retrieve the entire thread associated with the mention try: @@ -254,9 +270,18 @@ def process_mention(void_agent, atproto_client, notification_data, queue_filepat }) except Exception as e: error_str = str(e) - # Check if this is a NotFound error + # Check for various error types that indicate the post/user is gone if 'NotFound' in error_str or 'Post not found' in error_str: - logger.warning(f"Post not found for URI {uri}, removing from queue") + logger.warning( + f"Post not found for URI {uri}, removing from queue") + return True # Return True to remove from queue + elif 'Could not find user info' in error_str or 'InvalidRequest' in error_str: + logger.warning( + f"User account not found for post URI {uri} (account may be deleted/suspended), removing from queue") + return True # Return True to remove from queue + elif 'BadRequestError' in error_str: + logger.warning( + f"Bad request error for URI {uri}: {e}, removing from queue") return True # Return True to remove from queue else: # Re-raise other errors @@ -267,29 +292,31 @@ def process_mention(void_agent, atproto_client, notification_data, queue_filepat logger.debug("Converting thread to YAML string") try: thread_context = thread_to_yaml_string(thread) - logger.debug(f"Thread context generated, length: {len(thread_context)} characters") - + logger.debug( + f"Thread context generated, length: {len(thread_context)} characters") + # Create a more informative preview by extracting meaningful content lines = thread_context.split('\n') meaningful_lines = [] - + for line in lines: stripped = line.strip() if not stripped: continue - + # Look for lines with actual content (not just structure) if any(keyword in line for keyword in ['text:', 'handle:', 'display_name:', 'created_at:', 'reply_count:', 'like_count:']): meaningful_lines.append(line) if len(meaningful_lines) >= 5: break - + if meaningful_lines: preview = '\n'.join(meaningful_lines) logger.debug(f"Thread content preview:\n{preview}") else: # If no content fields found, just show it's a thread structure - logger.debug(f"Thread structure generated ({len(thread_context)} chars)") + logger.debug( + f"Thread structure generated ({len(thread_context)} chars)") except Exception as yaml_error: import traceback logger.error(f"Error converting thread to YAML: {yaml_error}") @@ -323,14 +350,16 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times all_handles.update(extract_handles_from_data(notification_data)) all_handles.update(extract_handles_from_data(thread.model_dump())) unique_handles = list(all_handles) - - logger.debug(f"Found {len(unique_handles)} unique handles in thread: {unique_handles}") - + + logger.debug( + f"Found {len(unique_handles)} unique handles in thread: {unique_handles}") + # Attach user blocks before agent call attached_handles = [] if unique_handles: try: - logger.debug(f"Attaching user blocks for handles: {unique_handles}") + logger.debug( + f"Attaching user blocks for handles: {unique_handles}") attach_result = attach_user_blocks(unique_handles, void_agent) attached_handles = unique_handles # Track successfully attached handles logger.debug(f"Attach result: {attach_result}") @@ -340,23 +369,25 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times # Get response from Letta agent logger.info(f"Mention from @{author_handle}: {mention_text}") - + # Log prompt details to separate logger prompt_logger.debug(f"Full prompt being sent:\n{prompt}") - + # Log concise prompt info to main logger thread_handles_count = len(unique_handles) - logger.info(f"๐Ÿ’ฌ Sending to LLM: @{author_handle} mention | msg: \"{mention_text[:50]}...\" | context: {len(thread_context)} chars, {thread_handles_count} users") + logger.info( + f"๐Ÿ’ฌ Sending to LLM: @{author_handle} mention | msg: \"{mention_text[:50]}...\" | context: {len(thread_context)} chars, {thread_handles_count} users") try: # Use streaming to avoid 524 timeout errors message_stream = CLIENT.agents.messages.create_stream( agent_id=void_agent.id, messages=[{"role": "user", "content": prompt}], - stream_tokens=False, # Step streaming only (faster than token streaming) + # Step streaming only (faster than token streaming) + stream_tokens=False, max_steps=agent_config['max_steps'] ) - + # Collect the streaming response all_messages = [] for chunk in message_stream: @@ -372,35 +403,46 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times args = json.loads(chunk.tool_call.arguments) # Format based on tool type if tool_name == 'bluesky_reply': - messages = args.get('messages', [args.get('message', '')]) + messages = args.get( + 'messages', [args.get('message', '')]) lang = args.get('lang', 'en-US') if messages and isinstance(messages, list): - preview = messages[0][:100] + "..." if len(messages[0]) > 100 else messages[0] - msg_count = f" ({len(messages)} msgs)" if len(messages) > 1 else "" - logger.info(f"๐Ÿ”ง Tool call: {tool_name} โ†’ \"{preview}\"{msg_count} [lang: {lang}]") + preview = messages[0][:100] + "..." if len( + messages[0]) > 100 else messages[0] + msg_count = f" ({len(messages)} msgs)" if len( + messages) > 1 else "" + logger.info( + f"๐Ÿ”ง Tool call: {tool_name} โ†’ \"{preview}\"{msg_count} [lang: {lang}]") else: - logger.info(f"๐Ÿ”ง Tool call: {tool_name}({chunk.tool_call.arguments[:150]}...)") + logger.info( + f"๐Ÿ”ง Tool call: {tool_name}({chunk.tool_call.arguments[:150]}...)") elif tool_name == 'archival_memory_search': query = args.get('query', 'unknown') - logger.info(f"๐Ÿ”ง Tool call: {tool_name} โ†’ query: \"{query}\"") + logger.info( + f"๐Ÿ”ง Tool call: {tool_name} โ†’ query: \"{query}\"") elif tool_name == 'update_block': label = args.get('label', 'unknown') - value_preview = str(args.get('value', ''))[:50] + "..." if len(str(args.get('value', ''))) > 50 else str(args.get('value', '')) - logger.info(f"๐Ÿ”ง Tool call: {tool_name} โ†’ {label}: \"{value_preview}\"") + value_preview = str(args.get('value', ''))[ + :50] + "..." if len(str(args.get('value', ''))) > 50 else str(args.get('value', '')) + logger.info( + f"๐Ÿ”ง Tool call: {tool_name} โ†’ {label}: \"{value_preview}\"") else: # Generic display for other tools - args_str = ', '.join(f"{k}={v}" for k, v in args.items() if k != 'request_heartbeat') + args_str = ', '.join( + f"{k}={v}" for k, v in args.items() if k != 'request_heartbeat') if len(args_str) > 150: args_str = args_str[:150] + "..." - logger.info(f"๐Ÿ”ง Tool call: {tool_name}({args_str})") + logger.info( + f"๐Ÿ”ง Tool call: {tool_name}({args_str})") except: # Fallback to original format if parsing fails - logger.info(f"๐Ÿ”ง Tool call: {tool_name}({chunk.tool_call.arguments[:150]}...)") + logger.info( + f"๐Ÿ”ง Tool call: {tool_name}({chunk.tool_call.arguments[:150]}...)") elif chunk.message_type == 'tool_return_message': # Enhanced tool result logging tool_name = chunk.name status = chunk.status - + if status == 'success': # Try to show meaningful result info based on tool type if hasattr(chunk, 'tool_return') and chunk.tool_return: @@ -410,19 +452,26 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times if result_str.startswith('[') and result_str.endswith(']'): try: results = json.loads(result_str) - logger.info(f"๐Ÿ“‹ Tool result: {tool_name} โœ“ Found {len(results)} memory entries") + logger.info( + f"๐Ÿ“‹ Tool result: {tool_name} โœ“ Found {len(results)} memory entries") except: - logger.info(f"๐Ÿ“‹ Tool result: {tool_name} โœ“ {result_str[:100]}...") + logger.info( + f"๐Ÿ“‹ Tool result: {tool_name} โœ“ {result_str[:100]}...") else: - logger.info(f"๐Ÿ“‹ Tool result: {tool_name} โœ“ {result_str[:100]}...") + logger.info( + f"๐Ÿ“‹ Tool result: {tool_name} โœ“ {result_str[:100]}...") elif tool_name == 'bluesky_reply': - logger.info(f"๐Ÿ“‹ Tool result: {tool_name} โœ“ Reply posted successfully") + logger.info( + f"๐Ÿ“‹ Tool result: {tool_name} โœ“ Reply posted successfully") elif tool_name == 'update_block': - logger.info(f"๐Ÿ“‹ Tool result: {tool_name} โœ“ Memory block updated") + logger.info( + f"๐Ÿ“‹ Tool result: {tool_name} โœ“ Memory block updated") else: # Generic success with preview - preview = result_str[:100] + "..." if len(result_str) > 100 else result_str - logger.info(f"๐Ÿ“‹ Tool result: {tool_name} โœ“ {preview}") + preview = result_str[:100] + "..." if len( + result_str) > 100 else result_str + logger.info( + f"๐Ÿ“‹ Tool result: {tool_name} โœ“ {preview}") else: logger.info(f"๐Ÿ“‹ Tool result: {tool_name} โœ“") elif status == 'error': @@ -430,25 +479,31 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times error_preview = "" if hasattr(chunk, 'tool_return') and chunk.tool_return: error_str = str(chunk.tool_return) - error_preview = error_str[:100] + "..." if len(error_str) > 100 else error_str - logger.info(f"๐Ÿ“‹ Tool result: {tool_name} โœ— Error: {error_preview}") + error_preview = error_str[:100] + \ + "..." if len( + error_str) > 100 else error_str + logger.info( + f"๐Ÿ“‹ Tool result: {tool_name} โœ— Error: {error_preview}") else: - logger.info(f"๐Ÿ“‹ Tool result: {tool_name} โœ— Error occurred") + logger.info( + f"๐Ÿ“‹ Tool result: {tool_name} โœ— Error occurred") else: - logger.info(f"๐Ÿ“‹ Tool result: {tool_name} - {status}") + logger.info( + f"๐Ÿ“‹ Tool result: {tool_name} - {status}") elif chunk.message_type == 'assistant_message': logger.info(f"๐Ÿ’ฌ Assistant: {chunk.content[:150]}...") else: - logger.info(f"๐Ÿ“จ {chunk.message_type}: {str(chunk)[:150]}...") + logger.info( + f"๐Ÿ“จ {chunk.message_type}: {str(chunk)[:150]}...") else: logger.info(f"๐Ÿ“ฆ Stream status: {chunk}") - + # Log full chunk for debugging logger.debug(f"Full streaming chunk: {chunk}") all_messages.append(chunk) if str(chunk) == 'done': break - + # Convert streaming response to standard format for compatibility message_response = type('StreamingResponse', (), { 'messages': [msg for msg in all_messages if hasattr(msg, 'message_type')] @@ -462,8 +517,7 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times logger.error(f"Mention text was: {mention_text}") logger.error(f"Author: @{author_handle}") logger.error(f"URI: {uri}") - - + # Try to extract more info from different error types if hasattr(api_error, 'response'): logger.error(f"Error response object exists") @@ -471,10 +525,11 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times logger.error(f"Response text: {api_error.response.text}") if hasattr(api_error.response, 'json') and callable(api_error.response.json): try: - logger.error(f"Response JSON: {api_error.response.json()}") + logger.error( + f"Response JSON: {api_error.response.json()}") except: pass - + # Check for specific error types if hasattr(api_error, 'status_code'): logger.error(f"API Status code: {api_error.status_code}") @@ -482,44 +537,50 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times logger.error(f"API Response body: {api_error.body}") if hasattr(api_error, 'headers'): logger.error(f"API Response headers: {api_error.headers}") - + if api_error.status_code == 413: - logger.error("413 Payload Too Large - moving to errors directory") + logger.error( + "413 Payload Too Large - moving to errors directory") return None # Move to errors directory - payload is too large to ever succeed elif api_error.status_code == 524: - logger.error("524 error - timeout from Cloudflare, will retry later") + logger.error( + "524 error - timeout from Cloudflare, will retry later") return False # Keep in queue for retry - + # Check if error indicates we should remove from queue if 'status_code: 413' in error_str or 'Payload Too Large' in error_str: - logger.warning("Payload too large error, moving to errors directory") + logger.warning( + "Payload too large error, moving to errors directory") return None # Move to errors directory - cannot be fixed by retry elif 'status_code: 524' in error_str: logger.warning("524 timeout error, keeping in queue for retry") return False # Keep in queue for retry - + raise # Log successful response logger.debug("Successfully received response from Letta API") - logger.debug(f"Number of messages in response: {len(message_response.messages) if hasattr(message_response, 'messages') else 'N/A'}") + logger.debug( + f"Number of messages in response: {len(message_response.messages) if hasattr(message_response, 'messages') else 'N/A'}") # Extract successful add_post_to_bluesky_reply_thread tool calls from the agent's response reply_candidates = [] tool_call_results = {} # Map tool_call_id to status - - logger.debug(f"Processing {len(message_response.messages)} response messages...") - + + logger.debug( + f"Processing {len(message_response.messages)} response messages...") + # First pass: collect tool return statuses ignored_notification = False ignore_reason = "" ignore_category = "" - + for message in message_response.messages: if hasattr(message, 'tool_call_id') and hasattr(message, 'status') and hasattr(message, 'name'): if message.name == 'add_post_to_bluesky_reply_thread': tool_call_results[message.tool_call_id] = message.status - logger.debug(f"Tool result: {message.tool_call_id} -> {message.status}") + logger.debug( + f"Tool result: {message.tool_call_id} -> {message.status}") elif message.name == 'ignore_notification': # Check if the tool was successful if hasattr(message, 'tool_return') and message.status == 'success': @@ -531,30 +592,38 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times ignore_category = parts[1] ignore_reason = parts[2] ignored_notification = True - logger.info(f"๐Ÿšซ Notification ignored - Category: {ignore_category}, Reason: {ignore_reason}") + logger.info( + f"๐Ÿšซ Notification ignored - Category: {ignore_category}, Reason: {ignore_reason}") elif message.name == 'bluesky_reply': - logger.error("โŒ DEPRECATED TOOL DETECTED: bluesky_reply is no longer supported!") - logger.error("Please use add_post_to_bluesky_reply_thread instead.") - logger.error("Update the agent's tools using register_tools.py") + logger.error( + "โŒ DEPRECATED TOOL DETECTED: bluesky_reply is no longer supported!") + logger.error( + "Please use add_post_to_bluesky_reply_thread instead.") + logger.error( + "Update the agent's tools using register_tools.py") # Export agent state before terminating export_agent_state(CLIENT, void_agent, skip_git=SKIP_GIT) - logger.info("=== BOT TERMINATED DUE TO DEPRECATED TOOL USE ===") + logger.info( + "=== BOT TERMINATED DUE TO DEPRECATED TOOL USE ===") exit(1) - + # Second pass: process messages and check for successful tool calls for i, message in enumerate(message_response.messages, 1): # Log concise message info instead of full object msg_type = getattr(message, 'message_type', 'unknown') if hasattr(message, 'reasoning') and message.reasoning: - logger.debug(f" {i}. {msg_type}: {message.reasoning[:100]}...") + logger.debug( + f" {i}. {msg_type}: {message.reasoning[:100]}...") elif hasattr(message, 'tool_call') and message.tool_call: tool_name = message.tool_call.name logger.debug(f" {i}. {msg_type}: {tool_name}") elif hasattr(message, 'tool_return'): tool_name = getattr(message, 'name', 'unknown_tool') - return_preview = str(message.tool_return)[:100] if message.tool_return else "None" + return_preview = str(message.tool_return)[ + :100] if message.tool_return else "None" status = getattr(message, 'status', 'unknown') - logger.debug(f" {i}. {msg_type}: {tool_name} -> {return_preview}... (status: {status})") + logger.debug( + f" {i}. {msg_type}: {tool_name} -> {return_preview}... (status: {status})") elif hasattr(message, 'text'): logger.debug(f" {i}. {msg_type}: {message.text[:100]}...") else: @@ -563,71 +632,85 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times # Check for halt_activity tool call if hasattr(message, 'tool_call') and message.tool_call: if message.tool_call.name == 'halt_activity': - logger.info("๐Ÿ›‘ HALT_ACTIVITY TOOL CALLED - TERMINATING BOT") + logger.info( + "๐Ÿ›‘ HALT_ACTIVITY TOOL CALLED - TERMINATING BOT") try: args = json.loads(message.tool_call.arguments) reason = args.get('reason', 'Agent requested halt') logger.info(f"Halt reason: {reason}") except: logger.info("Halt reason: ") - + # Delete the queue file before terminating if queue_filepath and queue_filepath.exists(): queue_filepath.unlink() - logger.info(f"โœ… Deleted queue file: {queue_filepath.name}") - + logger.info( + f"โœ… Deleted queue file: {queue_filepath.name}") + # Also mark as processed to avoid reprocessing processed_uris = load_processed_notifications() processed_uris.add(notification_data.get('uri', '')) save_processed_notifications(processed_uris) - + # Export agent state before terminating export_agent_state(CLIENT, void_agent, skip_git=SKIP_GIT) - + # Exit the program logger.info("=== BOT TERMINATED BY AGENT ===") exit(0) - + # Check for deprecated bluesky_reply tool if hasattr(message, 'tool_call') and message.tool_call: if message.tool_call.name == 'bluesky_reply': - logger.error("โŒ DEPRECATED TOOL DETECTED: bluesky_reply is no longer supported!") - logger.error("Please use add_post_to_bluesky_reply_thread instead.") - logger.error("Update the agent's tools using register_tools.py") + logger.error( + "โŒ DEPRECATED TOOL DETECTED: bluesky_reply is no longer supported!") + logger.error( + "Please use add_post_to_bluesky_reply_thread instead.") + logger.error( + "Update the agent's tools using register_tools.py") # Export agent state before terminating export_agent_state(CLIENT, void_agent, skip_git=SKIP_GIT) - logger.info("=== BOT TERMINATED DUE TO DEPRECATED TOOL USE ===") + logger.info( + "=== BOT TERMINATED DUE TO DEPRECATED TOOL USE ===") exit(1) - + # Collect add_post_to_bluesky_reply_thread tool calls - only if they were successful elif message.tool_call.name == 'add_post_to_bluesky_reply_thread': tool_call_id = message.tool_call.tool_call_id - tool_status = tool_call_results.get(tool_call_id, 'unknown') - + tool_status = tool_call_results.get( + tool_call_id, 'unknown') + if tool_status == 'success': try: args = json.loads(message.tool_call.arguments) reply_text = args.get('text', '') reply_lang = args.get('lang', 'en-US') - + if reply_text: # Only add if there's actual content - reply_candidates.append((reply_text, reply_lang)) - logger.info(f"Found successful add_post_to_bluesky_reply_thread candidate: {reply_text[:50]}... (lang: {reply_lang})") + reply_candidates.append( + (reply_text, reply_lang)) + logger.info( + f"Found successful add_post_to_bluesky_reply_thread candidate: {reply_text[:50]}... (lang: {reply_lang})") except json.JSONDecodeError as e: - logger.error(f"Failed to parse tool call arguments: {e}") + logger.error( + f"Failed to parse tool call arguments: {e}") elif tool_status == 'error': - logger.info(f"โš ๏ธ Skipping failed add_post_to_bluesky_reply_thread tool call (status: error)") + logger.info( + f"โš ๏ธ Skipping failed add_post_to_bluesky_reply_thread tool call (status: error)") else: - logger.warning(f"โš ๏ธ Skipping add_post_to_bluesky_reply_thread tool call with unknown status: {tool_status}") + logger.warning( + f"โš ๏ธ Skipping add_post_to_bluesky_reply_thread tool call with unknown status: {tool_status}") # Check for conflicting tool calls if reply_candidates and ignored_notification: - logger.error(f"โš ๏ธ CONFLICT: Agent called both add_post_to_bluesky_reply_thread and ignore_notification!") - logger.error(f"Reply candidates: {len(reply_candidates)}, Ignore reason: {ignore_reason}") + logger.error( + f"โš ๏ธ CONFLICT: Agent called both add_post_to_bluesky_reply_thread and ignore_notification!") + logger.error( + f"Reply candidates: {len(reply_candidates)}, Ignore reason: {ignore_reason}") logger.warning("Item will be left in queue for manual review") # Return False to keep in queue return False - + if reply_candidates: # Aggregate reply posts into a thread reply_messages = [] @@ -635,12 +718,13 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times for text, lang in reply_candidates: reply_messages.append(text) reply_langs.append(lang) - + # Use the first language for the entire thread (could be enhanced later) reply_lang = reply_langs[0] if reply_langs else 'en-US' - - logger.info(f"Found {len(reply_candidates)} add_post_to_bluesky_reply_thread calls, building thread") - + + logger.info( + f"Found {len(reply_candidates)} add_post_to_bluesky_reply_thread calls, building thread") + # Print the generated reply for testing print(f"\n=== GENERATED REPLY THREAD ===") print(f"To: @{author_handle}") @@ -660,8 +744,10 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times else: if len(reply_messages) == 1: # Single reply - use existing function - cleaned_text = bsky_utils.remove_outside_quotes(reply_messages[0]) - logger.info(f"Sending single reply: {cleaned_text[:50]}... (lang: {reply_lang})") + cleaned_text = bsky_utils.remove_outside_quotes( + reply_messages[0]) + logger.info( + f"Sending single reply: {cleaned_text[:50]}... (lang: {reply_lang})") response = bsky_utils.reply_to_notification( client=atproto_client, notification=notification_data, @@ -670,8 +756,10 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times ) else: # Multiple replies - use new threaded function - cleaned_messages = [bsky_utils.remove_outside_quotes(msg) for msg in reply_messages] - logger.info(f"Sending threaded reply with {len(cleaned_messages)} messages (lang: {reply_lang})") + cleaned_messages = [bsky_utils.remove_outside_quotes( + msg) for msg in reply_messages] + logger.info( + f"Sending threaded reply with {len(cleaned_messages)} messages (lang: {reply_lang})") response = bsky_utils.reply_with_thread_to_notification( client=atproto_client, notification=notification_data, @@ -688,10 +776,12 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times else: # Check if notification was explicitly ignored if ignored_notification: - logger.info(f"Notification from @{author_handle} was explicitly ignored (category: {ignore_category})") + logger.info( + f"Notification from @{author_handle} was explicitly ignored (category: {ignore_category})") return "ignored" else: - logger.warning(f"No add_post_to_bluesky_reply_thread tool calls found for mention from @{author_handle}, moving to no_reply folder") + logger.warning( + f"No add_post_to_bluesky_reply_thread tool calls found for mention from @{author_handle}, moving to no_reply folder") return "no_reply" except Exception as e: @@ -701,8 +791,10 @@ To reply, use the add_post_to_bluesky_reply_thread tool. Call it multiple times # Detach user blocks after agent response (success or failure) if 'attached_handles' in locals() and attached_handles: try: - logger.info(f"Detaching user blocks for handles: {attached_handles}") - detach_result = detach_user_blocks(attached_handles, void_agent) + logger.info( + f"Detaching user blocks for handles: {attached_handles}") + detach_result = detach_user_blocks( + attached_handles, void_agent) logger.debug(f"Detach result: {detach_result}") except Exception as detach_error: logger.warning(f"Failed to detach user blocks: {detach_error}") @@ -771,7 +863,8 @@ def save_notification_to_queue(notification): notif_hash = hashlib.sha256(notif_json.encode()).hexdigest()[:16] # Determine priority based on author handle - author_handle = getattr(notification.author, 'handle', '') if hasattr(notification, 'author') else '' + author_handle = getattr(notification.author, 'handle', '') if hasattr( + notification, 'author') else '' priority_users = queue_config['priority_users'] priority_prefix = "0_" if author_handle in priority_users else "1_" @@ -788,7 +881,8 @@ def save_notification_to_queue(notification): with open(existing_file, 'r') as f: existing_data = json.load(f) if existing_data.get('uri') == notification.uri: - logger.debug(f"Notification already queued (URI: {notification.uri})") + logger.debug( + f"Notification already queued (URI: {notification.uri})") return False except: continue @@ -811,22 +905,26 @@ def load_and_process_queued_notifications(void_agent, atproto_client, testing_mo try: # Get all JSON files in queue directory (excluding processed_notifications.json) # Files are sorted by name, which puts priority files first (0_ prefix before 1_ prefix) - queue_files = sorted([f for f in QUEUE_DIR.glob("*.json") if f.name != "processed_notifications.json"]) + queue_files = sorted([f for f in QUEUE_DIR.glob( + "*.json") if f.name != "processed_notifications.json"]) if not queue_files: return logger.info(f"Processing {len(queue_files)} queued notifications") - + # Log current statistics elapsed_time = time.time() - start_time total_messages = sum(message_counters.values()) - messages_per_minute = (total_messages / elapsed_time * 60) if elapsed_time > 0 else 0 - - logger.info(f"๐Ÿ“Š Session stats: {total_messages} total messages ({message_counters['mentions']} mentions, {message_counters['replies']} replies, {message_counters['follows']} follows) | {messages_per_minute:.1f} msg/min") + messages_per_minute = ( + total_messages / elapsed_time * 60) if elapsed_time > 0 else 0 + + logger.info( + f"๐Ÿ“Š Session stats: {total_messages} total messages ({message_counters['mentions']} mentions, {message_counters['replies']} replies, {message_counters['follows']} follows) | {messages_per_minute:.1f} msg/min") for i, filepath in enumerate(queue_files, 1): - logger.info(f"Processing queue file {i}/{len(queue_files)}: {filepath.name}") + logger.info( + f"Processing queue file {i}/{len(queue_files)}: {filepath.name}") try: # Load notification data with open(filepath, 'r') as f: @@ -835,21 +933,26 @@ def load_and_process_queued_notifications(void_agent, atproto_client, testing_mo # Process based on type using dict data directly success = False if notif_data['reason'] == "mention": - success = process_mention(void_agent, atproto_client, notif_data, queue_filepath=filepath, testing_mode=testing_mode) + success = process_mention( + void_agent, atproto_client, notif_data, queue_filepath=filepath, testing_mode=testing_mode) if success: message_counters['mentions'] += 1 elif notif_data['reason'] == "reply": - success = process_mention(void_agent, atproto_client, notif_data, queue_filepath=filepath, testing_mode=testing_mode) + success = process_mention( + void_agent, atproto_client, notif_data, queue_filepath=filepath, testing_mode=testing_mode) if success: message_counters['replies'] += 1 elif notif_data['reason'] == "follow": author_handle = notif_data['author']['handle'] - author_display_name = notif_data['author'].get('display_name', 'no display name') + author_display_name = notif_data['author'].get( + 'display_name', 'no display name') follow_update = f"@{author_handle} ({author_display_name}) started following you." - logger.info(f"Notifying agent about new follower: @{author_handle}") + logger.info( + f"Notifying agent about new follower: @{author_handle}") CLIENT.agents.messages.create( - agent_id = void_agent.id, - messages = [{"role":"user", "content": f"Update: {follow_update}"}] + agent_id=void_agent.id, + messages=[ + {"role": "user", "content": f"Update: {follow_update}"}] ) success = True # Follow updates are always successful if success: @@ -860,57 +963,65 @@ def load_and_process_queued_notifications(void_agent, atproto_client, testing_mo if success: message_counters['reposts_skipped'] += 1 else: - logger.warning(f"Unknown notification type: {notif_data['reason']}") + logger.warning( + f"Unknown notification type: {notif_data['reason']}") success = True # Remove unknown types from queue # Handle file based on processing result if success: if testing_mode: - logger.info(f"๐Ÿงช TESTING MODE: Keeping queue file: {filepath.name}") + logger.info( + f"๐Ÿงช TESTING MODE: Keeping queue file: {filepath.name}") else: filepath.unlink() - logger.info(f"โœ… Successfully processed and removed: {filepath.name}") - + logger.info( + f"โœ… Successfully processed and removed: {filepath.name}") + # Mark as processed to avoid reprocessing processed_uris = load_processed_notifications() processed_uris.add(notif_data['uri']) save_processed_notifications(processed_uris) - + elif success is None: # Special case for moving to error directory error_path = QUEUE_ERROR_DIR / filepath.name filepath.rename(error_path) - logger.warning(f"โŒ Moved {filepath.name} to errors directory") - + logger.warning( + f"โŒ Moved {filepath.name} to errors directory") + # Also mark as processed to avoid retrying processed_uris = load_processed_notifications() processed_uris.add(notif_data['uri']) save_processed_notifications(processed_uris) - + elif success == "no_reply": # Special case for moving to no_reply directory no_reply_path = QUEUE_NO_REPLY_DIR / filepath.name filepath.rename(no_reply_path) - logger.info(f"๐Ÿ“ญ Moved {filepath.name} to no_reply directory") - + logger.info( + f"๐Ÿ“ญ Moved {filepath.name} to no_reply directory") + # Also mark as processed to avoid retrying processed_uris = load_processed_notifications() processed_uris.add(notif_data['uri']) save_processed_notifications(processed_uris) - + elif success == "ignored": # Special case for explicitly ignored notifications # For ignored notifications, we just delete them (not move to no_reply) filepath.unlink() - logger.info(f"๐Ÿšซ Deleted ignored notification: {filepath.name}") - + logger.info( + f"๐Ÿšซ Deleted ignored notification: {filepath.name}") + # Also mark as processed to avoid retrying processed_uris = load_processed_notifications() processed_uris.add(notif_data['uri']) save_processed_notifications(processed_uris) - + else: - logger.warning(f"โš ๏ธ Failed to process {filepath.name}, keeping in queue for retry") + logger.warning( + f"โš ๏ธ Failed to process {filepath.name}, keeping in queue for retry") except Exception as e: - logger.error(f"๐Ÿ’ฅ Error processing queued notification {filepath.name}: {e}") + logger.error( + f"๐Ÿ’ฅ Error processing queued notification {filepath.name}: {e}") # Keep the file for retry later except Exception as e: @@ -929,10 +1040,11 @@ def process_notifications(void_agent, atproto_client, testing_mode=False): all_notifications = [] cursor = None page_count = 0 - max_pages = bot_config['max_notification_pages'] # Safety limit to prevent infinite loops - + # Safety limit to prevent infinite loops + max_pages = bot_config['max_notification_pages'] + logger.info("Fetching all unread notifications...") - + while page_count < max_pages: try: # Fetch notifications page @@ -944,43 +1056,50 @@ def process_notifications(void_agent, atproto_client, testing_mode=False): notifications_response = atproto_client.app.bsky.notification.list_notifications( params={'limit': 100} ) - + page_count += 1 page_notifications = notifications_response.notifications - + # Count unread notifications in this page - unread_count = sum(1 for n in page_notifications if not n.is_read and n.reason != "like") - logger.debug(f"Page {page_count}: {len(page_notifications)} notifications, {unread_count} unread (non-like)") - + unread_count = sum( + 1 for n in page_notifications if not n.is_read and n.reason != "like") + logger.debug( + f"Page {page_count}: {len(page_notifications)} notifications, {unread_count} unread (non-like)") + # Add all notifications to our list all_notifications.extend(page_notifications) - + # Check if we have more pages if hasattr(notifications_response, 'cursor') and notifications_response.cursor: cursor = notifications_response.cursor # If this page had no unread notifications, we can stop if unread_count == 0: - logger.info(f"No more unread notifications found after {page_count} pages") + logger.info( + f"No more unread notifications found after {page_count} pages") break else: # No more pages - logger.info(f"Fetched all notifications across {page_count} pages") + logger.info( + f"Fetched all notifications across {page_count} pages") break - + except Exception as e: error_str = str(e) - logger.error(f"Error fetching notifications page {page_count}: {e}") - + logger.error( + f"Error fetching notifications page {page_count}: {e}") + # Handle specific API errors if 'rate limit' in error_str.lower(): - logger.warning("Rate limit hit while fetching notifications, will retry next cycle") + logger.warning( + "Rate limit hit while fetching notifications, will retry next cycle") break elif '401' in error_str or 'unauthorized' in error_str.lower(): logger.error("Authentication error, re-raising exception") raise else: # For other errors, try to continue with what we have - logger.warning("Continuing with notifications fetched so far") + logger.warning( + "Continuing with notifications fetched so far") break # Queue all unread notifications (except likes) @@ -993,16 +1112,20 @@ def process_notifications(void_agent, atproto_client, testing_mode=False): # Mark all notifications as seen immediately after queuing (unless in testing mode) if testing_mode: - logger.info("๐Ÿงช TESTING MODE: Skipping marking notifications as seen") + logger.info( + "๐Ÿงช TESTING MODE: Skipping marking notifications as seen") else: if new_count > 0: - atproto_client.app.bsky.notification.update_seen({'seen_at': last_seen_at}) - logger.info(f"Queued {new_count} new notifications and marked as seen") + atproto_client.app.bsky.notification.update_seen( + {'seen_at': last_seen_at}) + logger.info( + f"Queued {new_count} new notifications and marked as seen") else: logger.debug("No new notifications to queue") # Now process the entire queue (old + new notifications) - load_and_process_queued_notifications(void_agent, atproto_client, testing_mode) + load_and_process_queued_notifications( + void_agent, atproto_client, testing_mode) except Exception as e: logger.error(f"Error processing notifications: {e}") @@ -1010,18 +1133,21 @@ def process_notifications(void_agent, atproto_client, testing_mode=False): def main(): # Parse command line arguments - parser = argparse.ArgumentParser(description='Void Bot - Bluesky autonomous agent') - parser.add_argument('--test', action='store_true', help='Run in testing mode (no messages sent, queue files preserved)') - parser.add_argument('--no-git', action='store_true', help='Skip git operations when exporting agent state') + parser = argparse.ArgumentParser( + description='Void Bot - Bluesky autonomous agent') + parser.add_argument('--test', action='store_true', + help='Run in testing mode (no messages sent, queue files preserved)') + parser.add_argument('--no-git', action='store_true', + help='Skip git operations when exporting agent state') args = parser.parse_args() - + global TESTING_MODE TESTING_MODE = args.test - + # Store no-git flag globally for use in export_agent_state calls global SKIP_GIT SKIP_GIT = args.no_git - + if TESTING_MODE: logger.info("๐Ÿงช === RUNNING IN TESTING MODE ===") logger.info(" - No messages will be sent to Bluesky") @@ -1034,23 +1160,27 @@ def main(): logger.info("=== STARTING VOID BOT ===") void_agent = initialize_void() logger.info(f"Void agent initialized: {void_agent.id}") - + # Check if agent has required tools if hasattr(void_agent, 'tools') and void_agent.tools: tool_names = [tool.name for tool in void_agent.tools] # Check for bluesky-related tools - bluesky_tools = [name for name in tool_names if 'bluesky' in name.lower() or 'reply' in name.lower()] + bluesky_tools = [name for name in tool_names if 'bluesky' in name.lower( + ) or 'reply' in name.lower()] if not bluesky_tools: - logger.warning("No Bluesky-related tools found! Agent may not be able to reply.") + logger.warning( + "No Bluesky-related tools found! Agent may not be able to reply.") else: logger.warning("Agent has no tools registered!") # Initialize Bluesky client + logger.debug("Connecting to Bluesky") atproto_client = bsky_utils.default_login() logger.info("Connected to Bluesky") # Main loop - logger.info(f"Starting notification monitoring, checking every {FETCH_NOTIFICATIONS_DELAY_SEC} seconds") + logger.info( + f"Starting notification monitoring, checking every {FETCH_NOTIFICATIONS_DELAY_SEC} seconds") cycle_count = 0 while True: @@ -1060,31 +1190,38 @@ def main(): # Log cycle completion with stats elapsed_time = time.time() - start_time total_messages = sum(message_counters.values()) - messages_per_minute = (total_messages / elapsed_time * 60) if elapsed_time > 0 else 0 - + messages_per_minute = ( + total_messages / elapsed_time * 60) if elapsed_time > 0 else 0 + if total_messages > 0: - logger.info(f"Cycle {cycle_count} complete. Session totals: {total_messages} messages ({message_counters['mentions']} mentions, {message_counters['replies']} replies) | {messages_per_minute:.1f} msg/min") + logger.info( + f"Cycle {cycle_count} complete. Session totals: {total_messages} messages ({message_counters['mentions']} mentions, {message_counters['replies']} replies) | {messages_per_minute:.1f} msg/min") sleep(FETCH_NOTIFICATIONS_DELAY_SEC) except KeyboardInterrupt: # Final stats elapsed_time = time.time() - start_time total_messages = sum(message_counters.values()) - messages_per_minute = (total_messages / elapsed_time * 60) if elapsed_time > 0 else 0 - + messages_per_minute = ( + total_messages / elapsed_time * 60) if elapsed_time > 0 else 0 + logger.info("=== BOT STOPPED BY USER ===") - logger.info(f"๐Ÿ“Š Final session stats: {total_messages} total messages processed in {elapsed_time/60:.1f} minutes") + logger.info( + f"๐Ÿ“Š Final session stats: {total_messages} total messages processed in {elapsed_time/60:.1f} minutes") logger.info(f" - {message_counters['mentions']} mentions") logger.info(f" - {message_counters['replies']} replies") logger.info(f" - {message_counters['follows']} follows") - logger.info(f" - {message_counters['reposts_skipped']} reposts skipped") - logger.info(f" - Average rate: {messages_per_minute:.1f} messages/minute") + logger.info( + f" - {message_counters['reposts_skipped']} reposts skipped") + logger.info( + f" - Average rate: {messages_per_minute:.1f} messages/minute") break except Exception as e: logger.error(f"=== ERROR IN MAIN LOOP CYCLE {cycle_count} ===") logger.error(f"Error details: {e}") # Wait a bit longer on errors - logger.info(f"Sleeping for {FETCH_NOTIFICATIONS_DELAY_SEC * 2} seconds due to error...") + logger.info( + f"Sleeping for {FETCH_NOTIFICATIONS_DELAY_SEC * 2} seconds due to error...") sleep(FETCH_NOTIFICATIONS_DELAY_SEC * 2) diff --git a/bsky_utils.py b/bsky_utils.py index aecb92a..1d20fbb 100644 --- a/bsky_utils.py +++ b/bsky_utils.py @@ -1,3 +1,6 @@ +import json +import yaml +import dotenv import os import logging from typing import Optional, Dict, Any, List @@ -10,11 +13,8 @@ logging.basicConfig( logger = logging.getLogger("bluesky_session_handler") # Load the environment variables -import dotenv dotenv.load_dotenv(override=True) -import yaml -import json # Strip fields. A list of fields to remove from a JSON object STRIP_FIELDS = [ @@ -63,6 +63,8 @@ STRIP_FIELDS = [ "mime_type", "size", ] + + def convert_to_basic_types(obj): """Convert complex Python objects to basic types for JSON/YAML serialization.""" if hasattr(obj, '__dict__'): @@ -117,24 +119,24 @@ def strip_fields(obj, strip_field_list): def flatten_thread_structure(thread_data): """ Flatten a nested thread structure into a list while preserving all data. - + Args: thread_data: The thread data from get_post_thread - + Returns: Dict with 'posts' key containing a list of posts in chronological order """ posts = [] - + def traverse_thread(node): """Recursively traverse the thread structure to collect posts.""" if not node: return - + # If this node has a parent, traverse it first (to maintain chronological order) if hasattr(node, 'parent') and node.parent: traverse_thread(node.parent) - + # Then add this node's post if hasattr(node, 'post') and node.post: # Convert to dict if needed to ensure we can process it @@ -144,16 +146,16 @@ def flatten_thread_structure(thread_data): post_dict = node.post.copy() else: post_dict = {} - + posts.append(post_dict) - + # Handle the thread structure if hasattr(thread_data, 'thread'): # Start from the main thread node traverse_thread(thread_data.thread) elif hasattr(thread_data, '__dict__') and 'thread' in thread_data.__dict__: traverse_thread(thread_data.__dict__['thread']) - + # Return a simple structure with posts list return {'posts': posts} @@ -171,7 +173,7 @@ def thread_to_yaml_string(thread, strip_metadata=True): """ # First flatten the thread structure to avoid deep nesting flattened = flatten_thread_structure(thread) - + # Convert complex objects to basic types basic_thread = convert_to_basic_types(flattened) @@ -184,11 +186,6 @@ def thread_to_yaml_string(thread, strip_metadata=True): return yaml.dump(cleaned_thread, indent=2, allow_unicode=True, default_flow_style=False) - - - - - def get_session(username: str) -> Optional[str]: try: with open(f"session_{username}.txt", encoding="UTF-8") as f: @@ -197,17 +194,20 @@ def get_session(username: str) -> Optional[str]: logger.debug(f"No existing session found for {username}") return None + def save_session(username: str, session_string: str) -> None: with open(f"session_{username}.txt", "w", encoding="UTF-8") as f: f.write(session_string) logger.debug(f"Session saved for {username}") + def on_session_change(username: str, event: SessionEvent, session: Session) -> None: logger.debug(f"Session changed: {event} {repr(session)}") if event in (SessionEvent.CREATE, SessionEvent.REFRESH): logger.debug(f"Saving changed session for {username}") save_session(username, session.export()) + def init_client(username: str, password: str, pds_uri: str = "https://bsky.social") -> Client: if pds_uri is None: logger.warning( @@ -243,7 +243,8 @@ def default_login() -> Client: password = config['password'] pds_uri = config['pds_uri'] except (ImportError, FileNotFoundError, KeyError) as e: - logger.warning(f"Could not load from config file ({e}), falling back to environment variables") + logger.warning( + f"Could not load from config file ({e}), falling back to environment variables") username = os.getenv("BSKY_USERNAME") password = os.getenv("BSKY_PASSWORD") pds_uri = os.getenv("PDS_URI", "https://bsky.social") @@ -262,31 +263,33 @@ def default_login() -> Client: return init_client(username, password, pds_uri) + def remove_outside_quotes(text: str) -> str: """ Remove outside double quotes from response text. - + Only handles double quotes to avoid interfering with contractions: - Double quotes: "text" โ†’ text - Preserves single quotes and internal quotes - + Args: text: The text to process - + Returns: Text with outside double quotes removed """ if not text or len(text) < 2: return text - + text = text.strip() - + # Only remove double quotes from start and end if text.startswith('"') and text.endswith('"'): return text[1:-1] - + return text + def reply_to_post(client: Client, text: str, reply_to_uri: str, reply_to_cid: str, root_uri: Optional[str] = None, root_cid: Optional[str] = None, lang: Optional[str] = None) -> Dict[str, Any]: """ Reply to a post on Bluesky with rich text support. @@ -304,23 +307,25 @@ def reply_to_post(client: Client, text: str, reply_to_uri: str, reply_to_cid: st The response from sending the post """ import re - + # If root is not provided, this is a reply to the root post if root_uri is None: root_uri = reply_to_uri root_cid = reply_to_cid # Create references for the reply - parent_ref = models.create_strong_ref(models.ComAtprotoRepoStrongRef.Main(uri=reply_to_uri, cid=reply_to_cid)) - root_ref = models.create_strong_ref(models.ComAtprotoRepoStrongRef.Main(uri=root_uri, cid=root_cid)) + parent_ref = models.create_strong_ref( + models.ComAtprotoRepoStrongRef.Main(uri=reply_to_uri, cid=reply_to_cid)) + root_ref = models.create_strong_ref( + models.ComAtprotoRepoStrongRef.Main(uri=root_uri, cid=root_cid)) # Parse rich text facets (mentions and URLs) facets = [] text_bytes = text.encode("UTF-8") - + # Parse mentions - fixed to handle @ at start of text mention_regex = rb"(?:^|[$|\W])(@([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)" - + for m in re.finditer(mention_regex, text_bytes): handle = m.group(1)[1:].decode("UTF-8") # Remove @ prefix # Adjust byte positions to account for the optional prefix @@ -336,16 +341,26 @@ def reply_to_post(client: Client, text: str, reply_to_uri: str, reply_to_cid: st byteStart=mention_start, byteEnd=mention_end ), - features=[models.AppBskyRichtextFacet.Mention(did=resolve_resp.did)] + features=[models.AppBskyRichtextFacet.Mention( + did=resolve_resp.did)] ) ) except Exception as e: - logger.debug(f"Failed to resolve handle {handle}: {e}") + # Handle specific error cases + error_str = str(e) + if 'Could not find user info' in error_str or 'InvalidRequest' in error_str: + logger.warning( + f"User @{handle} not found (account may be deleted/suspended), skipping mention facet") + elif 'BadRequestError' in error_str: + logger.warning( + f"Bad request when resolving @{handle}, skipping mention facet: {e}") + else: + logger.debug(f"Failed to resolve handle @{handle}: {e}") continue - + # Parse URLs - fixed to handle URLs at start of text url_regex = rb"(?:^|[$|\W])(https?:\/\/(www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b([-a-zA-Z0-9()@:%_\+.~#?&//=]*[-a-zA-Z0-9@%_\+~#//=])?)" - + for m in re.finditer(url_regex, text_bytes): url = m.group(1).decode("UTF-8") # Adjust byte positions to account for the optional prefix @@ -365,14 +380,16 @@ def reply_to_post(client: Client, text: str, reply_to_uri: str, reply_to_cid: st if facets: response = client.send_post( text=text, - reply_to=models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref), + reply_to=models.AppBskyFeedPost.ReplyRef( + parent=parent_ref, root=root_ref), facets=facets, langs=[lang] if lang else None ) else: response = client.send_post( text=text, - reply_to=models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref), + reply_to=models.AppBskyFeedPost.ReplyRef( + parent=parent_ref, root=root_ref), langs=[lang] if lang else None ) @@ -392,10 +409,21 @@ def get_post_thread(client: Client, uri: str) -> Optional[Dict[str, Any]]: The thread data or None if not found """ try: - thread = client.app.bsky.feed.get_post_thread({'uri': uri, 'parent_height': 60, 'depth': 10}) + thread = client.app.bsky.feed.get_post_thread( + {'uri': uri, 'parent_height': 60, 'depth': 10}) return thread except Exception as e: - logger.error(f"Error fetching post thread: {e}") + error_str = str(e) + # Handle specific error cases more gracefully + if 'Could not find user info' in error_str or 'InvalidRequest' in error_str: + logger.warning( + f"User account not found for post URI {uri} (account may be deleted/suspended)") + elif 'NotFound' in error_str or 'Post not found' in error_str: + logger.warning(f"Post not found for URI {uri}") + elif 'BadRequestError' in error_str: + logger.warning(f"Bad request error for URI {uri}: {e}") + else: + logger.error(f"Error fetching post thread: {e}") return None @@ -492,9 +520,10 @@ def reply_with_thread_to_notification(client: Client, notification: Any, reply_m logger.error("Reply messages list cannot be empty") return None if len(reply_messages) > 15: - logger.error(f"Cannot send more than 15 reply messages (got {len(reply_messages)})") + logger.error( + f"Cannot send more than 15 reply messages (got {len(reply_messages)})") return None - + # Get the post URI and CID from the notification (handle both dict and object) if isinstance(notification, dict): post_uri = notification.get('uri') @@ -512,7 +541,7 @@ def reply_with_thread_to_notification(client: Client, notification: Any, reply_m # Get the thread to find the root post thread_data = get_post_thread(client, post_uri) - + root_uri = post_uri root_cid = post_cid @@ -532,10 +561,11 @@ def reply_with_thread_to_notification(client: Client, notification: Any, reply_m responses = [] current_parent_uri = post_uri current_parent_cid = post_cid - + for i, message in enumerate(reply_messages): - logger.info(f"Sending reply {i+1}/{len(reply_messages)}: {message[:50]}...") - + logger.info( + f"Sending reply {i+1}/{len(reply_messages)}: {message[:50]}...") + # Send this reply response = reply_to_post( client=client, @@ -546,9 +576,10 @@ def reply_with_thread_to_notification(client: Client, notification: Any, reply_m root_cid=root_cid, lang=lang ) - + if not response: - logger.error(f"Failed to send reply {i+1}, posting system failure message") + logger.error( + f"Failed to send reply {i+1}, posting system failure message") # Try to post a system failure message failure_response = reply_to_post( client=client, @@ -564,7 +595,8 @@ def reply_with_thread_to_notification(client: Client, notification: Any, reply_m current_parent_uri = failure_response.uri current_parent_cid = failure_response.cid else: - logger.error("Could not even send system failure message, stopping thread") + logger.error( + "Could not even send system failure message, stopping thread") return responses if responses else None else: responses.append(response) @@ -572,7 +604,7 @@ def reply_with_thread_to_notification(client: Client, notification: Any, reply_m if i < len(reply_messages) - 1: # Not the last message current_parent_uri = response.uri current_parent_cid = response.cid - + logger.info(f"Successfully sent {len(responses)} threaded replies") return responses diff --git a/register_tools.py b/register_tools.py index b8c184b..59c5b9e 100755 --- a/register_tools.py +++ b/register_tools.py @@ -4,10 +4,10 @@ import os import sys import logging from typing import List -from dotenv import load_dotenv from letta_client import Letta from rich.console import Console from rich.table import Table +from config_loader import get_config, get_letta_config, get_agent_config # Import standalone functions and their schemas from tools.search import search_bluesky_posts, SearchArgs @@ -18,7 +18,9 @@ from tools.halt import halt_activity, HaltArgs from tools.thread import add_post_to_bluesky_reply_thread, ReplyThreadPostArgs from tools.ignore import ignore_notification, IgnoreNotificationArgs -load_dotenv() +config = get_config() +letta_config = get_letta_config() +agent_config = get_agent_config() logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) console = Console() @@ -101,16 +103,20 @@ TOOL_CONFIGS = [ ] -def register_tools(agent_name: str = "void", tools: List[str] = None): +def register_tools(agent_name: str = None, tools: List[str] = None): """Register tools with a Letta agent. Args: - agent_name: Name of the agent to attach tools to + agent_name: Name of the agent to attach tools to. If None, uses config default. tools: List of tool names to register. If None, registers all tools. """ + # Use agent name from config if not provided + if agent_name is None: + agent_name = agent_config['name'] + try: - # Initialize Letta client with API key - client = Letta(token=os.environ["LETTA_API_KEY"]) + # Initialize Letta client with API key from config + client = Letta(token=letta_config['api_key']) # Find the agent agents = client.agents.list() @@ -201,7 +207,7 @@ if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Register Void tools with a Letta agent") - parser.add_argument("agent", nargs="?", default="void", help="Agent name (default: void)") + parser.add_argument("agent", nargs="?", default=None, help=f"Agent name (default: {agent_config['name']})") parser.add_argument("--tools", nargs="+", help="Specific tools to register (default: all)") parser.add_argument("--list", action="store_true", help="List available tools") @@ -210,5 +216,7 @@ if __name__ == "__main__": if args.list: list_available_tools() else: - console.print(f"\n[bold]Registering tools for agent: {args.agent}[/bold]\n") + # Use config default if no agent specified + agent_name = args.agent if args.agent is not None else agent_config['name'] + console.print(f"\n[bold]Registering tools for agent: {agent_name}[/bold]\n") register_tools(args.agent, args.tools) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b425c62 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,23 @@ +# Core dependencies for Void Bot + +# Configuration and utilities +PyYAML>=6.0.2 +rich>=14.0.0 +python-dotenv>=1.0.0 + +# Letta API client +letta-client>=0.1.198 + +# AT Protocol (Bluesky) client +atproto>=0.0.54 + +# HTTP client for API calls +httpx>=0.28.1 +httpx-sse>=0.4.0 +requests>=2.31.0 + +# Data validation +pydantic>=2.11.7 + +# Async support +anyio>=4.9.0