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