fix: resolve CI/CD issues - all tests pass locally

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)