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.blocks import (
12 attach_x_user_blocks, detach_x_user_blocks,
13 x_user_note_append, x_user_note_replace, x_user_note_set, x_user_note_view,
14 AttachXUserBlocksArgs, DetachXUserBlocksArgs,
15 XUserNoteAppendArgs, XUserNoteReplaceArgs, XUserNoteSetArgs, XUserNoteViewArgs
16)
17from tools.halt import halt_activity, HaltArgs
18from tools.ignore import ignore_notification, IgnoreNotificationArgs
19from tools.whitewind import create_whitewind_blog_post, WhitewindPostArgs
20from tools.ack import annotate_ack, AnnotateAckArgs
21from tools.webpage import fetch_webpage, WebpageArgs
22
23# Import X thread tool
24from tools.x_thread import add_post_to_x_thread, XThreadPostArgs
25
26# Import X search tool
27from tools.search_x import search_x_posts, SearchXArgs
28
29letta_config = get_letta_config()
30logging.basicConfig(level=logging.INFO)
31logger = logging.getLogger(__name__)
32console = Console()
33
34# X-specific tool configurations
35X_TOOL_CONFIGS = [
36 # Keep these core tools
37 {
38 "func": halt_activity,
39 "args_schema": HaltArgs,
40 "description": "Signal to halt all bot activity and terminate X bot",
41 "tags": ["control", "halt", "terminate"]
42 },
43 {
44 "func": ignore_notification,
45 "args_schema": IgnoreNotificationArgs,
46 "description": "Explicitly ignore an X notification without replying",
47 "tags": ["notification", "ignore", "control", "bot"]
48 },
49 {
50 "func": annotate_ack,
51 "args_schema": AnnotateAckArgs,
52 "description": "Add a note to the acknowledgment record for the current X interaction",
53 "tags": ["acknowledgment", "note", "annotation", "metadata"]
54 },
55 {
56 "func": create_whitewind_blog_post,
57 "args_schema": WhitewindPostArgs,
58 "description": "Create a blog post on Whitewind with markdown support",
59 "tags": ["whitewind", "blog", "post", "markdown"]
60 },
61 {
62 "func": fetch_webpage,
63 "args_schema": WebpageArgs,
64 "description": "Fetch a webpage and convert it to markdown/text format using Jina AI reader",
65 "tags": ["web", "fetch", "webpage", "markdown", "jina"]
66 },
67
68 # X user block management tools
69 {
70 "func": attach_x_user_blocks,
71 "args_schema": AttachXUserBlocksArgs,
72 "description": "Attach X user-specific memory blocks to the agent. Creates blocks if they don't exist.",
73 "tags": ["memory", "blocks", "user", "x", "twitter"]
74 },
75 {
76 "func": detach_x_user_blocks,
77 "args_schema": DetachXUserBlocksArgs,
78 "description": "Detach X user-specific memory blocks from the agent. Blocks are preserved for later use.",
79 "tags": ["memory", "blocks", "user", "x", "twitter"]
80 },
81 {
82 "func": x_user_note_append,
83 "args_schema": XUserNoteAppendArgs,
84 "description": "Append a note to an X user's memory block. Creates the block if it doesn't exist.",
85 "tags": ["memory", "blocks", "user", "append", "x", "twitter"]
86 },
87 {
88 "func": x_user_note_replace,
89 "args_schema": XUserNoteReplaceArgs,
90 "description": "Replace text in an X user's memory block.",
91 "tags": ["memory", "blocks", "user", "replace", "x", "twitter"]
92 },
93 {
94 "func": x_user_note_set,
95 "args_schema": XUserNoteSetArgs,
96 "description": "Set the complete content of an X user's memory block.",
97 "tags": ["memory", "blocks", "user", "set", "x", "twitter"]
98 },
99 {
100 "func": x_user_note_view,
101 "args_schema": XUserNoteViewArgs,
102 "description": "View the content of an X user's memory block.",
103 "tags": ["memory", "blocks", "user", "view", "x", "twitter"]
104 },
105
106 # X thread tool
107 {
108 "func": add_post_to_x_thread,
109 "args_schema": XThreadPostArgs,
110 "description": "Add a single post to the current X reply thread atomically",
111 "tags": ["x", "twitter", "reply", "thread", "atomic"]
112 },
113
114 # X search tool
115 {
116 "func": search_x_posts,
117 "args_schema": SearchXArgs,
118 "description": "Get recent posts from a specific X (Twitter) user",
119 "tags": ["x", "twitter", "search", "posts", "user"]
120 }
121]
122
123def register_x_tools(agent_id: str = None, tools: List[str] = None):
124 """Register X-specific tools with a Letta agent.
125
126 Args:
127 agent_id: ID of the agent to attach tools to. If None, uses config default.
128 tools: List of tool names to register. If None, registers all tools.
129 """
130 # Use agent ID from config if not provided
131 if agent_id is None:
132 agent_id = letta_config['agent_id']
133
134 try:
135 # Initialize Letta client with API key from config
136 client = Letta(token=letta_config['api_key'], timeout=letta_config['timeout'])
137
138 # Get the agent by ID
139 try:
140 agent = client.agents.retrieve(agent_id=agent_id)
141 except Exception as e:
142 console.print(f"[red]Error: Agent '{agent_id}' not found[/red]")
143 console.print(f"Error details: {e}")
144 return
145
146 # Filter tools if specific ones requested
147 tools_to_register = X_TOOL_CONFIGS
148 if tools:
149 tools_to_register = [t for t in X_TOOL_CONFIGS if t["func"] and t["func"].__name__ in tools]
150 if len(tools_to_register) != len(tools):
151 registered_names = {t["func"].__name__ for t in tools_to_register if t["func"]}
152 missing = set(tools) - registered_names
153 console.print(f"[yellow]Warning: Unknown tools: {missing}[/yellow]")
154
155 # Create results table
156 table = Table(title=f"X Tool Registration for Agent '{agent.name}' ({agent_id})")
157 table.add_column("Tool", style="cyan")
158 table.add_column("Status", style="green")
159 table.add_column("Description")
160
161 # Register each tool
162 for tool_config in tools_to_register:
163 func = tool_config["func"]
164 if not func:
165 continue
166
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_x_tools():
204 """List all available X tools."""
205 table = Table(title="Available X 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 X_TOOL_CONFIGS:
211 if tool_config["func"]:
212 table.add_row(
213 tool_config["func"].__name__,
214 tool_config["description"],
215 ", ".join(tool_config["tags"])
216 )
217
218 console.print(table)
219
220
221if __name__ == "__main__":
222 import argparse
223
224 parser = argparse.ArgumentParser(description="Register X tools with a Letta agent")
225 parser.add_argument("--agent-id", help=f"Agent ID (default: from config)")
226 parser.add_argument("--tools", nargs="+", help="Specific tools to register (default: all)")
227 parser.add_argument("--list", action="store_true", help="List available tools")
228
229 args = parser.parse_args()
230
231 if args.list:
232 list_available_x_tools()
233 else:
234 # Use config default if no agent specified
235 agent_id = args.agent_id if args.agent_id else letta_config['agent_id']
236 console.print(f"\n[bold]Registering X tools for agent: {agent_id}[/bold]\n")
237 register_x_tools(agent_id, args.tools)