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