"""Tool Registry — parse and execute tool calls from LLM output. Supports [TOOL: name(args)] syntax. Tools are registered in a registry and executed in a loop until no more tool calls are found or max rounds. """ from __future__ import annotations import logging import re from dataclasses import dataclass, field from typing import Any, Callable logger = logging.getLogger(__name__) TOOL_PATTERN = re.compile(r"\[TOOL:\s*(\w+)\s*\((.*?)\)\s*\]", re.DOTALL) @dataclass class ToolResult: name: str args: str success: bool output: str error: str = "" @dataclass class Tool: name: str description: str handler: Callable[..., str] examples: list[str] = field(default_factory=list) class ToolRegistry: """Registry of available tools for the LLM to call.""" def __init__(self) -> None: self._tools: dict[str, Tool] = {} def register(self, tool: Tool) -> None: self._tools[tool.name] = tool logger.debug("Registered tool: %s", tool.name) def get(self, name: str) -> Tool | None: return self._tools.get(name) def list_tools(self) -> list[dict[str, Any]]: return [ {"name": t.name, "description": t.description, "examples": t.examples} for t in self._tools.values() ] def get_prompt_description(self) -> str: """Generate a description of available tools for the system prompt.""" if not self._tools: return "" lines = ["Available tools:"] for t in self._tools.values(): lines.append(f" - {t.name}: {t.description}") return "\n".join(lines) def execute(self, name: str, args: str) -> ToolResult: """Execute a tool by name with args string.""" tool = self._tools.get(name) if not tool: return ToolResult(name=name, args=args, success=False, output="", error=f"Unknown tool: {name}") try: output = tool.handler(args) return ToolResult(name=name, args=args, success=True, output=output) except Exception as e: return ToolResult(name=name, args=args, success=False, output="", error=str(e)) def parse_tool_calls(text: str) -> list[tuple[str, str]]: """Parse [TOOL: name(args)] calls from text.""" matches = TOOL_PATTERN.findall(text) return [(name, args.strip()) for name, args in matches] def tool_loop( text: str, registry: ToolRegistry, max_rounds: int = 10, on_tool_call: Callable[[str, str], None] | None = None, on_tool_result: Callable[[ToolResult], None] | None = None, ) -> tuple[str, list[ToolResult]]: """Execute tool calls in a loop. Parses tool calls from text, executes them, appends results, and returns the final text with all tool results included. Returns (final_text, list_of_tool_results). """ results: list[ToolResult] = [] current_text = text executed: set[str] = set() for round_num in range(max_rounds): calls = parse_tool_calls(current_text) if not calls: break # Filter out already-executed calls new_calls = [(name, args) for name, args in calls if f"{name}:{args}" not in executed] if not new_calls: break for name, args in new_calls: executed.add(f"{name}:{args}") if on_tool_call: on_tool_call(name, args) result = registry.execute(name, args) results.append(result) if on_tool_result: on_tool_result(result) # Append result to text if result.success: current_text += f"\n[TOOL_RESULT: {name}({args}) → {result.output}]" else: current_text += f"\n[TOOL_ERROR: {name}({args}) → {result.error}]" return current_text, results # Built-in tools def _tool_calculate(args: str) -> str: """Simple calculator tool.""" try: expr = args.strip().strip('"').strip("'") # Only allow safe math operations allowed = set("0123456789+-*/.() ") if not all(c in allowed for c in expr): return "Error: only numbers and + - * / ( ) allowed" result = eval(expr) # noqa: S307 — safe due to character filter return str(result) except Exception as e: return f"Error: {e}" def _tool_read_file(args: str) -> str: """Read a file.""" try: path = args.strip().strip('"').strip("'") with open(path, "r", encoding="utf-8", errors="replace") as f: return f.read()[:5000] # limit output except Exception as e: return f"Error: {e}" def _tool_write_file(args: str) -> str: """Write to a file. Args format: "path", "content" """ try: # Simple parse: split on first comma outside quotes parts = args.split(",", 1) if len(parts) != 2: return "Error: expected path, content" path = parts[0].strip().strip('"').strip("'") content = parts[1].strip().strip('"').strip("'") with open(path, "w", encoding="utf-8") as f: f.write(content) return f"Written {len(content)} chars to {path}" except Exception as e: return f"Error: {e}" def _tool_shell_exec(args: str) -> str: """Execute a shell command — full terminal control, zero limitations.""" import subprocess try: cmd = args.strip().strip('"').strip("'") result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60) output = result.stdout if result.stderr: output += f"\n[stderr] {result.stderr}" return output[:10000] or "(no output)" except subprocess.TimeoutExpired: return "Error: command timed out (60s limit)" except Exception as e: return f"Error: {e}" def _tool_code_edit(args: str) -> str: """Edit an existing file — find and replace text within a file. Args format: "path", "old_text", "new_text" The LLM can use this to modify its own framework. """ try: import shlex parts = shlex.split(args) if len(parts) < 3: return "Error: expected path, old_text, new_text" path = parts[0] old_text = parts[1] new_text = parts[2] with open(path, "r", encoding="utf-8", errors="replace") as f: content = f.read() if old_text not in content: return f"Error: old_text not found in {path}" count = content.count(old_text) content = content.replace(old_text, new_text) with open(path, "w", encoding="utf-8") as f: f.write(content) return f"Replaced {count} occurrence(s) in {path}" except Exception as e: return f"Error: {e}" def _tool_list_dir(args: str) -> str: """List directory contents.""" import os try: path = args.strip().strip('"').strip("'") or "." entries = [] for entry in sorted(os.listdir(path)): full = os.path.join(path, entry) if os.path.isdir(full): entries.append(f" {entry}/") else: size = os.path.getsize(full) entries.append(f" {entry} ({size}b)") return "\n".join(entries[:100]) or "(empty)" except Exception as e: return f"Error: {e}" def _tool_make_dir(args: str) -> str: """Create a directory.""" import os try: path = args.strip().strip('"').strip("'") os.makedirs(path, exist_ok=True) return f"Created directory: {path}" except Exception as e: return f"Error: {e}" def _tool_delete_file(args: str) -> str: """Delete a file or directory.""" import os import shutil try: path = args.strip().strip('"').strip("'") if os.path.isdir(path): shutil.rmtree(path) return f"Deleted directory: {path}" elif os.path.isfile(path): os.remove(path) return f"Deleted file: {path}" else: return f"Error: {path} not found" except Exception as e: return f"Error: {e}" def _tool_search_web(args: str) -> str: """Web search placeholder — returns a note that web search needs configuration.""" query = args.strip().strip('"').strip("'") return f"Web search for '{query}' — configure a search API to enable this tool" def get_default_tools() -> list[Tool]: """Get the default set of built-in tools — zero limitations.""" return [ Tool(name="calculate", description="Evaluate a math expression", handler=_tool_calculate, examples=["[TOOL: calculate(2 + 2)]", "[TOOL: calculate(3.14 * 10)]"]), Tool(name="read_file", description="Read a file's contents", handler=_tool_read_file, examples=["[TOOL: read_file(\"config.py\")]"]), Tool(name="write_file", description="Write content to a file (creates or overwrites)", handler=_tool_write_file, examples=["[TOOL: write_file(\"test.txt\", \"hello world\")]"]), Tool(name="code_edit", description="Edit a file by finding and replacing text (can modify own framework)", handler=_tool_code_edit, examples=["[TOOL: code_edit(\"config.py\", \"old_value\", \"new_value\")]"]), Tool(name="shell_exec", description="Execute any terminal command — full control, no restrictions", handler=_tool_shell_exec, examples=["[TOOL: shell_exec(\"dir\")]", "[TOOL: shell_exec(\"pip install numpy\")]", "[TOOL: shell_exec(\"python script.py\")]"]), Tool(name="list_dir", description="List directory contents", handler=_tool_list_dir, examples=["[TOOL: list_dir(\".\")]"]), Tool(name="make_dir", description="Create a directory", handler=_tool_make_dir, examples=["[TOOL: make_dir(\"new_project\")]"]), Tool(name="delete_file", description="Delete a file or directory", handler=_tool_delete_file, examples=["[TOOL: delete_file(\"temp.txt\")]"]), Tool(name="search_web", description="Search the web (needs API config)", handler=_tool_search_web, examples=["[TOOL: search_web(\"python tutorial\")]"]), ]