a digital person for bluesky
cameron.stream
1#!/usr/bin/env python3
2"""Platform-specific tool management for Void agent."""
3import logging
4from typing import List, Set
5from letta_client import Letta
6from config_loader import get_letta_config, get_agent_config
7
8logger = logging.getLogger(__name__)
9
10# Define platform-specific tool sets
11BLUESKY_TOOLS = {
12 'search_bluesky_posts',
13 'create_new_bluesky_post',
14 'get_bluesky_feed',
15 'add_post_to_bluesky_reply_thread',
16 'attach_user_blocks',
17 'detach_user_blocks',
18 'user_note_append',
19 'user_note_replace',
20 'user_note_set',
21 'user_note_view',
22}
23
24X_TOOLS = {
25 'add_post_to_x_thread',
26 'search_x_posts',
27 'detach_x_user_blocks',
28 'attach_x_user_blocks',
29 'update_x_user_blocks',
30 'search_x_profile',
31}
32
33# Common tools shared across platforms
34COMMON_TOOLS = {
35 'halt_activity',
36 'ignore_notification',
37 'annotate_ack',
38 'create_whitewind_blog_post',
39 'fetch_webpage',
40 'web_search',
41 'research_bluesky_profile',
42}
43
44# Core Letta tools that should always be present
45CORE_LETTA_TOOLS = {
46 'archival_memory_insert',
47 'archival_memory_search',
48 'conversation_search',
49 'memory_insert',
50 'memory_replace',
51 'memory_rethink',
52 'send_message',
53}
54
55
56def ensure_platform_tools(platform: str, agent_id: str = None) -> None:
57 """
58 Ensure the correct tools are attached for the specified platform.
59
60 This function will:
61 1. Detach tools that belong to other platforms
62 2. Keep common tools attached
63 3. Ensure platform-specific tools are attached
64
65 Args:
66 platform: Either 'bluesky' or 'x'
67 agent_id: Agent ID to manage tools for (uses config default if None)
68 """
69 if platform not in ['bluesky', 'x']:
70 raise ValueError(f"Platform must be 'bluesky' or 'x', got '{platform}'")
71
72 letta_config = get_letta_config()
73 agent_config = get_agent_config()
74
75 # Use agent ID from config if not provided
76 if agent_id is None:
77 agent_id = letta_config.get('agent_id', agent_config.get('id'))
78
79 try:
80 # Initialize Letta client
81 client = Letta(token=letta_config['api_key'])
82
83 # Get the agent
84 try:
85 agent = client.agents.retrieve(agent_id=agent_id)
86 logger.info(f"Managing tools for agent '{agent.name}' ({agent_id}) for platform '{platform}'")
87 except Exception as e:
88 logger.error(f"Could not retrieve agent {agent_id}: {e}")
89 return
90
91 # Get current attached tools
92 current_tools = client.agents.tools.list(agent_id=str(agent.id))
93 current_tool_names = {tool.name for tool in current_tools}
94 current_tool_mapping = {tool.name: tool for tool in current_tools}
95
96 # Determine which tools to keep and which to remove
97 if platform == 'bluesky':
98 tools_to_keep = BLUESKY_TOOLS | COMMON_TOOLS | CORE_LETTA_TOOLS
99 tools_to_remove = X_TOOLS
100 required_tools = BLUESKY_TOOLS
101 else: # platform == 'x'
102 tools_to_keep = X_TOOLS | COMMON_TOOLS | CORE_LETTA_TOOLS
103 tools_to_remove = BLUESKY_TOOLS
104 required_tools = X_TOOLS
105
106 # Detach tools that shouldn't be on this platform
107 tools_to_detach = tools_to_remove & current_tool_names
108 for tool_name in tools_to_detach:
109 try:
110 tool = current_tool_mapping[tool_name]
111 client.agents.tools.detach(
112 agent_id=str(agent.id),
113 tool_id=str(tool.id)
114 )
115 logger.info(f"Detached {tool_name} (not needed for {platform})")
116 except Exception as e:
117 logger.error(f"Failed to detach {tool_name}: {e}")
118
119 # Check which required tools are missing
120 missing_tools = required_tools - current_tool_names
121
122 if missing_tools:
123 logger.info(f"Missing {len(missing_tools)} {platform} tools: {missing_tools}")
124 logger.info(f"Please run the appropriate registration script:")
125 if platform == 'bluesky':
126 logger.info(" python register_tools.py")
127 else:
128 logger.info(" python register_x_tools.py")
129 else:
130 logger.info(f"All required {platform} tools are already attached")
131
132 # Log final state
133 remaining_tools = (current_tool_names - tools_to_detach) & tools_to_keep
134 logger.info(f"Tools configured for {platform}: {len(remaining_tools)} tools active")
135
136 except Exception as e:
137 logger.error(f"Error managing platform tools: {e}")
138 raise
139
140
141def get_attached_tools(agent_id: str = None) -> Set[str]:
142 """
143 Get the currently attached tools for an agent.
144
145 Args:
146 agent_id: Agent ID to check (uses config default if None)
147
148 Returns:
149 Set of tool names currently attached
150 """
151 letta_config = get_letta_config()
152 agent_config = get_agent_config()
153
154 # Use agent ID from config if not provided
155 if agent_id is None:
156 agent_id = letta_config.get('agent_id', agent_config.get('id'))
157
158 try:
159 client = Letta(token=letta_config['api_key'])
160 agent = client.agents.retrieve(agent_id=agent_id)
161 current_tools = client.agents.tools.list(agent_id=str(agent.id))
162 return {tool.name for tool in current_tools}
163 except Exception as e:
164 logger.error(f"Error getting attached tools: {e}")
165 return set()
166
167
168if __name__ == "__main__":
169 import argparse
170
171 parser = argparse.ArgumentParser(description="Manage platform-specific tools for Void agent")
172 parser.add_argument("platform", choices=['bluesky', 'x'], nargs='?', help="Platform to configure tools for")
173 parser.add_argument("--agent-id", help="Agent ID (default: from config)")
174 parser.add_argument("--list", action="store_true", help="List current tools without making changes")
175
176 args = parser.parse_args()
177
178 if args.list:
179 tools = get_attached_tools(args.agent_id)
180 print(f"\nCurrently attached tools ({len(tools)}):")
181 for tool in sorted(tools):
182 platform_indicator = ""
183 if tool in BLUESKY_TOOLS:
184 platform_indicator = " [Bluesky]"
185 elif tool in X_TOOLS:
186 platform_indicator = " [X]"
187 elif tool in COMMON_TOOLS:
188 platform_indicator = " [Common]"
189 elif tool in CORE_LETTA_TOOLS:
190 platform_indicator = " [Core]"
191 print(f" - {tool}{platform_indicator}")
192 else:
193 if not args.platform:
194 parser.error("platform is required when not using --list")
195 ensure_platform_tools(args.platform, args.agent_id)