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, AttachUserBlocksArgs, DetachUserBlocksArgs
17from tools.reply import bluesky_reply, ReplyArgs
18
19load_dotenv()
20logging.basicConfig(level=logging.INFO)
21logger = logging.getLogger(__name__)
22console = Console()
23
24
25# Tool configurations: function paired with its args_schema and metadata
26TOOL_CONFIGS = [
27 {
28 "func": search_bluesky_posts,
29 "args_schema": SearchArgs,
30 "description": "Search for posts on Bluesky matching the given criteria",
31 "tags": ["bluesky", "search", "posts"]
32 },
33 {
34 "func": create_new_bluesky_post,
35 "args_schema": PostArgs,
36 "description": "Create a new Bluesky post or thread",
37 "tags": ["bluesky", "post", "create", "thread"]
38 },
39 # {
40 # "func": get_bluesky_feed,
41 # "args_schema": FeedArgs,
42 # "description": "Retrieve a Bluesky feed (home timeline or custom feed)",
43 # "tags": ["bluesky", "feed", "timeline"]
44 # },
45 {
46 "func": attach_user_blocks,
47 "args_schema": AttachUserBlocksArgs,
48 "description": "Attach user-specific memory blocks to the agent. Creates blocks if they don't exist.",
49 "tags": ["memory", "blocks", "user"]
50 },
51 {
52 "func": detach_user_blocks,
53 "args_schema": DetachUserBlocksArgs,
54 "description": "Detach user-specific memory blocks from the agent. Blocks are preserved for later use.",
55 "tags": ["memory", "blocks", "user"]
56 },
57 {
58 "func": bluesky_reply,
59 "args_schema": ReplyArgs,
60 "description": "Simple reply indicator for the Letta agent (max 300 chars)",
61 "tags": ["bluesky", "reply", "response"]
62 },
63]
64
65
66def register_tools(agent_name: str = "void", tools: List[str] = None):
67 """Register tools with a Letta agent.
68
69 Args:
70 agent_name: Name of the agent to attach tools to
71 tools: List of tool names to register. If None, registers all tools.
72 """
73 try:
74 # Initialize Letta client with API key
75 client = Letta(token=os.environ["LETTA_API_KEY"])
76
77 # Find the agent
78 agents = client.agents.list()
79 agent = None
80 for a in agents:
81 if a.name == agent_name:
82 agent = a
83 break
84
85 if not agent:
86 console.print(f"[red]Error: Agent '{agent_name}' not found[/red]")
87 console.print("\nAvailable agents:")
88 for a in agents:
89 console.print(f" - {a.name}")
90 return
91
92 # Filter tools if specific ones requested
93 tools_to_register = TOOL_CONFIGS
94 if tools:
95 tools_to_register = [t for t in TOOL_CONFIGS if t["func"].__name__ in tools]
96 if len(tools_to_register) != len(tools):
97 missing = set(tools) - {t["func"].__name__ for t in tools_to_register}
98 console.print(f"[yellow]Warning: Unknown tools: {missing}[/yellow]")
99
100 # Create results table
101 table = Table(title=f"Tool Registration for Agent '{agent_name}'")
102 table.add_column("Tool", style="cyan")
103 table.add_column("Status", style="green")
104 table.add_column("Description")
105
106 # Register each tool
107 for tool_config in tools_to_register:
108 func = tool_config["func"]
109 tool_name = func.__name__
110
111 try:
112 # Create or update the tool using the standalone function
113 created_tool = client.tools.upsert_from_function(
114 func=func,
115 args_schema=tool_config["args_schema"],
116 tags=tool_config["tags"]
117 )
118
119 # Get current agent tools
120 current_tools = client.agents.tools.list(agent_id=str(agent.id))
121 tool_names = [t.name for t in current_tools]
122
123 # Check if already attached
124 if created_tool.name in tool_names:
125 table.add_row(tool_name, "Already Attached", tool_config["description"])
126 else:
127 # Attach to agent
128 client.agents.tools.attach(
129 agent_id=str(agent.id),
130 tool_id=str(created_tool.id)
131 )
132 table.add_row(tool_name, "✓ Attached", tool_config["description"])
133
134 except Exception as e:
135 table.add_row(tool_name, f"✗ Error: {str(e)}", tool_config["description"])
136 logger.error(f"Error registering tool {tool_name}: {e}")
137
138 console.print(table)
139
140 except Exception as e:
141 console.print(f"[red]Error: {str(e)}[/red]")
142 logger.error(f"Fatal error: {e}")
143
144
145def list_available_tools():
146 """List all available tools."""
147 table = Table(title="Available Void Tools")
148 table.add_column("Tool Name", style="cyan")
149 table.add_column("Description")
150 table.add_column("Tags", style="dim")
151
152 for tool_config in TOOL_CONFIGS:
153 table.add_row(
154 tool_config["func"].__name__,
155 tool_config["description"],
156 ", ".join(tool_config["tags"])
157 )
158
159 console.print(table)
160
161
162if __name__ == "__main__":
163 import argparse
164
165 parser = argparse.ArgumentParser(description="Register Void tools with a Letta agent")
166 parser.add_argument("agent", nargs="?", default="void", help="Agent name (default: void)")
167 parser.add_argument("--tools", nargs="+", help="Specific tools to register (default: all)")
168 parser.add_argument("--list", action="store_true", help="List available tools")
169
170 args = parser.parse_args()
171
172 if args.list:
173 list_available_tools()
174 else:
175 console.print(f"\n[bold]Registering tools for agent: {args.agent}[/bold]\n")
176 register_tools(args.agent, args.tools)