cameron.stream
a digital person for bluesky
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 'attach_x_user_blocks',
28 'detach_x_user_blocks',
29 'x_user_note_append',
30 'x_user_note_replace',
31 'x_user_note_set',
32 'x_user_note_view',
33}
34
35# Common tools shared across platforms
36COMMON_TOOLS = {
37 'halt_activity',
38 'ignore_notification',
39 'annotate_ack',
40 'blog_post_create',
41 'fetch_webpage',
42}
43
44
45def ensure_platform_tools(platform: str, agent_id: str = None, api_key: str = None) -> None:
46 """
47 Ensure the correct tools are attached for the specified platform.
48
49 This function will:
50 1. Detach tools that belong to other platforms
51 2. Keep common tools attached
52 3. Ensure platform-specific tools are attached
53
54 Args:
55 platform: Either 'bluesky' or 'x'
56 agent_id: Agent ID to manage tools for (uses config default if None)
57 api_key: Letta API key to use (uses config default if None)
58 """
59 if platform not in ['bluesky', 'x']:
60 raise ValueError(f"Platform must be 'bluesky' or 'x', got '{platform}'")
61
62 letta_config = get_letta_config()
63 agent_config = get_agent_config()
64
65 # Use agent ID from config if not provided
66 if agent_id is None:
67 agent_id = letta_config.get('agent_id', agent_config.get('id'))
68
69 # Use API key from parameter or config
70 if api_key is None:
71 api_key = letta_config['api_key']
72
73 try:
74 # Initialize Letta client with proper base_url for self-hosted servers
75 client_params = {'api_key': api_key} # v1.0: token → api_key
76 if letta_config.get('base_url'):
77 client_params['base_url'] = letta_config['base_url']
78 client = Letta(**client_params)
79
80 # Get the agent
81 try:
82 agent = client.agents.retrieve(agent_id=agent_id)
83 logger.info(f"Managing tools for agent '{agent.name}' ({agent_id}) for platform '{platform}'")
84 except Exception as e:
85 logger.error(f"Could not retrieve agent {agent_id}: {e}")
86 return
87
88 # Get current attached tools
89 current_tools = client.agents.tools.list(agent_id=str(agent.id))
90 current_tool_names = {tool.name for tool in current_tools}
91 current_tool_mapping = {tool.name: tool for tool in current_tools}
92
93 # Determine which tools to keep and which to remove
94 if platform == 'bluesky':
95 tools_to_keep = BLUESKY_TOOLS | COMMON_TOOLS
96 tools_to_remove = X_TOOLS
97 required_tools = BLUESKY_TOOLS
98 else: # platform == 'x'
99 tools_to_keep = X_TOOLS | COMMON_TOOLS
100 tools_to_remove = BLUESKY_TOOLS
101 required_tools = X_TOOLS
102
103 # Detach tools that shouldn't be on this platform
104 tools_to_detach = tools_to_remove & current_tool_names
105 for tool_name in tools_to_detach:
106 try:
107 tool = current_tool_mapping[tool_name]
108 client.agents.tools.detach(
109 agent_id=str(agent.id),
110 tool_id=str(tool.id)
111 )
112 logger.info(f"Detached {tool_name} (not needed for {platform})")
113 except Exception as e:
114 logger.error(f"Failed to detach {tool_name}: {e}")
115
116 # Check which required tools are missing
117 missing_tools = required_tools - current_tool_names
118
119 if missing_tools:
120 logger.info(f"Missing {len(missing_tools)} {platform} tools: {missing_tools}")
121 logger.info(f"Please run the appropriate registration script:")
122 if platform == 'bluesky':
123 logger.info(" python register_tools.py")
124 else:
125 logger.info(" python register_x_tools.py")
126 else:
127 logger.info(f"All required {platform} tools are already attached")
128
129 # Log final state
130 remaining_tools = (current_tool_names - tools_to_detach) & tools_to_keep
131 logger.info(f"Tools configured for {platform}: {len(remaining_tools)} tools active")
132
133 except Exception as e:
134 logger.error(f"Error managing platform tools: {e}")
135 raise
136
137
138def get_attached_tools(agent_id: str = None, api_key: str = None) -> Set[str]:
139 """
140 Get the currently attached tools for an agent.
141
142 Args:
143 agent_id: Agent ID to check (uses config default if None)
144 api_key: Letta API key to use (uses config default if None)
145
146 Returns:
147 Set of tool names currently attached
148 """
149 letta_config = get_letta_config()
150 agent_config = get_agent_config()
151
152 # Use agent ID from config if not provided
153 if agent_id is None:
154 agent_id = letta_config.get('agent_id', agent_config.get('id'))
155
156 # Use API key from parameter or config
157 if api_key is None:
158 api_key = letta_config['api_key']
159
160 try:
161 # Initialize Letta client with proper base_url for self-hosted servers
162 client_params = {'api_key': api_key} # v1.0: token → api_key
163 if letta_config.get('base_url'):
164 client_params['base_url'] = letta_config['base_url']
165 client = Letta(**client_params)
166 agent = client.agents.retrieve(agent_id=agent_id)
167 # v1.0: list returns page object
168 current_tools_page = client.agents.tools.list(agent_id=str(agent.id))
169 current_tools = current_tools_page.items if hasattr(current_tools_page, 'items') else current_tools_page
170 return {tool.name for tool in current_tools}
171 except Exception as e:
172 logger.error(f"Error getting attached tools: {e}")
173 return set()
174
175
176if __name__ == "__main__":
177 import argparse
178
179 parser = argparse.ArgumentParser(description="Manage platform-specific tools for Void agent")
180 parser.add_argument("platform", choices=['bluesky', 'x'], nargs='?', help="Platform to configure tools for")
181 parser.add_argument("--agent-id", help="Agent ID (default: from config)")
182 parser.add_argument("--list", action="store_true", help="List current tools without making changes")
183
184 args = parser.parse_args()
185
186 if args.list:
187 tools = get_attached_tools(args.agent_id)
188 print(f"\nCurrently attached tools ({len(tools)}):")
189 for tool in sorted(tools):
190 platform_indicator = ""
191 if tool in BLUESKY_TOOLS:
192 platform_indicator = " [Bluesky]"
193 elif tool in X_TOOLS:
194 platform_indicator = " [X]"
195 elif tool in COMMON_TOOLS:
196 platform_indicator = " [Common]"
197 print(f" - {tool}{platform_indicator}")
198 else:
199 if not args.platform:
200 parser.error("platform is required when not using --list")
201 ensure_platform_tools(args.platform, args.agent_id)