"""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, ctx=None) -> 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