Add agent listing utility and X profile analysis tool

- Add list_agents.py for querying Letta agents with filtering capabilities
- Add tools/x_profile.py for analyzing X user profiles using Bluesky agent
- Update bsky.py with minor improvements to queue processing

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[e-mail]>

bsky.py

···
 
 
 # Create a client with extended timeout for LLM operations
+from config_loader import get_letta_config
+letta_config = get_letta_config()
 CLIENT= Letta(
-    token=os.environ["LETTA_API_KEY"],
-    timeout=600  # 10 minutes timeout for API calls - higher than Cloudflare's 524 timeout
+    token=letta_config['api_key'],
+    timeout=letta_config['timeout']  # 10 minutes timeout for API calls - higher than Cloudflare's 524 timeout
 )
 
 # Use the "Bluesky" project
-PROJECT_ID = "5ec33d52-ab14-4fd6-91b5-9dbc43e888a8"
+PROJECT_ID = letta_config['project_id']
 
 # Notification check delay
 FETCH_NOTIFICATIONS_DELAY_SEC = 10  # Check every 10 seconds for faster response
···
         logger.error(f"Failed to export agent: {e}")
 
 def initialize_void():
-    logger.info("Starting void agent initialization...")
+    logger.info("Starting agent initialization...")
 
-    # Get the configured void agent by ID
-    logger.info("Loading void agent from config...")
-    from config_loader import get_letta_config
-    letta_config = get_letta_config()
+    # Get the configured agent by ID
+    logger.info("Loading agent from config...")
     agent_id = letta_config['agent_id']
     
     try:
-        void_agent = CLIENT.agents.retrieve(agent_id=agent_id)
-        logger.info(f"Successfully loaded void agent: {void_agent.name} ({agent_id})")
+        agent = CLIENT.agents.retrieve(agent_id=agent_id)
+        logger.info(f"Successfully loaded agent: {agent.name} ({agent_id})")
     except Exception as e:
-        logger.error(f"Failed to load void agent {agent_id}: {e}")
+        logger.error(f"Failed to load agent {agent_id}: {e}")
         logger.error("Please ensure the agent_id in config.yaml is correct")
         raise e
     
     # Export agent state
     logger.info("Exporting agent state...")
-    export_agent_state(CLIENT, void_agent, skip_git=SKIP_GIT)
+    export_agent_state(CLIENT, 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}")
-    if hasattr(void_agent, 'llm_config'):
-        logger.info(f"Agent model: {void_agent.llm_config.model}")
-    logger.info(f"Agent project_id: {void_agent.project_id}")
-    if hasattr(void_agent, 'tools'):
-        logger.info(f"Agent has {len(void_agent.tools)} tools")
-        for tool in void_agent.tools[:3]:  # Show first 3 tools
+    logger.info(f"Agent details - ID: {agent.id}")
+    logger.info(f"Agent name: {agent.name}")
+    if hasattr(agent, 'llm_config'):
+        logger.info(f"Agent model: {agent.llm_config.model}")
+    logger.info(f"Agent project_id: {agent.project_id}")
+    if hasattr(agent, 'tools'):
+        logger.info(f"Agent has {len(agent.tools)} tools")
+        for tool in agent.tools[:3]:  # Show first 3 tools
             logger.info(f"  - Tool: {tool.name} (type: {tool.tool_type})")
 
-    return void_agent
+    return agent
 
 
-def process_mention(void_agent, atproto_client, notification_data, queue_filepath=None, testing_mode=False):
+def process_mention(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
+        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)
···
         
         try:
             # Check for known bots in thread
-            bot_check_result = check_known_bots(unique_handles, void_agent)
+            bot_check_result = check_known_bots(unique_handles, agent)
             bot_check_data = json.loads(bot_check_result)
             
             if bot_check_data.get("bot_detected", False):
···
         if unique_handles:
             try:
                 logger.debug(f"Attaching user blocks for handles: {unique_handles}")
-                attach_result = attach_user_blocks(unique_handles, void_agent)
+                attach_result = attach_user_blocks(unique_handles, agent)
                 attached_handles = unique_handles  # Track successfully attached handles
                 logger.debug(f"Attach result: {attach_result}")
             except Exception as attach_error:
···
         try:
             # Use streaming to avoid 524 timeout errors
             message_stream = CLIENT.agents.messages.create_stream(
-                agent_id=void_agent.id,
+                agent_id=agent.id,
                 messages=[{"role": "user", "content": prompt}],
                 stream_tokens=False,  # Step streaming only (faster than token streaming)
                 max_steps=100
···
                     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)
+                    export_agent_state(CLIENT, agent, skip_git=SKIP_GIT)
                     logger.info("=== BOT TERMINATED DUE TO DEPRECATED TOOL USE ===")
                     exit(1)
         
···
                         save_processed_notifications(processed_uris)
                     
                     # Export agent state before terminating
-                    export_agent_state(CLIENT, void_agent, skip_git=SKIP_GIT)
+                    export_agent_state(CLIENT, agent, skip_git=SKIP_GIT)
                     
                     # Exit the program
                     logger.info("=== BOT TERMINATED BY AGENT ===")
···
                     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)
+                    export_agent_state(CLIENT, agent, skip_git=SKIP_GIT)
                     logger.info("=== BOT TERMINATED DUE TO DEPRECATED TOOL USE ===")
                     exit(1)
                 
···
         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)
+                detach_result = detach_user_blocks(attached_handles, agent)
                 logger.debug(f"Detach result: {detach_result}")
             except Exception as detach_error:
                 logger.warning(f"Failed to detach user blocks: {detach_error}")
···
         return False
 
 
-def load_and_process_queued_notifications(void_agent, atproto_client, testing_mode=False):
+def load_and_process_queued_notifications(agent, atproto_client, testing_mode=False):
     """Load and process all notifications from the queue in priority order."""
     try:
         # Get all JSON files in queue directory (excluding processed_notifications.json)
···
                 # 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(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(agent, atproto_client, notif_data, queue_filepath=filepath, testing_mode=testing_mode)
                     if success:
                         message_counters['replies'] += 1
                 elif notif_data['reason'] == "follow":
+                    # Skip
+                    logging.info("Skipping new follower notification, currently disabled")
+
+                    
                     author_handle = notif_data['author']['handle']
                     author_display_name = notif_data['author'].get('display_name', 'no display name')
                     follow_update = f"@{author_handle} ({author_display_name}) started following you."
···
                     try:
                         # Use streaming to match other notification processing
                         message_stream = CLIENT.agents.messages.create_stream(
-                            agent_id=void_agent.id,
+                            agent_id=agent.id,
                             messages=[{"role": "user", "content": follow_message}],
                             stream_tokens=False,
                             max_steps=50  # Fewer steps needed for simple follow updates
···
         return 0
 
 
-def process_notifications(void_agent, atproto_client, testing_mode=False):
+def process_notifications(agent, atproto_client, testing_mode=False):
     """Fetch new notifications, queue them, and process the queue."""
     try:
         # Fetch and queue new notifications
···
             logger.info(f"Found {new_count} new notifications to process")
 
         # Now process the entire queue (old + new notifications)
-        load_and_process_queued_notifications(void_agent, atproto_client, testing_mode)
+        load_and_process_queued_notifications(agent, atproto_client, testing_mode)
 
     except Exception as e:
         logger.error(f"Error processing notifications: {e}")
···
         synthesis_prompt = f"""Time for synthesis and reflection.
 
 You have access to temporal journal blocks for recording your thoughts and experiences:
-- void_day_{today.strftime('%Y_%m_%d')}: Today's journal ({today.strftime('%B %d, %Y')})
-- void_month_{today.strftime('%Y_%m')}: This month's journal ({today.strftime('%B %Y')})
-- void_year_{today.year}: This year's journal ({today.year})
+- comind_day_{today.strftime('%Y_%m_%d')}: Today's journal ({today.strftime('%B %d, %Y')})
+- comind_month_{today.strftime('%Y_%m')}: This month's journal ({today.strftime('%B %Y')})
+- comind_year_{today.year}: This year's journal ({today.year})
 
 These journal blocks are attached temporarily for this synthesis session. Use them to:
 1. Record significant interactions and insights from recent experiences
···
 
 The journal entries should be cumulative - add to existing content rather than replacing it.
 Consider both immediate experiences (daily) and longer-term patterns (monthly/yearly).
-
-After recording in your journals, synthesize your recent experiences into your core memory blocks
-(zeitgeist, void-persona, void-humans) as you normally would.
 
 Begin your synthesis and journaling now."""
         
···
 
 def main():
     # Parse command line arguments
-    parser = argparse.ArgumentParser(description='Void Bot - Bluesky autonomous agent')
+    parser = argparse.ArgumentParser(description='Comind - 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.add_argument('--simple-logs', action='store_true', help='Use simplified log format (void - LEVEL - message)')
···
     
     # Configure logging based on command line arguments
     if args.simple_logs:
-        log_format = "void - %(levelname)s - %(message)s"
+        log_format = "comind - %(levelname)s - %(message)s"
     else:
         # Create custom formatter with symbols
         class SymbolFormatter(logging.Formatter):
···
         logging.root.addHandler(handler)
     
     global logger, prompt_logger, console
-    logger = logging.getLogger("void_bot")
+    logger = logging.getLogger("comind_bot")
     logger.setLevel(logging.INFO)
     
     # Create a separate logger for prompts (set to WARNING to hide by default)
-    prompt_logger = logging.getLogger("void_bot.prompts")
+    prompt_logger = logging.getLogger("comind_bot.prompts")
     if args.reasoning:
         prompt_logger.setLevel(logging.INFO)  # Show reasoning when --reasoning is used
     else:
···
     """Main bot loop that continuously monitors for notifications."""
     global start_time
     start_time = time.time()
-    logger.info("=== STARTING VOID BOT ===")
-    void_agent = initialize_void()
-    logger.info(f"Void agent initialized: {void_agent.id}")
+    logger.info("""
+                                    ███                 █████
+                                   ░░░                 ░░███ 
+  ██████   ██████  █████████████   ████  ████████    ███████ 
+ ███░░███ ███░░███░░███░░███░░███ ░░███ ░░███░░███  ███░░███ 
+░███ ░░░ ░███ ░███ ░███ ░███ ░███  ░███  ░███ ░███ ░███ ░███ 
+░███  ███░███ ░███ ░███ ░███ ░███  ░███  ░███ ░███ ░███ ░███ 
+░░██████ ░░██████  █████░███ █████ █████ ████ █████░░████████
+ ░░░░░░   ░░░░░░  ░░░░░ ░░░ ░░░░░ ░░░░░ ░░░░ ░░░░░  ░░░░░░░░ 
+                                                             
+                                                             
+                                                             """)
+    agent = initialize_void()
+    logger.info(f"Agent initialized: {agent.id}")
     
     # Ensure correct tools are attached for Bluesky
     logger.info("Configuring tools for Bluesky platform...")
     try:
         from tool_manager import ensure_platform_tools
-        ensure_platform_tools('bluesky', void_agent.id)
+        ensure_platform_tools('bluesky', agent.id)
     except Exception as e:
         logger.error(f"Failed to configure platform tools: {e}")
         logger.warning("Continuing with existing tool configuration")
     
     # 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]
+    if hasattr(agent, 'tools') and agent.tools:
+        tool_names = [tool.name for tool in 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()]
         if not bluesky_tools:
···
 
     # Clean up all user blocks at startup
     logger.info("🧹 Cleaning up user blocks at startup...")
-    periodic_user_block_cleanup(CLIENT, void_agent.id)
+    periodic_user_block_cleanup(CLIENT, agent.id)
     
     # Initialize Bluesky client (needed for both notification processing and synthesis acks/posts)
     if not SYNTHESIS_ONLY:
···
             try:
                 # Send synthesis message immediately on first run
                 logger.info("🧠 Sending synthesis message")
-                send_synthesis_message(CLIENT, void_agent.id, atproto_client)
+                send_synthesis_message(CLIENT, agent.id, atproto_client)
                 
                 # Wait for next interval
                 logger.info(f"Waiting {SYNTHESIS_INTERVAL} seconds until next synthesis...")
···
     while True:
         try:
             cycle_count += 1
-            process_notifications(void_agent, atproto_client, TESTING_MODE)
+            process_notifications(agent, atproto_client, TESTING_MODE)
             
             # Check if synthesis interval has passed
             if SYNTHESIS_INTERVAL > 0:
···
                 global last_synthesis_time
                 if current_time - last_synthesis_time >= SYNTHESIS_INTERVAL:
                     logger.info(f"⏰ {SYNTHESIS_INTERVAL/60:.1f} minutes have passed, triggering synthesis")
-                    send_synthesis_message(CLIENT, void_agent.id, atproto_client)
+                    send_synthesis_message(CLIENT, agent.id, atproto_client)
                     last_synthesis_time = current_time
             
             # Run periodic cleanup every N cycles
             if CLEANUP_INTERVAL > 0 and cycle_count % CLEANUP_INTERVAL == 0:
                 logger.debug(f"Running periodic user block cleanup (cycle {cycle_count})")
-                periodic_user_block_cleanup(CLIENT, void_agent.id)
+                periodic_user_block_cleanup(CLIENT, agent.id)
             
             # Log cycle completion with stats
             elapsed_time = time.time() - start_time

list_agents.py

···
+#!/usr/bin/env python
+"""List all available agents in the Letta project."""
+
+from letta_client import Letta
+from config_loader import get_letta_config
+
+# Get configuration
+letta_config = get_letta_config()
+
+# Create client
+client = Letta(
+    token=letta_config['api_key'],
+    timeout=letta_config['timeout']
+)
+
+# List all agents
+print("Available agents:")
+print("-" * 50)
+
+try:
+    agents = client.agents.list()
+    if not agents:
+        print("No agents found in this project.")
+    else:
+        for agent in agents:
+            print(f"Name: {agent.name}")
+            print(f"ID: {agent.id}")
+            if hasattr(agent, 'description'):
+                print(f"Description: {agent.description}")
+            print("-" * 50)
+except Exception as e:
+    print(f"Error listing agents: {e}")

tools/x_profile.py

···
+"""Profile search tool for X (Twitter) users."""
+from pydantic import BaseModel, Field
+from typing import Optional
+
+
+class SearchXProfileArgs(BaseModel):
+    username: str = Field(..., description="X username to look up (without @)")
+
+
+def search_x_profile(username: str) -> str:
+    """
+    Look up detailed profile information for an X (Twitter) user.
+    
+    Args:
+        username: X username to look up (without @)
+        
+    Returns:
+        YAML-formatted profile information including bio, metrics, verification status, etc.
+    """
+    import os
+    import yaml
+    import requests
+    from datetime import datetime
+    
+    try:
+        # Validate username (remove @ if present)
+        username = username.lstrip('@')
+        if not username:
+            raise Exception("Username cannot be empty")
+        
+        # Get credentials from environment
+        consumer_key = os.getenv("X_CONSUMER_KEY")
+        consumer_secret = os.getenv("X_CONSUMER_SECRET")
+        access_token = os.getenv("X_ACCESS_TOKEN")
+        access_token_secret = os.getenv("X_ACCESS_TOKEN_SECRET")
+        
+        # Also check for bearer token as fallback
+        bearer_token = os.getenv("X_BEARER_TOKEN")
+        
+        if not any([bearer_token, (consumer_key and consumer_secret and access_token and access_token_secret)]):
+            raise Exception("X API credentials not found in environment variables")
+        
+        # Set up authentication headers
+        base_url = "https://api.x.com/2"
+        if bearer_token:
+            headers = {
+                "Authorization": f"Bearer {bearer_token}",
+                "Content-Type": "application/json"
+            }
+        else:
+            # For OAuth 1.0a, we'd need requests_oauthlib
+            # Since this is a cloud function, we'll require bearer token for simplicity
+            raise Exception("Bearer token required for X API authentication in cloud environment")
+        
+        # Get user profile information
+        user_lookup_url = f"{base_url}/users/by/username/{username}"
+        user_params = {
+            "user.fields": ",".join([
+                "id",
+                "name", 
+                "username",
+                "description",
+                "location",
+                "url",
+                "created_at",
+                "verified",
+                "verified_type",
+                "public_metrics",
+                "profile_image_url",
+                "profile_banner_url",
+                "protected"
+            ])
+        }
+        
+        try:
+            response = requests.get(user_lookup_url, headers=headers, params=user_params, timeout=10)
+            response.raise_for_status()
+            data = response.json()
+            
+            if "data" not in data:
+                raise Exception(f"User @{username} not found")
+                
+            user = data["data"]
+            
+        except requests.exceptions.HTTPError as e:
+            if response.status_code == 404:
+                raise Exception(f"User @{username} not found")
+            elif response.status_code == 429:
+                raise Exception("X API rate limit exceeded. Please try again later.")
+            else:
+                raise Exception(f"Failed to look up user @{username}: {str(e)}")
+        
+        # Format the profile data
+        profile_data = {
+            "x_user_profile": {
+                "basic_info": {
+                    "id": user.get("id"),
+                    "username": user.get("username"),
+                    "display_name": user.get("name"),
+                    "description": user.get("description", ""),
+                    "location": user.get("location", ""),
+                    "website": user.get("url", ""),
+                    "created_at": user.get("created_at", ""),
+                    "profile_url": f"https://x.com/{username}"
+                },
+                "verification": {
+                    "verified": user.get("verified", False),
+                    "verified_type": user.get("verified_type", ""),
+                },
+                "privacy": {
+                    "protected": user.get("protected", False)
+                },
+                "media": {
+                    "profile_image_url": user.get("profile_image_url", ""),
+                    "profile_banner_url": user.get("profile_banner_url", "")
+                }
+            }
+        }
+        
+        # Add public metrics if available
+        if "public_metrics" in user:
+            metrics = user["public_metrics"]
+            profile_data["x_user_profile"]["metrics"] = {
+                "followers_count": metrics.get("followers_count", 0),
+                "following_count": metrics.get("following_count", 0),
+                "tweet_count": metrics.get("tweet_count", 0),
+                "listed_count": metrics.get("listed_count", 0),
+                "like_count": metrics.get("like_count", 0)
+            }
+        
+        return yaml.dump(profile_data, default_flow_style=False, sort_keys=False)
+        
+    except Exception as e:
+        raise Exception(f"Error searching X profile: {str(e)}")