fix: resolve CI/CD issues - all tests pass locally
Some checks failed
CI / test (push) Has been cancelled

This commit is contained in:
2026-02-05 13:39:04 +00:00
parent 8a8e1172a2
commit 5b36551b34

View File

@@ -1,62 +1,91 @@
"""Base classes for MCP tools.""" """Base tool infrastructure for MCP Server CLI."""
from abc import ABC, abstractmethod 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 from mcp_server_cli.models import ToolParameter, ToolSchema
class ToolResult: class ToolResult(BaseModel):
"""Result of a tool execution.""" """Result of tool execution."""
def __init__( success: bool = True
self, output: str
success: bool, error: Optional[str] = None
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
class ToolBase(ABC): class ToolBase(ABC):
"""Abstract base class for all MCP tools.""" """Abstract base class for all MCP tools."""
def __init__(self, name: str, description: str): def __init__(
"""Initialize the tool. self,
name: str,
description: str,
annotations: Optional[Dict[str, Any]] = None,
):
"""Initialize a tool.
Args: Args:
name: Tool name. name: Tool name.
description: Tool description. description: Tool description.
annotations: Optional tool annotations.
""" """
self.name = name self.name = name
self.description = description self.description = description
self.annotations = annotations
@abstractmethod self._input_schema: Optional[ToolSchema] = None
def _create_input_schema(self) -> ToolSchema:
"""Create the input schema for the tool.
Returns:
ToolSchema defining the tool's input parameters.
"""
pass
@property @property
def input_schema(self) -> ToolSchema: 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: 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 @abstractmethod
async def execute(self, arguments: Dict[str, Any]) -> ToolResult: async def execute(self, arguments: Dict[str, Any]) -> ToolResult:
@@ -66,18 +95,67 @@ class ToolBase(ABC):
arguments: Tool arguments. arguments: Tool arguments.
Returns: Returns:
ToolResult with execution outcome. ToolResult with execution output.
""" """
pass 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: Returns:
Dictionary representation of the tool. Tool or None if not found.
""" """
return { return self._tools.get(name)
"name": self.name,
"description": self.description, def unregister(self, name: str) -> bool:
"input_schema": self.input_schema.model_dump(), """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()