a digital person for bluesky
cameron.stream
1#!/usr/bin/env python3
2"""Register X-specific tools with a Letta agent."""
3import logging
4from typing import List
5from letta_client import Letta
6from rich.console import Console
7from rich.table import Table
8from config_loader import get_letta_config
9
10# Import standalone functions and their schemas
11from tools.halt import halt_activity, HaltArgs
12from tools.ignore import ignore_notification, IgnoreNotificationArgs
13from tools.whitewind import create_whitewind_blog_post, WhitewindPostArgs
14from tools.ack import annotate_ack, AnnotateAckArgs
15from tools.webpage import fetch_webpage, WebpageArgs
16
17# Import X thread tool
18from tools.x_thread import add_post_to_x_thread, XThreadPostArgs
19
20# Import X post tool
21from tools.x_post import post_to_x, PostToXArgs
22
23# Import X search tools
24from tools.search_x import search_x_posts, SearchXArgs
25from tools.x_profile import search_x_profile, SearchXProfileArgs
26
27# Import X user block management tools
28from tools.blocks import (
29 attach_x_user_blocks, AttachXUserBlocksArgs,
30 detach_x_user_blocks, DetachXUserBlocksArgs,
31 x_user_note_append, XUserNoteAppendArgs,
32 x_user_note_replace, XUserNoteReplaceArgs,
33 x_user_note_set, XUserNoteSetArgs,
34 x_user_note_view, XUserNoteViewArgs
35)
36
37letta_config = get_letta_config()
38logging.basicConfig(level=logging.INFO)
39logger = logging.getLogger(__name__)
40console = Console()
41
42# X-specific tool configurations
43X_TOOL_CONFIGS = [
44 # Keep these core tools
45 {
46 "func": halt_activity,
47 "args_schema": HaltArgs,
48 "description": "Signal to halt all bot activity and terminate X bot",
49 "tags": ["control", "halt", "terminate"]
50 },
51 {
52 "func": ignore_notification,
53 "args_schema": IgnoreNotificationArgs,
54 "description": "Explicitly ignore an X notification without replying",
55 "tags": ["notification", "ignore", "control", "bot"]
56 },
57 {
58 "func": annotate_ack,
59 "args_schema": AnnotateAckArgs,
60 "description": "Add a note to the acknowledgment record for the current X interaction",
61 "tags": ["acknowledgment", "note", "annotation", "metadata"]
62 },
63 {
64 "func": create_whitewind_blog_post,
65 "args_schema": WhitewindPostArgs,
66 "description": "Create a blog post on Whitewind with markdown support",
67 "tags": ["whitewind", "blog", "post", "markdown"]
68 },
69 {
70 "func": fetch_webpage,
71 "args_schema": WebpageArgs,
72 "description": "Fetch a webpage and convert it to markdown/text format using Jina AI reader",
73 "tags": ["web", "fetch", "webpage", "markdown", "jina"]
74 },
75
76 # X thread tool
77 {
78 "func": add_post_to_x_thread,
79 "args_schema": XThreadPostArgs,
80 "description": "Add a single post to the current X reply thread atomically",
81 "tags": ["x", "twitter", "reply", "thread", "atomic"]
82 },
83
84 # X post tool
85 {
86 "func": post_to_x,
87 "args_schema": PostToXArgs,
88 "description": "Create a new standalone post on X (Twitter)",
89 "tags": ["x", "twitter", "post", "create", "standalone"]
90 },
91
92 # X search tools
93 {
94 "func": search_x_posts,
95 "args_schema": SearchXArgs,
96 "description": "Get recent posts from a specific X (Twitter) user",
97 "tags": ["x", "twitter", "search", "posts", "user"]
98 },
99 {
100 "func": search_x_profile,
101 "args_schema": SearchXProfileArgs,
102 "description": "Look up detailed profile information for an X (Twitter) user including recent posts and conversations",
103 "tags": ["x", "twitter", "profile", "lookup", "user", "posts", "activity"]
104 },
105
106 # X user block management tools
107 {
108 "func": attach_x_user_blocks,
109 "args_schema": AttachXUserBlocksArgs,
110 "description": "Attach X user-specific memory blocks to the agent",
111 "tags": ["x", "user", "memory", "block", "attach"]
112 },
113 {
114 "func": detach_x_user_blocks,
115 "args_schema": DetachXUserBlocksArgs,
116 "description": "Detach X user-specific memory blocks from the agent",
117 "tags": ["x", "user", "memory", "block", "detach"]
118 },
119 {
120 "func": x_user_note_append,
121 "args_schema": XUserNoteAppendArgs,
122 "description": "Append a note to an X user's memory block",
123 "tags": ["x", "user", "memory", "note", "append"]
124 },
125 {
126 "func": x_user_note_replace,
127 "args_schema": XUserNoteReplaceArgs,
128 "description": "Replace text in an X user's memory block",
129 "tags": ["x", "user", "memory", "note", "replace"]
130 },
131 {
132 "func": x_user_note_set,
133 "args_schema": XUserNoteSetArgs,
134 "description": "Set the complete content of an X user's memory block",
135 "tags": ["x", "user", "memory", "note", "set"]
136 },
137 {
138 "func": x_user_note_view,
139 "args_schema": XUserNoteViewArgs,
140 "description": "View the content of an X user's memory block",
141 "tags": ["x", "user", "memory", "note", "view"]
142 }
143]
144
145def register_x_tools(agent_id: str = None, tools: List[str] = None):
146 """Register X-specific tools with a Letta agent.
147
148 Args:
149 agent_id: ID of the agent to attach tools to. If None, uses config default.
150 tools: List of tool names to register. If None, registers all tools.
151 """
152 # Use agent ID from config if not provided
153 if agent_id is None:
154 agent_id = letta_config['agent_id']
155
156 try:
157 # Initialize Letta client with API key from config
158 client = Letta(token=letta_config['api_key'], timeout=letta_config['timeout'])
159
160 # Get the agent by ID
161 try:
162 agent = client.agents.retrieve(agent_id=agent_id)
163 except Exception as e:
164 console.print(f"[red]Error: Agent '{agent_id}' not found[/red]")
165 console.print(f"Error details: {e}")
166 return
167
168 # Filter tools if specific ones requested
169 tools_to_register = X_TOOL_CONFIGS
170 if tools:
171 tools_to_register = [t for t in X_TOOL_CONFIGS if t["func"] and t["func"].__name__ in tools]
172 if len(tools_to_register) != len(tools):
173 registered_names = {t["func"].__name__ for t in tools_to_register if t["func"]}
174 missing = set(tools) - registered_names
175 console.print(f"[yellow]Warning: Unknown tools: {missing}[/yellow]")
176
177 # Create results table
178 table = Table(title=f"X Tool Registration for Agent '{agent.name}' ({agent_id})")
179 table.add_column("Tool", style="cyan")
180 table.add_column("Status", style="green")
181 table.add_column("Description")
182
183 # Register each tool
184 for tool_config in tools_to_register:
185 func = tool_config["func"]
186 if not func:
187 continue
188
189 tool_name = func.__name__
190
191 try:
192 # Create or update the tool using the standalone function
193 created_tool = client.tools.upsert_from_function(
194 func=func,
195 args_schema=tool_config["args_schema"],
196 tags=tool_config["tags"]
197 )
198
199 # Get current agent tools
200 current_tools = client.agents.tools.list(agent_id=str(agent.id))
201 tool_names = [t.name for t in current_tools]
202
203 # Check if already attached
204 if created_tool.name in tool_names:
205 table.add_row(tool_name, "Already Attached", tool_config["description"])
206 else:
207 # Attach to agent
208 client.agents.tools.attach(
209 agent_id=str(agent.id),
210 tool_id=str(created_tool.id)
211 )
212 table.add_row(tool_name, "✓ Attached", tool_config["description"])
213
214 except Exception as e:
215 table.add_row(tool_name, f"✗ Error: {str(e)}", tool_config["description"])
216 logger.error(f"Error registering tool {tool_name}: {e}")
217
218 console.print(table)
219
220 except Exception as e:
221 console.print(f"[red]Error: {str(e)}[/red]")
222 logger.error(f"Fatal error: {e}")
223
224
225def list_available_x_tools():
226 """List all available X tools."""
227 table = Table(title="Available X Tools")
228 table.add_column("Tool Name", style="cyan")
229 table.add_column("Description")
230 table.add_column("Tags", style="dim")
231
232 for tool_config in X_TOOL_CONFIGS:
233 if tool_config["func"]:
234 table.add_row(
235 tool_config["func"].__name__,
236 tool_config["description"],
237 ", ".join(tool_config["tags"])
238 )
239
240 console.print(table)
241
242
243if __name__ == "__main__":
244 import argparse
245
246 parser = argparse.ArgumentParser(description="Register X tools with a Letta agent")
247 parser.add_argument("--agent-id", help=f"Agent ID (default: from config)")
248 parser.add_argument("--tools", nargs="+", help="Specific tools to register (default: all)")
249 parser.add_argument("--list", action="store_true", help="List available tools")
250
251 args = parser.parse_args()
252
253 if args.list:
254 list_available_x_tools()
255 else:
256 # Use config default if no agent specified
257 agent_id = args.agent_id if args.agent_id else letta_config['agent_id']
258 console.print(f"\n[bold]Registering X tools for agent: {agent_id}[/bold]\n")
259 register_x_tools(agent_id, args.tools)