a digital person for bluesky
cameron.stream
1#!/usr/bin/env python3
2"""Register all Void tools with a Letta agent."""
3import os
4import sys
5import logging
6from typing import List
7from letta_client import Letta
8from rich.console import Console
9from rich.table import Table
10from config_loader import get_letta_config, get_bluesky_config, get_config
11
12# Import standalone functions and their schemas
13from tools.search import search_bluesky_posts, SearchArgs
14from tools.post import create_new_bluesky_post, PostArgs
15from tools.feed import get_bluesky_feed, FeedArgs
16from tools.halt import halt_activity, HaltArgs
17from tools.thread import add_post_to_bluesky_reply_thread, ReplyThreadPostArgs
18from tools.ignore import ignore_notification, IgnoreNotificationArgs
19from tools.whitewind import create_whitewind_blog_post, WhitewindPostArgs
20from tools.ack import annotate_ack, AnnotateAckArgs
21from tools.webpage import fetch_webpage, WebpageArgs
22from tools.flag_memory_deletion import flag_archival_memory_for_deletion, FlagArchivalMemoryForDeletionArgs
23
24logging.basicConfig(level=logging.INFO)
25logger = logging.getLogger(__name__)
26console = Console()
27
28
29# Tool configurations: function paired with its args_schema and metadata
30TOOL_CONFIGS = [
31 {
32 "func": search_bluesky_posts,
33 "args_schema": SearchArgs,
34 "description": "Search for posts on Bluesky matching the given criteria",
35 "tags": ["bluesky", "search", "posts"]
36 },
37 {
38 "func": create_new_bluesky_post,
39 "args_schema": PostArgs,
40 "description": "Create a new Bluesky post or thread",
41 "tags": ["bluesky", "post", "create", "thread"]
42 },
43 {
44 "func": get_bluesky_feed,
45 "args_schema": FeedArgs,
46 "description": "Retrieve a Bluesky feed (home timeline or custom feed)",
47 "tags": ["bluesky", "feed", "timeline"]
48 },
49 {
50 "func": halt_activity,
51 "args_schema": HaltArgs,
52 "description": "Signal to halt all bot activity and terminate bsky.py",
53 "tags": ["control", "halt", "terminate"]
54 },
55 {
56 "func": add_post_to_bluesky_reply_thread,
57 "args_schema": ReplyThreadPostArgs,
58 "description": "Add a single post to the current Bluesky reply thread atomically",
59 "tags": ["bluesky", "reply", "thread", "atomic"]
60 },
61 {
62 "func": ignore_notification,
63 "args_schema": IgnoreNotificationArgs,
64 "description": "Explicitly ignore a notification without replying (useful for ignoring bot interactions)",
65 "tags": ["notification", "ignore", "control", "bot"]
66 },
67 {
68 "func": create_whitewind_blog_post,
69 "args_schema": WhitewindPostArgs,
70 "description": "Create a blog post on Whitewind with markdown support",
71 "tags": ["whitewind", "blog", "post", "markdown"]
72 },
73 {
74 "func": annotate_ack,
75 "args_schema": AnnotateAckArgs,
76 "description": "Add a note to the acknowledgment record for the current post interaction",
77 "tags": ["acknowledgment", "note", "annotation", "metadata"]
78 },
79 {
80 "func": fetch_webpage,
81 "args_schema": WebpageArgs,
82 "description": "Fetch a webpage and convert it to markdown/text format using Jina AI reader",
83 "tags": ["web", "fetch", "webpage", "markdown", "jina"]
84 },
85 {
86 "func": flag_archival_memory_for_deletion,
87 "args_schema": FlagArchivalMemoryForDeletionArgs,
88 "description": "Flag an archival memory for deletion based on its exact text content",
89 "tags": ["memory", "archival", "delete", "cleanup"]
90 },
91]
92
93
94def register_tools(agent_id: str = None, tools: List[str] = None, set_env: bool = True):
95 """Register tools with a Letta agent.
96
97 Args:
98 agent_id: ID of the agent to attach tools to. If None, uses config default.
99 tools: List of tool names to register. If None, registers all tools.
100 set_env: If True, set environment variables for tool execution. Defaults to True.
101 """
102 # Load config fresh (uses global config instance from get_config())
103 letta_config = get_letta_config()
104
105 # Use agent ID from config if not provided
106 if agent_id is None:
107 agent_id = letta_config['agent_id']
108
109 try:
110 # Initialize Letta client with API key and base_url from config
111 client_params = {
112 'token': letta_config['api_key'],
113 'timeout': letta_config['timeout']
114 }
115 if letta_config.get('base_url'):
116 client_params['base_url'] = letta_config['base_url']
117 client = Letta(**client_params)
118
119 # Get the agent by ID
120 try:
121 agent = client.agents.retrieve(agent_id=agent_id)
122 except Exception as e:
123 console.print(f"[red]Error: Agent '{agent_id}' not found[/red]")
124 console.print(f"Error details: {e}")
125 return
126
127 # Set environment variables for tool execution if requested
128 if set_env:
129 try:
130 bsky_config = get_bluesky_config()
131 env_vars = {
132 'BSKY_USERNAME': bsky_config['username'],
133 'BSKY_PASSWORD': bsky_config['password'],
134 'PDS_URI': bsky_config['pds_uri']
135 }
136
137 console.print(f"\n[bold cyan]Setting tool execution environment variables:[/bold cyan]")
138 console.print(f" BSKY_USERNAME: {env_vars['BSKY_USERNAME']}")
139 console.print(f" PDS_URI: {env_vars['PDS_URI']}")
140 console.print(f" BSKY_PASSWORD: {'*' * len(env_vars['BSKY_PASSWORD'])}\n")
141
142 # Modify agent with environment variables
143 client.agents.modify(
144 agent_id=agent_id,
145 tool_exec_environment_variables=env_vars
146 )
147
148 console.print("[green]✓ Environment variables set successfully[/green]\n")
149 except Exception as e:
150 console.print(f"[yellow]Warning: Failed to set environment variables: {e}[/yellow]\n")
151 logger.warning(f"Failed to set environment variables: {e}")
152
153 # Filter tools if specific ones requested
154 tools_to_register = TOOL_CONFIGS
155 if tools:
156 tools_to_register = [t for t in TOOL_CONFIGS if t["func"].__name__ in tools]
157 if len(tools_to_register) != len(tools):
158 missing = set(tools) - {t["func"].__name__ for t in tools_to_register}
159 console.print(f"[yellow]Warning: Unknown tools: {missing}[/yellow]")
160
161 # Create results table
162 table = Table(title=f"Tool Registration for Agent '{agent.name}' ({agent_id})")
163 table.add_column("Tool", style="cyan")
164 table.add_column("Status", style="green")
165 table.add_column("Description")
166
167 # Register each tool
168 for tool_config in tools_to_register:
169 func = tool_config["func"]
170 tool_name = func.__name__
171
172 try:
173 # Create or update the tool using the standalone function
174 created_tool = client.tools.upsert_from_function(
175 func=func,
176 args_schema=tool_config["args_schema"],
177 tags=tool_config["tags"]
178 )
179
180 # Get current agent tools
181 current_tools = client.agents.tools.list(agent_id=str(agent.id))
182 tool_names = [t.name for t in current_tools]
183
184 # Check if already attached
185 if created_tool.name in tool_names:
186 table.add_row(tool_name, "Already Attached", tool_config["description"])
187 else:
188 # Attach to agent
189 client.agents.tools.attach(
190 agent_id=str(agent.id),
191 tool_id=str(created_tool.id)
192 )
193 table.add_row(tool_name, "✓ Attached", tool_config["description"])
194
195 except Exception as e:
196 table.add_row(tool_name, f"✗ Error: {str(e)}", tool_config["description"])
197 logger.error(f"Error registering tool {tool_name}: {e}")
198
199 console.print(table)
200
201 except Exception as e:
202 console.print(f"[red]Error: {str(e)}[/red]")
203 logger.error(f"Fatal error: {e}")
204
205
206def list_available_tools():
207 """List all available tools."""
208 table = Table(title="Available Void Tools")
209 table.add_column("Tool Name", style="cyan")
210 table.add_column("Description")
211 table.add_column("Tags", style="dim")
212
213 for tool_config in TOOL_CONFIGS:
214 table.add_row(
215 tool_config["func"].__name__,
216 tool_config["description"],
217 ", ".join(tool_config["tags"])
218 )
219
220 console.print(table)
221
222
223if __name__ == "__main__":
224 import argparse
225
226 parser = argparse.ArgumentParser(description="Register Void tools with a Letta agent")
227 parser.add_argument("--config", type=str, default='configs/config.yaml', help="Path to config file (default: configs/config.yaml)")
228 parser.add_argument("--agent-id", help=f"Agent ID (default: from config)")
229 parser.add_argument("--tools", nargs="+", help="Specific tools to register (default: all)")
230 parser.add_argument("--list", action="store_true", help="List available tools")
231 parser.add_argument("--no-env", action="store_true", help="Skip setting environment variables")
232
233 args = parser.parse_args()
234
235 # Initialize config with custom path (sets global config instance)
236 get_config(args.config)
237
238 if args.list:
239 list_available_tools()
240 else:
241 # Load config and get agent ID
242 letta_config = get_letta_config()
243 agent_id = args.agent_id if args.agent_id else letta_config['agent_id']
244 console.print(f"\n[bold]Registering tools for agent: {agent_id}[/bold]\n")
245 register_tools(agent_id, args.tools, set_env=not args.no_env)