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