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