Something went wrong. Try again.
Something went wrong. Try again.
void register_tools.py
6.6 kB · 182 lines
Python
at toolchange
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183#!/usr/bin/env python3"""Register all Void tools with a Letta agent."""import osimport sysimport loggingfrom typing import Listfrom dotenv import load_dotenvfrom letta_client import Lettafrom rich.console import Consolefrom rich.table import Table
# Import standalone functions and their schemasfrom tools.search import search_bluesky_posts, SearchArgsfrom tools.post import create_new_bluesky_post, PostArgsfrom tools.reply import reply_to_bluesky_post, ReplyArgsfrom tools.feed import get_bluesky_feed, FeedArgsfrom tools.blocks import attach_user_blocks, detach_user_blocks, AttachUserBlocksArgs, DetachUserBlocksArgs
load_dotenv()logging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)console = Console()
# Tool configurations: function paired with its args_schema and metadataTOOL_CONFIGS = [ { "func": search_bluesky_posts, "args_schema": SearchArgs, "description": "Search for posts on Bluesky matching the given criteria", "tags": ["bluesky", "search", "posts"] }, { "func": create_new_bluesky_post, "args_schema": PostArgs, "description": "Create a NEW standalone post on Bluesky. DO NOT use for replies.", "tags": ["bluesky", "post", "create", "new"] }, { "func": reply_to_bluesky_post, "args_schema": ReplyArgs, "description": "Reply to an existing Bluesky post. Requires the URI of the post being replied to.", "tags": ["bluesky", "reply", "thread", "conversation"] }, { "func": get_bluesky_feed, "args_schema": FeedArgs, "description": "Retrieve a Bluesky feed (home timeline or custom feed)", "tags": ["bluesky", "feed", "timeline"] }, { "func": attach_user_blocks, "args_schema": AttachUserBlocksArgs, "description": "Attach user-specific memory blocks to the agent. Creates blocks if they don't exist.", "tags": ["memory", "blocks", "user"] }, { "func": detach_user_blocks, "args_schema": DetachUserBlocksArgs, "description": "Detach user-specific memory blocks from the agent. Blocks are preserved for later use.", "tags": ["memory", "blocks", "user"] }, # { # "func": update_user_blocks, # "args_schema": UpdateUserBlockArgs, # "description": "Update the content of user-specific memory blocks", # "tags": ["memory", "blocks", "user"] # },]
def register_tools(agent_name: str = "void", tools: List[str] = None): """Register tools with a Letta agent.
Args: agent_name: Name of the agent to attach tools to tools: List of tool names to register. If None, registers all tools. """ try: # Initialize Letta client with API key client = Letta(token=os.environ["LETTA_API_KEY"])
# Find the agent agents = client.agents.list() agent = None for a in agents: if a.name == agent_name: agent = a break
if not agent: console.print(f"[red]Error: Agent '{agent_name}' not found[/red]") console.print("\nAvailable agents:") for a in agents: console.print(f" - {a.name}") return
# Filter tools if specific ones requested tools_to_register = TOOL_CONFIGS if tools: tools_to_register = [t for t in TOOL_CONFIGS if t["func"].__name__ in tools] if len(tools_to_register) != len(tools): missing = set(tools) - {t["func"].__name__ for t in tools_to_register} console.print(f"[yellow]Warning: Unknown tools: {missing}[/yellow]")
# Create results table table = Table(title=f"Tool Registration for Agent '{agent_name}'") table.add_column("Tool", style="cyan") table.add_column("Status", style="green") table.add_column("Description")
# Register each tool for tool_config in tools_to_register: func = tool_config["func"] tool_name = func.__name__
try: # Create or update the tool using the standalone function created_tool = client.tools.upsert_from_function( func=func, args_schema=tool_config["args_schema"], tags=tool_config["tags"] )
# Get current agent tools current_tools = client.agents.tools.list(agent_id=str(agent.id)) tool_names = [t.name for t in current_tools]
# Check if already attached if created_tool.name in tool_names: table.add_row(tool_name, "Already Attached", tool_config["description"]) else: # Attach to agent client.agents.tools.attach( agent_id=str(agent.id), tool_id=str(created_tool.id) ) table.add_row(tool_name, "✓ Attached", tool_config["description"])
except Exception as e: table.add_row(tool_name, f"✗ Error: {str(e)}", tool_config["description"]) logger.error(f"Error registering tool {tool_name}: {e}")
console.print(table)
except Exception as e: console.print(f"[red]Error: {str(e)}[/red]") logger.error(f"Fatal error: {e}")
def list_available_tools(): """List all available tools.""" table = Table(title="Available Void Tools") table.add_column("Tool Name", style="cyan") table.add_column("Description") table.add_column("Tags", style="dim")
for tool_config in TOOL_CONFIGS: table.add_row( tool_config["func"].__name__, tool_config["description"], ", ".join(tool_config["tags"]) )
console.print(table)
if __name__ == "__main__": import argparse
parser = argparse.ArgumentParser(description="Register Void tools with a Letta agent") parser.add_argument("agent", nargs="?", default="void", help="Agent name (default: void)") parser.add_argument("--tools", nargs="+", help="Specific tools to register (default: all)") parser.add_argument("--list", action="store_true", help="List available tools")
args = parser.parse_args()
if args.list: list_available_tools() else: console.print(f"\n[bold]Registering tools for agent: {args.agent}[/bold]\n") register_tools(args.agent, args.tools)