diff --git a/src/mcp_server_cli/tools/custom_tools.py b/src/mcp_server_cli/tools/custom_tools.py index 03d7c0e..e27e297 100644 --- a/src/mcp_server_cli/tools/custom_tools.py +++ b/src/mcp_server_cli/tools/custom_tools.py @@ -2,14 +2,14 @@ import asyncio import json -import os -from pathlib import Path -from typing import Any, Dict, List, Optional, Callable from datetime import datetime +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional + import yaml -from mcp_server_cli.tools.base import ToolBase, ToolResult, ToolRegistry -from mcp_server_cli.models import ToolSchema, ToolParameter, ToolDefinition +from mcp_server_cli.models import ToolDefinition, ToolParameter, ToolSchema +from mcp_server_cli.tools.base import ToolBase, ToolRegistry, ToolResult class CustomToolLoader: @@ -177,131 +177,3 @@ class CustomToolLoader: return ToolResult(success=False, output="", error="No executor configured") return DynamicTool(definition, executor) - - def register_tool_from_file( - self, - file_path: str, - executor: Optional[Callable[[Dict[str, Any]], Any]] = None, - ) -> Optional[ToolBase]: - """Load and register a tool from file. - - Args: - file_path: Path to tool definition file. - executor: Optional executor function. - - Returns: - Registered tool or None. - """ - tools = self.load_file(file_path) - for tool_def in tools: - tool = self.create_tool_from_definition(tool_def, executor) - self.registry.register(tool) - return tool - return None - - def reload_if_changed(self) -> List[ToolDefinition]: - """Reload tools if files have changed. - - Returns: - List of reloaded tool definitions. - """ - reloaded = [] - - for file_path, last_mtime in list(self._file_watchers.items()): - path = Path(file_path) - if not path.exists(): - continue - - current_mtime = path.stat().st_mtime - if current_mtime > last_mtime: - try: - tools = self.load_file(file_path) - reloaded.extend(tools) - self._file_watchers[file_path] = current_mtime - except Exception as e: - print(f"Warning: Failed to reload {file_path}: {e}") - - return reloaded - - def watch_file(self, file_path: str): - """Add a file to be watched for changes. - - Args: - file_path: Path to watch. - """ - path = Path(file_path) - if path.exists(): - self._file_watchers[file_path] = path.stat().st_mtime - - def list_loaded(self) -> Dict[str, Dict[str, Any]]: - """List all loaded custom tools. - - Returns: - Dictionary of tool name to metadata. - """ - return dict(self._loaded_tools) - - def get_registry(self) -> ToolRegistry: - """Get the internal tool registry. - - Returns: - ToolRegistry with all loaded tools. - """ - return self.registry - - -class DynamicTool(ToolBase): - """A dynamically created tool from a definition.""" - - def __init__( - self, - name: str, - description: str, - input_schema: ToolSchema, - executor: Callable[[Dict[str, Any]], Any], - annotations: Optional[Dict[str, Any]] = None, - ): - """Initialize a dynamic tool. - - Args: - name: Tool name. - description: Tool description. - input_schema: Tool input schema. - executor: Function to execute the tool. - annotations: Optional annotations. - """ - super().__init__(name=name, description=description, annotations=annotations) - self._input_schema = input_schema - self._executor = executor - - def _create_input_schema(self) -> ToolSchema: - return self._input_schema - - async def execute(self, arguments: Dict[str, Any]) -> ToolResult: - """Execute the dynamic tool.""" - try: - result = self._executor(arguments) - if asyncio.iscoroutine(result): - result = await result - return ToolResult(success=True, output=str(result)) - except Exception as e: - return ToolResult(success=False, output="", error=str(e)) - - -def create_python_executor(module_path: str, function_name: str) -> Callable: - """Create an executor from a Python function. - - Args: - module_path: Path to Python module. - function_name: Name of function to call. - - Returns: - Callable executor function. - """ - import importlib.util - - spec = importlib.util.spec_from_file_location("dynamic_tool", module_path) - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - - return getattr(module, function_name)