From 5b36551b34b2a7051ecb0f995b0b572feee03940 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Thu, 5 Feb 2026 13:39:04 +0000 Subject: [PATCH] fix: resolve CI/CD issues - all tests pass locally --- src/mcp_server_cli/tools/base.py | 164 +++++++++++++++++++++++-------- 1 file changed, 121 insertions(+), 43 deletions(-) diff --git a/src/mcp_server_cli/tools/base.py b/src/mcp_server_cli/tools/base.py index 0277f2a..270d496 100644 --- a/src/mcp_server_cli/tools/base.py +++ b/src/mcp_server_cli/tools/base.py @@ -1,62 +1,91 @@ -"""Base classes for MCP tools.""" +"""Base tool infrastructure for MCP Server CLI.""" from abc import ABC, abstractmethod -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel from mcp_server_cli.models import ToolParameter, ToolSchema -class ToolResult: - """Result of a tool execution.""" +class ToolResult(BaseModel): + """Result of tool execution.""" - def __init__( - self, - success: bool, - output: str, - error: Optional[str] = None, - ): - """Initialize tool result. - - Args: - success: Whether the tool executed successfully. - output: Tool output. - error: Error message if failed. - """ - self.success = success - self.output = output - self.error = error + success: bool = True + output: str + error: Optional[str] = None class ToolBase(ABC): """Abstract base class for all MCP tools.""" - def __init__(self, name: str, description: str): - """Initialize the tool. + def __init__( + self, + name: str, + description: str, + annotations: Optional[Dict[str, Any]] = None, + ): + """Initialize a tool. Args: name: Tool name. description: Tool description. + annotations: Optional tool annotations. """ self.name = name self.description = description - - @abstractmethod - def _create_input_schema(self) -> ToolSchema: - """Create the input schema for the tool. - - Returns: - ToolSchema defining the tool's input parameters. - """ - pass + self.annotations = annotations + self._input_schema: Optional[ToolSchema] = None @property def input_schema(self) -> ToolSchema: - """Get the input schema for the tool. + """Get the tool's input schema.""" + if self._input_schema is None: + self._input_schema = self._create_input_schema() + return self._input_schema + + @abstractmethod + def _create_input_schema(self) -> ToolSchema: + """Create the tool's input schema. Returns: - ToolSchema defining the tool's input parameters. + ToolSchema defining expected parameters. """ - return self._create_input_schema() + pass + + def get_parameters(self) -> Dict[str, ToolParameter]: + """Get the tool's parameters. + + Returns: + Dictionary of parameter name to ToolParameter. + """ + return self.input_schema.properties + + def validate_arguments(self, arguments: Dict[str, Any]) -> Dict[str, Any]: + """Validate and sanitize tool arguments. + + Args: + arguments: Input arguments to validate. + + Returns: + Validated arguments. + """ + required = self.input_schema.required + properties = self.input_schema.properties + + for param_name in required: + if param_name not in arguments: + raise ValueError(f"Missing required parameter: {param_name}") + + validated = {} + for name, param in properties.items(): + if name in arguments: + value = arguments[name] + if param.enum and value not in param.enum: + raise ValueError(f"Invalid value for {name}: {value}") + validated[name] = value + + return validated @abstractmethod async def execute(self, arguments: Dict[str, Any]) -> ToolResult: @@ -66,18 +95,67 @@ class ToolBase(ABC): arguments: Tool arguments. Returns: - ToolResult with execution outcome. + ToolResult with execution output. """ pass - def model_dump(self) -> Dict[str, Any]: - """Convert tool to dictionary representation. + +class ToolRegistry: + """Registry for managing tools.""" + + def __init__(self): + """Initialize the tool registry.""" + self._tools: Dict[str, ToolBase] = {} + + def register(self, tool: ToolBase): + """Register a tool. + + Args: + tool: Tool to register. + """ + self._tools[tool.name] = tool + + def get(self, name: str) -> Optional[ToolBase]: + """Get a tool by name. + + Args: + name: Tool name. Returns: - Dictionary representation of the tool. + Tool or None if not found. """ - return { - "name": self.name, - "description": self.description, - "input_schema": self.input_schema.model_dump(), - } + return self._tools.get(name) + + def unregister(self, name: str) -> bool: + """Unregister a tool. + + Args: + name: Tool name. + + Returns: + True if tool was unregistered. + """ + if name in self._tools: + del self._tools[name] + return True + return False + + def list(self) -> List[ToolBase]: + """List all registered tools. + + Returns: + List of registered tools. + """ + return list(self._tools.values()) + + def list_names(self) -> List[str]: + """List all tool names. + + Returns: + List of tool names. + """ + return list(self._tools.keys()) + + def clear(self): + """Clear all registered tools.""" + self._tools.clear()