bunware.org
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_config, get_letta_config, get_agent_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.blocks import attach_user_blocks, detach_user_blocks, user_note_append, user_note_replace, user_note_set, user_note_view, AttachUserBlocksArgs, DetachUserBlocksArgs, UserNoteAppendArgs, UserNoteReplaceArgs, UserNoteSetArgs, UserNoteViewArgs
17from tools.halt import halt_activity, HaltArgs
18from tools.thread import add_post_to_bluesky_reply_thread, ReplyThreadPostArgs
19from tools.ignore import ignore_notification, IgnoreNotificationArgs
20
21config = get_config()
22letta_config = get_letta_config()
23agent_config = get_agent_config()
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": attach_user_blocks,
51 "args_schema": AttachUserBlocksArgs,
52 "description": "Attach user-specific memory blocks to the agent. Creates blocks if they don't exist.",
53 "tags": ["memory", "blocks", "user"]
54 },
55 {
56 "func": detach_user_blocks,
57 "args_schema": DetachUserBlocksArgs,
58 "description": "Detach user-specific memory blocks from the agent. Blocks are preserved for later use.",
59 "tags": ["memory", "blocks", "user"]
60 },
61 {
62 "func": user_note_append,
63 "args_schema": UserNoteAppendArgs,
64 "description": "Append a note to a user's memory block. Creates the block if it doesn't exist.",
65 "tags": ["memory", "blocks", "user", "append"]
66 },
67 {
68 "func": user_note_replace,
69 "args_schema": UserNoteReplaceArgs,
70 "description": "Replace text in a user's memory block.",
71 "tags": ["memory", "blocks", "user", "replace"]
72 },
73 {
74 "func": user_note_set,
75 "args_schema": UserNoteSetArgs,
76 "description": "Set the complete content of a user's memory block.",
77 "tags": ["memory", "blocks", "user", "set"]
78 },
79 {
80 "func": user_note_view,
81 "args_schema": UserNoteViewArgs,
82 "description": "View the content of a user's memory block.",
83 "tags": ["memory", "blocks", "user", "view"]
84 },
85 {
86 "func": halt_activity,
87 "args_schema": HaltArgs,
88 "description": "Signal to halt all bot activity and terminate bsky.py",
89 "tags": ["control", "halt", "terminate"]
90 },
91 {
92 "func": add_post_to_bluesky_reply_thread,
93 "args_schema": ReplyThreadPostArgs,
94 "description": "Add a single post to the current Bluesky reply thread atomically",
95 "tags": ["bluesky", "reply", "thread", "atomic"]
96 },
97 {
98 "func": ignore_notification,
99 "args_schema": IgnoreNotificationArgs,
100 "description": "Explicitly ignore a notification without replying (useful for ignoring bot interactions)",
101 "tags": ["notification", "ignore", "control", "bot"]
102 },
103]
104
105
106def register_tools(agent_name: str = None, tools: List[str] = None):
107 """Register tools with a Letta agent.
108
109 Args:
110 agent_name: Name of the agent to attach tools to. If None, uses config default.
111 tools: List of tool names to register. If None, registers all tools.
112 """
113 # Use agent name from config if not provided
114 if agent_name is None:
115 agent_name = agent_config['name']
116
117 try:
118 # Initialize Letta client with API key from config
119 client = Letta(token=letta_config['api_key'])
120
121 # Find the agent
122 agents = client.agents.list()
123 agent = None
124 for a in agents:
125 if a.name == agent_name:
126 agent = a
127 break
128
129 if not agent:
130 console.print(f"[red]Error: Agent '{agent_name}' not found[/red]")
131 console.print("\nAvailable agents:")
132 for a in agents:
133 console.print(f" - {a.name}")
134 return
135
136 # Filter tools if specific ones requested
137 tools_to_register = TOOL_CONFIGS
138 if tools:
139 tools_to_register = [t for t in TOOL_CONFIGS if t["func"].__name__ in tools]
140 if len(tools_to_register) != len(tools):
141 missing = set(tools) - {t["func"].__name__ for t in tools_to_register}
142 console.print(f"[yellow]Warning: Unknown tools: {missing}[/yellow]")
143
144 # Create results table
145 table = Table(title=f"Tool Registration for Agent '{agent_name}'")
146 table.add_column("Tool", style="cyan")
147 table.add_column("Status", style="green")
148 table.add_column("Description")
149
150 # Register each tool
151 for tool_config in tools_to_register:
152 func = tool_config["func"]
153 tool_name = func.__name__
154
155 try:
156 # Create or update the tool using the standalone function
157 created_tool = client.tools.upsert_from_function(
158 func=func,
159 args_schema=tool_config["args_schema"],
160 tags=tool_config["tags"]
161 )
162
163 # Get current agent tools
164 current_tools = client.agents.tools.list(agent_id=str(agent.id))
165 tool_names = [t.name for t in current_tools]
166
167 # Check if already attached
168 if created_tool.name in tool_names:
169 table.add_row(tool_name, "Already Attached", tool_config["description"])
170 else:
171 # Attach to agent
172 client.agents.tools.attach(
173 agent_id=str(agent.id),
174 tool_id=str(created_tool.id)
175 )
176 table.add_row(tool_name, "✓ Attached", tool_config["description"])
177
178 except Exception as e:
179 table.add_row(tool_name, f"✗ Error: {str(e)}", tool_config["description"])
180 logger.error(f"Error registering tool {tool_name}: {e}")
181
182 console.print(table)
183
184 except Exception as e:
185 console.print(f"[red]Error: {str(e)}[/red]")
186 logger.error(f"Fatal error: {e}")
187
188
189def list_available_tools():
190 """List all available tools."""
191 table = Table(title="Available Void Tools")
192 table.add_column("Tool Name", style="cyan")
193 table.add_column("Description")
194 table.add_column("Tags", style="dim")
195
196 for tool_config in TOOL_CONFIGS:
197 table.add_row(
198 tool_config["func"].__name__,
199 tool_config["description"],
200 ", ".join(tool_config["tags"])
201 )
202
203 console.print(table)
204
205
206if __name__ == "__main__":
207 import argparse
208
209 parser = argparse.ArgumentParser(description="Register Void tools with a Letta agent")
210 parser.add_argument("agent", nargs="?", default=None, help=f"Agent name (default: {agent_config['name']})")
211 parser.add_argument("--tools", nargs="+", help="Specific tools to register (default: all)")
212 parser.add_argument("--list", action="store_true", help="List available tools")
213
214 args = parser.parse_args()
215
216 if args.list:
217 list_available_tools()
218 else:
219 # Use config default if no agent specified
220 agent_name = args.agent if args.agent is not None else agent_config['name']
221 console.print(f"\n[bold]Registering tools for agent: {agent_name}[/bold]\n")
222 register_tools(args.agent, args.tools)