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