diff --git a/navi/core/registry.py b/navi/core/registry.py index 6ecbb23..725bddd 100644 --- a/navi/core/registry.py +++ b/navi/core/registry.py @@ -16,14 +16,22 @@ Tool, WebSearchTool, ) +from navi.tools.list_tools import ListToolsTool +from navi.tools.reload_tools import ReloadToolsTool +from navi.tools.tool_manual import ToolManualTool +from navi.tools.write_tool import WriteToolTool +from navi.tools.loader import LoadResult, load_tools_from_dir class ToolRegistry: def __init__(self) -> None: self._tools: dict[str, Tool] = {} + self._builtin_names: set[str] = set() - def register(self, tool: Tool) -> None: + def register(self, tool: Tool, builtin: bool = False) -> None: self._tools[tool.name] = tool + if builtin: + self._builtin_names.add(tool.name) def get(self, name: str) -> Tool: if name not in self._tools: @@ -36,6 +44,18 @@ def all(self) -> list[Tool]: return list(self._tools.values()) + def reload_user_tools(self, tools_dir: str) -> LoadResult: + """Remove all user tools and reload from disk. Safe: errors are isolated.""" + # Drop previously loaded user tools + for name in list(self._tools): + if name not in self._builtin_names: + del self._tools[name] + + result = load_tools_from_dir(tools_dir) + for tool in result.loaded: + self._tools[tool.name] = tool + return result + class ProfileRegistry: def __init__(self) -> None: @@ -74,13 +94,19 @@ """Build and populate registries with all built-in components.""" tools = ToolRegistry() - tools.register(WebSearchTool()) - tools.register(FilesystemTool()) - tools.register(HttpRequestTool()) - tools.register(CodeExecTool()) - tools.register(TerminalTool()) - tools.register(SshExecTool()) - tools.register(ImageViewTool()) + reload_tool = ReloadToolsTool(registry=tools) + write_tool = WriteToolTool(registry=tools) + list_tool = ListToolsTool(registry=tools) + manual_tool = ToolManualTool(registry=tools) + for builtin in [WebSearchTool(), FilesystemTool(), HttpRequestTool(), + CodeExecTool(), TerminalTool(), SshExecTool(), ImageViewTool(), + reload_tool, write_tool, list_tool, manual_tool]: + tools.register(builtin, builtin=True) + + # User-defined tools loaded from tools_dir + result = load_tools_from_dir(settings.tools_dir) + for user_tool in result.loaded: + tools.register(user_tool) profiles = ProfileRegistry() for p in ALL_PROFILES: diff --git a/navi/tools/list_tools.py b/navi/tools/list_tools.py new file mode 100644 index 0000000..45969d3 --- /dev/null +++ b/navi/tools/list_tools.py @@ -0,0 +1,29 @@ +"""Built-in tool that returns the current list of available tools.""" + +from .base import Tool, ToolResult + + +class ListToolsTool(Tool): + name = "list_tools" + description = ( + "Returns the actual list of tools currently available to you, with their descriptions. " + "Call this when asked what you can do, or before creating a new tool to check if it already exists." + ) + parameters = { + "type": "object", + "properties": {}, + "required": [], + } + + def __init__(self, registry=None) -> None: + self._registry = registry + + async def execute(self, params: dict) -> ToolResult: + if self._registry is None: + return ToolResult(success=False, output="Registry not available.", error="no_registry") + + lines = [] + for tool in self._registry.all(): + lines.append(f"• {tool.name}: {tool.description}") + + return ToolResult(success=True, output="\n".join(lines)) diff --git a/navi/tools/loader.py b/navi/tools/loader.py new file mode 100644 index 0000000..8b0620a --- /dev/null +++ b/navi/tools/loader.py @@ -0,0 +1,155 @@ +"""Dynamic tool loader — discovers tools from a directory. + +Two formats are supported: + +1. Module-level (preferred for user tools — no imports needed): + + name = "my_tool" + description = "What it does and when to use it." + parameters = {"type": "object", "properties": {...}, "required": [...]} + + async def execute(params: dict) -> str: + return "result" + + Return a string on success. Raise an exception on failure. + +2. Class-based (used internally): a class inheriting from navi.tools.base.Tool. + +Files starting with '_' are ignored. Errors are isolated per file. +""" + +import importlib.util +import inspect +from dataclasses import dataclass +from pathlib import Path + +import structlog + +from .base import Tool, ToolResult + +log = structlog.get_logger() + + +@dataclass +class LoadResult: + loaded: list[Tool] + errors: dict[str, str] # filename → error message + + +def load_tools_from_dir(tools_dir: str) -> LoadResult: + """Scan *tools_dir* for .py files and return instantiated Tool objects.""" + dir_path = Path(tools_dir) + loaded: list[Tool] = [] + errors: dict[str, str] = {} + + if not dir_path.exists(): + return LoadResult(loaded=loaded, errors=errors) + + for py_file in sorted(dir_path.glob("*.py")): + if py_file.name.startswith("_"): + continue + + file_tools, error = _load_file(py_file) + if error: + errors[py_file.name] = error + log.warning("tool_loader.file_error", file=py_file.name, error=error) + else: + loaded.extend(file_tools) + for t in file_tools: + log.info("tool_loader.loaded", name=t.name, file=py_file.name) + + return LoadResult(loaded=loaded, errors=errors) + + +def _load_file(py_file: Path) -> tuple[list[Tool], str | None]: + """Import one file and return (tools, error). Error is None on success.""" + spec = importlib.util.spec_from_file_location(f"user_tools.{py_file.stem}", py_file) + if spec is None or spec.loader is None: + return [], f"Cannot create module spec for {py_file.name}" + + module = importlib.util.module_from_spec(spec) + try: + spec.loader.exec_module(module) # type: ignore[union-attr] + except Exception as e: + return [], f"{type(e).__name__}: {e}" + + # --- Try module-level format first --- + tool = _try_module_level(module, py_file) + if tool is not None: + return [tool], None + + # --- Fall back to class-based format --- + return _try_class_based(module, py_file) + + +def _try_module_level(module, py_file: Path) -> Tool | None: + """Build a Tool from module-level name/description/parameters/execute.""" + has = lambda attr: hasattr(module, attr) # noqa: E731 + if not (has("name") and has("description") and has("parameters") and has("execute")): + return None + + name = module.name + description = module.description + parameters = module.parameters + execute_fn = module.execute + + if not callable(execute_fn): + return None + + async def _execute(self, params: dict) -> ToolResult: + try: + result = await execute_fn(params) + return ToolResult(success=True, output=str(result)) + except Exception as e: + return ToolResult(success=False, output=str(e), error=type(e).__name__) + + cls = type( + f"UserTool_{py_file.stem}", + (Tool,), + { + "name": name, + "description": description, + "parameters": parameters, + "execute": _execute, + }, + ) + return cls() + + +def _try_class_based(module, py_file: Path) -> tuple[list[Tool], str | None]: + """Discover Tool subclasses defined in the module.""" + tools: list[Tool] = [] + candidates: list[str] = [] + + for _, obj in inspect.getmembers(module, inspect.isclass): + if obj.__module__ != module.__name__: + continue + candidates.append(obj.__name__) + + if obj is Tool or not issubclass(obj, Tool) or inspect.isabstract(obj): + continue + + missing = [attr for attr in ("name", "description", "parameters") if not hasattr(obj, attr)] + if missing: + return [], ( + f"{obj.__name__} is missing required class attributes: {missing}. " + f"Add them as class-level variables (e.g. name = 'my_tool')." + ) + + sig = inspect.signature(obj.execute) + if list(sig.parameters.keys()) != ["self", "params"]: + return [], ( + f"{obj.__name__}.execute has wrong signature: {sig}. " + f"Required: async def execute(self, params: dict) -> ToolResult" + ) + + try: + tools.append(obj()) + except Exception as e: + return [], f"Failed to instantiate {obj.__name__}: {type(e).__name__}: {e}" + + if not tools: + hint = f" Classes found: {candidates}." if candidates else " No classes or module-level tool defined." + return [], f"No valid tool found in {py_file.name}.{hint}" + + return tools, None diff --git a/navi/tools/reload_tools.py b/navi/tools/reload_tools.py new file mode 100644 index 0000000..773bb51 --- /dev/null +++ b/navi/tools/reload_tools.py @@ -0,0 +1,46 @@ +"""Built-in tool to hot-reload user tools from the tools/ directory. + +The agent calls this after writing or editing a file in tools/. +Errors in individual tool files are reported without crashing anything. +""" + +from navi.config import settings + +from .base import Tool, ToolResult + + +class ReloadToolsTool(Tool): + name = "reload_tools" + description = ( + "Hot-reload all tools from the tools/ directory without restarting the server. " + "Call this after writing or editing a tool file. " + "Returns a report of what was loaded and any errors per file." + ) + parameters = { + "type": "object", + "properties": {}, + "required": [], + } + + def __init__(self, registry=None) -> None: # registry injected at startup + self._registry = registry + + async def execute(self, params: dict) -> ToolResult: + if self._registry is None: + return ToolResult(success=False, output="Tool registry not available.", error="no_registry") + + result = self._registry.reload_user_tools(settings.tools_dir) + + lines = [] + if result.loaded: + lines.append(f"Loaded ({len(result.loaded)}): {', '.join(t.name for t in result.loaded)}") + else: + lines.append("No tools loaded.") + + if result.errors: + lines.append(f"\nErrors ({len(result.errors)}):") + for filename, error in result.errors.items(): + lines.append(f" {filename}: {error}") + + success = not result.errors + return ToolResult(success=success, output="\n".join(lines)) diff --git a/navi/tools/tool_manual.py b/navi/tools/tool_manual.py new file mode 100644 index 0000000..d9e33a8 --- /dev/null +++ b/navi/tools/tool_manual.py @@ -0,0 +1,77 @@ +"""Built-in tool that returns the detailed manual for a given tool.""" + +from pathlib import Path + +from .base import Tool, ToolResult + +MANUALS_DIR = Path(__file__).parent.parent.parent / "manuals" + + +class ToolManualTool(Tool): + name = "tool_manual" + description = ( + "Returns the detailed manual for a tool: full usage instructions, parameter reference, and examples. " + "Call this before using an unfamiliar tool, or when you are unsure about the correct format or parameters." + ) + parameters = { + "type": "object", + "properties": { + "tool_name": { + "type": "string", + "description": "Name of the tool to look up, e.g. 'write_tool'", + } + }, + "required": ["tool_name"], + } + + def __init__(self, registry=None) -> None: + self._registry = registry + + async def execute(self, params: dict) -> ToolResult: + tool_name = params["tool_name"].strip() + + manual_file = MANUALS_DIR / f"{tool_name}.md" + if manual_file.exists(): + return ToolResult(success=True, output=manual_file.read_text(encoding="utf-8")) + + # No .md file — generate a manual from the tool's schema + if self._registry: + try: + tool = self._registry.get(tool_name) + return ToolResult(success=True, output=_auto_manual(tool)) + except Exception: + pass + + available = sorted(f.stem for f in MANUALS_DIR.glob("*.md")) if MANUALS_DIR.exists() else [] + hint = f"\nAvailable manuals: {', '.join(available)}" if available else "" + return ToolResult(success=False, output=f"No manual found for '{tool_name}'.{hint}", error="not_found") + + +def _auto_manual(tool) -> str: + """Generate a readable manual from a tool's schema.""" + lines = [f"# {tool.name}", "", tool.description, "", "## Parameters"] + + props = tool.parameters.get("properties", {}) + required = set(tool.parameters.get("required", [])) + + if not props: + lines.append("This tool takes no parameters.") + else: + for param_name, spec in props.items(): + req = " (required)" if param_name in required else " (optional)" + ptype = spec.get("type", "any") + desc = spec.get("description", "") + enum = spec.get("enum") + line = f"- `{param_name}` ({ptype}{req}): {desc}" + if enum: + line += f" — one of: {', '.join(repr(e) for e in enum)}" + lines.append(line) + + # Nested object properties + nested = spec.get("properties", {}) + for nname, nspec in nested.items(): + nreq = " (required)" if nname in spec.get("required", []) else " (optional)" + ndesc = nspec.get("description", "") + lines.append(f" - `{nname}` ({nspec.get('type', 'any')}{nreq}): {ndesc}") + + return "\n".join(lines) diff --git a/navi/tools/write_tool.py b/navi/tools/write_tool.py new file mode 100644 index 0000000..f2ff9e4 --- /dev/null +++ b/navi/tools/write_tool.py @@ -0,0 +1,113 @@ +"""Built-in tool that writes a new user tool file and immediately reloads it. + +This is the primary way for the agent to extend itself with new capabilities. +""" + +import json +from pathlib import Path + +from navi.config import settings + +from .base import Tool, ToolResult + + +def _register_user_tool(tool_name: str, tools_dir: Path) -> None: + """Add tool_name to tools/enabled.json so it's auto-included in all profiles.""" + enabled_file = tools_dir / "enabled.json" + try: + existing: list[str] = json.loads(enabled_file.read_text()) if enabled_file.exists() else [] + except Exception: + existing = [] + if tool_name not in existing: + existing.append(tool_name) + enabled_file.write_text(json.dumps(existing, indent=2)) + + +class WriteToolTool(Tool): + name = "write_tool" + description = ( + "Create a new permanent tool for yourself. " + "Writes the code to tools/.py and immediately reloads it — no server restart needed. " + "Use this whenever you need a capability you don't have yet: task tracking, reminders, " + "external APIs, data storage, calculations, etc. " + "The tool will be available in every future session, not just this one. " + "Read tools/_template.py first to see the exact required format." + ) + parameters = { + "type": "object", + "properties": { + "name": { + "type": "string", + "description": "Tool filename without .py extension, e.g. 'task_manager'. Also used as the tool's name attribute.", + }, + "code": { + "type": "string", + "description": ( + "Full Python source code for the tool. Must define exactly: " + "name (str), description (str), parameters (JSON Schema dict), " + "async def execute(params: dict) -> str. " + "No classes. No module-level print(). Return plain string. Raise on error." + ), + }, + }, + "required": ["name", "code"], + } + + def __init__(self, registry=None) -> None: + self._registry = registry + + async def execute(self, params: dict) -> ToolResult: + tool_name = params["name"].strip().removesuffix(".py") + code = params["code"] + + if not tool_name or tool_name.startswith("_"): + return ToolResult(success=False, output="Tool name must not be empty or start with '_'.", error="invalid_name") + + missing = [kw for kw in ("name", "description", "parameters", "async def execute") if kw not in code] + if missing: + example = 'name = "tool_name"\ndescription = "..."\nparameters = {"type": "object", "properties": {}, "required": []}\n\nasync def execute(params: dict) -> str:\n ...' + return ToolResult( + success=False, + output=( + f"Code rejected — missing required module-level definitions: {missing}.\n\n" + f"Every tool file must start with these four definitions before any other code:\n{example}\n\n" + f"Your execute() function logic can follow after. Also make sure execute is async." + ), + error="invalid_code", + ) + + tools_dir = Path(settings.tools_dir) + tools_dir.mkdir(exist_ok=True) + file_path = tools_dir / f"{tool_name}.py" + + try: + file_path.write_text(code, encoding="utf-8") + except OSError as e: + return ToolResult(success=False, output=f"Failed to write file: {e}", error="write_error") + + if self._registry is None: + return ToolResult(success=False, output=f"Written to {file_path}, but registry unavailable for reload.", error="no_registry") + + result = self._registry.reload_user_tools(settings.tools_dir) + + lines = [] + if result.loaded: + lines.append(f"Loaded ({len(result.loaded)}): {', '.join(t.name for t in result.loaded)}") + if result.errors: + lines.append(f"\nErrors ({len(result.errors)}):") + for filename, error in result.errors.items(): + lines.append(f" {filename}: {error}") + + if file_path.name in result.errors: + return ToolResult( + success=False, + output=f"File written but failed to load:\n {result.errors[file_path.name]}\n\nFix the code and call write_tool again.", + error="load_error", + ) + + loaded_names = [t.name for t in result.loaded] + if tool_name in loaded_names: + _register_user_tool(tool_name, tools_dir) + return ToolResult(success=True, output=f"Tool '{tool_name}' created and enabled globally.\n" + "\n".join(lines)) + + return ToolResult(success=True, output="\n".join(lines) if lines else "Done.")