From cc65f78c33fccc0016b680b4b6504123a9a39a5f Mon Sep 17 00:00:00 2001 From: Cameron Pfiffer Date: Wed, 20 Aug 2025 07:57:27 -0700 Subject: [PATCH] Add agent listing utility and X profile analysis tool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- bsky.py | 126 +++++++++++++++++++++++------------------- list_agents.py | 32 +++++++++++ tools/x_profile.py | 134 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 235 insertions(+), 57 deletions(-) create mode 100644 list_agents.py create mode 100644 tools/x_profile.py diff --git a/bsky.py b/bsky.py index ec4a2f3..a7e0dad 100644 --- a/bsky.py +++ b/bsky.py @@ -73,13 +73,15 @@ def log_with_panel(message, title=None, border_color="white"): # 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 @@ -157,45 +159,43 @@ def export_agent_state(client, agent, skip_git=False): 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) @@ -325,7 +325,7 @@ To reply, use the add_post_to_bluesky_reply_thread tool: 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): @@ -351,7 +351,7 @@ To reply, use the add_post_to_bluesky_reply_thread tool: 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: @@ -378,7 +378,7 @@ To reply, use the add_post_to_bluesky_reply_thread tool: 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 @@ -666,7 +666,7 @@ To reply, use the add_post_to_bluesky_reply_thread tool: 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) @@ -711,7 +711,7 @@ To reply, use the add_post_to_bluesky_reply_thread tool: 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 ===") @@ -724,7 +724,7 @@ To reply, use the add_post_to_bluesky_reply_thread tool: 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) @@ -871,7 +871,7 @@ To reply, use the add_post_to_bluesky_reply_thread tool: 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}") @@ -999,7 +999,7 @@ def save_notification_to_queue(notification, is_priority=None): 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) @@ -1078,14 +1078,18 @@ 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(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." @@ -1095,7 +1099,7 @@ def load_and_process_queued_notifications(void_agent, atproto_client, testing_mo 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 @@ -1288,7 +1292,7 @@ def fetch_and_queue_new_notifications(atproto_client): 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 @@ -1298,7 +1302,7 @@ def process_notifications(void_agent, atproto_client, testing_mode=False): 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}") @@ -1330,9 +1334,9 @@ def send_synthesis_message(client: Letta, agent_id: str, atproto_client=None) -> 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 @@ -1344,9 +1348,6 @@ These journal blocks are attached temporarily for this synthesis session. Use th 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.""" logger.info("๐Ÿง  Sending enhanced synthesis prompt to agent") @@ -1652,7 +1653,7 @@ def detach_temporal_blocks(client: Letta, agent_id: str, labels_to_detach: list 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)') @@ -1665,7 +1666,7 @@ def main(): # 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): @@ -1710,11 +1711,11 @@ def main(): 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: @@ -1756,22 +1757,33 @@ def main(): """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: @@ -1781,7 +1793,7 @@ def main(): # 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: @@ -1812,7 +1824,7 @@ def main(): 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...") @@ -1844,7 +1856,7 @@ def main(): 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: @@ -1852,13 +1864,13 @@ def main(): 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 diff --git a/list_agents.py b/list_agents.py new file mode 100644 index 0000000..20c92d4 --- /dev/null +++ b/list_agents.py @@ -0,0 +1,32 @@ +#!/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}") \ No newline at end of file diff --git a/tools/x_profile.py b/tools/x_profile.py new file mode 100644 index 0000000..1cc6168 --- /dev/null +++ b/tools/x_profile.py @@ -0,0 +1,134 @@ +"""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)}") \ No newline at end of file -- 2.51.2