Update sktime/mcp_output/mcp_plugin/mcp_service.py

sktime/mcp_output/mcp_plugin/mcp_service.py

@@ -1,72 +1,586 @@
 import os
 import sys
+from typing import List, Dict, Any, Optional
+import json
 
-# Path settings to include the local source directory
+# Add source path for local sktime
 source_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "source")
 if source_path not in sys.path:
     sys.path.insert(0, source_path)
 
 from fastmcp import FastMCP
+import pandas as pd
+import numpy as np
 
-# Import core modules from sktime
-from sktime.forecasting.base import BaseForecaster
-from sktime.classification.base import BaseClassifier
-from sktime.transformations.base import BaseTransformer
+# Import sktime registry and core functions
+from sktime.registry import all_estimators, craft
+from sktime.datasets import (
+    load_airline,
+    load_longley,
+    load_lynx,
+    load_shampoo_sales,
+    load_gunpoint,
+    load_arrow_head,
+    load_italy_power_demand,
+)
 
 # Create the FastMCP service application
 mcp = FastMCP("sktime_service")
 
-@mcp.tool(name="forecasting_tool", description="Tool for forecasting time series data")
-def forecasting_tool(forecaster: BaseForecaster, data) -> dict:
+# Storage for created estimators
+_estimators: Dict[str, Any] = {}
+_datasets: Dict[str, Any] = {}
+
+
+@mcp.tool(name="list_forecasters", description="List all available forecasters in sktime")
+def list_forecasters(filter_tags: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    """
+    List all available forecasters in sktime.
+
+    Args:
+        filter_tags: Optional dictionary of tags to filter forecasters
+
+    Returns:
+        Dictionary with list of forecaster names and their descriptions
+    """
+    try:
+        estimators = all_estimators(
+            estimator_types="forecaster",
+            filter_tags=filter_tags,
+            as_dataframe=True
+        )
+        result = []
+        for _, row in estimators.iterrows():
+            result.append({
+                "name": row["name"],
+                "class": str(row["estimator"])
+            })
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="list_classifiers", description="List all available time series classifiers")
+def list_classifiers(filter_tags: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    """
+    List all available time series classifiers in sktime.
+
+    Args:
+        filter_tags: Optional dictionary of tags to filter classifiers
+
+    Returns:
+        Dictionary with list of classifier names
+    """
+    try:
+        estimators = all_estimators(
+            estimator_types="classifier",
+            filter_tags=filter_tags,
+            as_dataframe=True
+        )
+        result = []
+        for _, row in estimators.iterrows():
+            result.append({
+                "name": row["name"],
+                "class": str(row["estimator"])
+            })
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="list_transformers", description="List all available transformers")
+def list_transformers(filter_tags: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
     """
-    Perform forecasting on the provided time series data using the specified forecaster.
+    List all available transformers in sktime.
 
-    :param forecaster: An instance of a sktime BaseForecaster.
-    :param data: The time series data to forecast.
-    :return: A dictionary with success, result, and error fields.
+    Args:
+        filter_tags: Optional dictionary of tags to filter transformers
+
+    Returns:
+        Dictionary with list of transformer names
     """
     try:
-        result = forecaster.fit_predict(data)
+        estimators = all_estimators(
+            estimator_types="transformer",
+            filter_tags=filter_tags,
+            as_dataframe=True
+        )
+        result = []
+        for _, row in estimators.iterrows():
+            result.append({
+                "name": row["name"],
+                "class": str(row["estimator"])
+            })
         return {"success": True, "result": result, "error": None}
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="classification_tool", description="Tool for classifying time series data")
-def classification_tool(classifier: BaseClassifier, data, labels) -> dict:
+
+@mcp.tool(name="create_estimator", description="Create an sktime estimator from a specification string")
+def create_estimator(spec: str, estimator_id: str) -> Dict[str, Any]:
+    """
+    Create an sktime estimator using the craft function.
+
+    Args:
+        spec: Specification string for the estimator (e.g., "NaiveForecaster(strategy='last')")
+        estimator_id: Unique identifier to store the estimator
+
+    Returns:
+        Dictionary with success status and estimator info
+    """
+    try:
+        estimator = craft(spec)
+        _estimators[estimator_id] = estimator
+        return {
+            "success": True,
+            "result": {
+                "estimator_id": estimator_id,
+                "type": type(estimator).__name__,
+                "spec": str(estimator)
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="load_dataset", description="Load a built-in sktime dataset")
+def load_dataset(dataset_name: str, dataset_id: str) -> Dict[str, Any]:
+    """
+    Load a built-in sktime dataset.
+
+    Args:
+        dataset_name: Name of the dataset (airline, longley, lynx, shampoo_sales, gunpoint, arrow_head, italy_power_demand)
+        dataset_id: Unique identifier to store the dataset
+
+    Returns:
+        Dictionary with dataset info
+    """
+    try:
+        loaders = {
+            "airline": load_airline,
+            "longley": load_longley,
+            "lynx": load_lynx,
+            "shampoo_sales": load_shampoo_sales,
+            "gunpoint": load_gunpoint,
+            "arrow_head": load_arrow_head,
+            "italy_power_demand": load_italy_power_demand,
+        }
+
+        if dataset_name not in loaders:
+            return {
+                "success": False,
+                "result": None,
+                "error": f"Unknown dataset: {dataset_name}. Available: {list(loaders.keys())}"
+            }
+
+        data = loaders[dataset_name]()
+        _datasets[dataset_id] = data
+
+        # Get info about the dataset
+        if isinstance(data, tuple):
+            # Classification dataset (X, y)
+            X, y = data
+            info = {
+                "type": "classification",
+                "X_shape": str(X.shape) if hasattr(X, 'shape') else str(len(X)),
+                "y_shape": str(y.shape) if hasattr(y, 'shape') else str(len(y)),
+                "y_unique": list(np.unique(y)) if hasattr(y, '__iter__') else None
+            }
+        else:
+            # Forecasting dataset (single series)
+            info = {
+                "type": "forecasting",
+                "shape": str(data.shape) if hasattr(data, 'shape') else str(len(data)),
+                "index_type": type(data.index).__name__ if hasattr(data, 'index') else None,
+                "head": data.head().to_dict() if hasattr(data, 'head') else str(data[:5])
+            }
+
+        return {
+            "success": True,
+            "result": {
+                "dataset_id": dataset_id,
+                "dataset_name": dataset_name,
+                "info": info
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="create_series_from_list", description="Create a pandas Series from a list of values")
+def create_series_from_list(
+    values: List[float],
+    dataset_id: str,
+    freq: Optional[str] = None,
+    start_date: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Create a pandas Series from a list of values for use with sktime.
+
+    Args:
+        values: List of numeric values
+        dataset_id: Unique identifier to store the series
+        freq: Optional frequency string (e.g., 'M' for monthly, 'D' for daily)
+        start_date: Optional start date string (e.g., '2020-01-01')
+
+    Returns:
+        Dictionary with series info
+    """
+    try:
+        if freq and start_date:
+            index = pd.date_range(start=start_date, periods=len(values), freq=freq)
+        else:
+            index = pd.RangeIndex(start=0, stop=len(values))
+
+        series = pd.Series(values, index=index)
+        _datasets[dataset_id] = series
+
+        return {
+            "success": True,
+            "result": {
+                "dataset_id": dataset_id,
+                "length": len(series),
+                "index_type": type(series.index).__name__,
+                "head": series.head().to_dict()
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="fit_forecaster", description="Fit a forecaster on time series data")
+def fit_forecaster(estimator_id: str, dataset_id: str) -> Dict[str, Any]:
     """
-    Perform classification on the provided time series data using the specified classifier.
+    Fit a forecaster on time series data.
 
-    :param classifier: An instance of a sktime BaseClassifier.
-    :param data: The time series data to classify.
-    :param labels: The labels for the classification task.
-    :return: A dictionary with success, result, and error fields.
+    Args:
+        estimator_id: ID of the forecaster to fit
+        dataset_id: ID of the dataset to fit on
+
+    Returns:
+        Dictionary with fit status
     """
     try:
-        classifier.fit(data, labels)
-        predictions = classifier.predict(data)
-        return {"success": True, "result": predictions, "error": None}
+        if estimator_id not in _estimators:
+            return {"success": False, "result": None, "error": f"Estimator '{estimator_id}' not found"}
+        if dataset_id not in _datasets:
+            return {"success": False, "result": None, "error": f"Dataset '{dataset_id}' not found"}
+
+        estimator = _estimators[estimator_id]
+        data = _datasets[dataset_id]
+
+        estimator.fit(data)
+
+        return {
+            "success": True,
+            "result": {
+                "estimator_id": estimator_id,
+                "fitted": True,
+                "data_length": len(data)
+            },
+            "error": None
+        }
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="transformation_tool", description="Tool for transforming time series data")
-def transformation_tool(transformer: BaseTransformer, data) -> dict:
+
+@mcp.tool(name="predict_forecaster", description="Make predictions with a fitted forecaster")
+def predict_forecaster(
+    estimator_id: str,
+    forecast_horizon: List[int]
+) -> Dict[str, Any]:
     """
-    Apply transformations to the provided time series data using the specified transformer.
+    Make predictions with a fitted forecaster.
 
-    :param transformer: An instance of a sktime BaseTransformer.
-    :param data: The time series data to transform.
-    :return: A dictionary with success, result, and error fields.
+    Args:
+        estimator_id: ID of the fitted forecaster
+        forecast_horizon: List of integers representing forecast steps (e.g., [1, 2, 3] for 3-step forecast)
+
+    Returns:
+        Dictionary with predictions
     """
     try:
-        transformed_data = transformer.fit_transform(data)
-        return {"success": True, "result": transformed_data, "error": None}
+        if estimator_id not in _estimators:
+            return {"success": False, "result": None, "error": f"Estimator '{estimator_id}' not found"}
+
+        estimator = _estimators[estimator_id]
+        predictions = estimator.predict(fh=forecast_horizon)
+
+        return {
+            "success": True,
+            "result": {
+                "estimator_id": estimator_id,
+                "forecast_horizon": forecast_horizon,
+                "predictions": predictions.to_dict() if hasattr(predictions, 'to_dict') else predictions.tolist()
+            },
+            "error": None
+        }
     except Exception as e:
         return {"success": False, "result": None, "error": str(e)}
 
+
+@mcp.tool(name="fit_classifier", description="Fit a time series classifier")
+def fit_classifier(estimator_id: str, dataset_id: str) -> Dict[str, Any]:
+    """
+    Fit a time series classifier on labeled data.
+
+    Args:
+        estimator_id: ID of the classifier to fit
+        dataset_id: ID of the classification dataset (must be a tuple of X, y)
+
+    Returns:
+        Dictionary with fit status
+    """
+    try:
+        if estimator_id not in _estimators:
+            return {"success": False, "result": None, "error": f"Estimator '{estimator_id}' not found"}
+        if dataset_id not in _datasets:
+            return {"success": False, "result": None, "error": f"Dataset '{dataset_id}' not found"}
+
+        estimator = _estimators[estimator_id]
+        data = _datasets[dataset_id]
+
+        if isinstance(data, tuple):
+            X, y = data
+        else:
+            return {"success": False, "result": None, "error": "Dataset must be a (X, y) tuple for classification"}
+
+        estimator.fit(X, y)
+
+        return {
+            "success": True,
+            "result": {
+                "estimator_id": estimator_id,
+                "fitted": True,
+                "n_samples": len(y),
+                "classes": list(np.unique(y))
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="predict_classifier", description="Make predictions with a fitted classifier")
+def predict_classifier(estimator_id: str, dataset_id: str) -> Dict[str, Any]:
+    """
+    Make predictions with a fitted time series classifier.
+
+    Args:
+        estimator_id: ID of the fitted classifier
+        dataset_id: ID of the dataset to predict on
+
+    Returns:
+        Dictionary with predictions
+    """
+    try:
+        if estimator_id not in _estimators:
+            return {"success": False, "result": None, "error": f"Estimator '{estimator_id}' not found"}
+        if dataset_id not in _datasets:
+            return {"success": False, "result": None, "error": f"Dataset '{dataset_id}' not found"}
+
+        estimator = _estimators[estimator_id]
+        data = _datasets[dataset_id]
+
+        if isinstance(data, tuple):
+            X, _ = data
+        else:
+            X = data
+
+        predictions = estimator.predict(X)
+
+        return {
+            "success": True,
+            "result": {
+                "estimator_id": estimator_id,
+                "predictions": predictions.tolist() if hasattr(predictions, 'tolist') else list(predictions)
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="fit_transform", description="Fit and transform data using a transformer")
+def fit_transform(estimator_id: str, dataset_id: str, output_id: str) -> Dict[str, Any]:
+    """
+    Fit a transformer and transform data.
+
+    Args:
+        estimator_id: ID of the transformer
+        dataset_id: ID of the dataset to transform
+        output_id: ID to store the transformed data
+
+    Returns:
+        Dictionary with transformation result info
+    """
+    try:
+        if estimator_id not in _estimators:
+            return {"success": False, "result": None, "error": f"Estimator '{estimator_id}' not found"}
+        if dataset_id not in _datasets:
+            return {"success": False, "result": None, "error": f"Dataset '{dataset_id}' not found"}
+
+        estimator = _estimators[estimator_id]
+        data = _datasets[dataset_id]
+
+        transformed = estimator.fit_transform(data)
+        _datasets[output_id] = transformed
+
+        return {
+            "success": True,
+            "result": {
+                "estimator_id": estimator_id,
+                "input_id": dataset_id,
+                "output_id": output_id,
+                "output_shape": str(transformed.shape) if hasattr(transformed, 'shape') else str(len(transformed)),
+                "head": transformed.head().to_dict() if hasattr(transformed, 'head') else str(transformed[:5])
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="get_estimator_info", description="Get information about a stored estimator")
+def get_estimator_info(estimator_id: str) -> Dict[str, Any]:
+    """
+    Get detailed information about a stored estimator.
+
+    Args:
+        estimator_id: ID of the estimator
+
+    Returns:
+        Dictionary with estimator information
+    """
+    try:
+        if estimator_id not in _estimators:
+            return {"success": False, "result": None, "error": f"Estimator '{estimator_id}' not found"}
+
+        estimator = _estimators[estimator_id]
+
+        info = {
+            "estimator_id": estimator_id,
+            "type": type(estimator).__name__,
+            "spec": str(estimator),
+            "tags": estimator.get_tags() if hasattr(estimator, 'get_tags') else {},
+            "is_fitted": estimator.is_fitted if hasattr(estimator, 'is_fitted') else None
+        }
+
+        # Get parameters
+        if hasattr(estimator, 'get_params'):
+            params = estimator.get_params()
+            # Convert non-serializable params to strings
+            info["params"] = {k: str(v) if not isinstance(v, (str, int, float, bool, type(None))) else v
+                            for k, v in params.items()}
+
+        return {"success": True, "result": info, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="get_dataset_info", description="Get information about a stored dataset")
+def get_dataset_info(dataset_id: str) -> Dict[str, Any]:
+    """
+    Get detailed information about a stored dataset.
+
+    Args:
+        dataset_id: ID of the dataset
+
+    Returns:
+        Dictionary with dataset information
+    """
+    try:
+        if dataset_id not in _datasets:
+            return {"success": False, "result": None, "error": f"Dataset '{dataset_id}' not found"}
+
+        data = _datasets[dataset_id]
+
+        if isinstance(data, tuple):
+            X, y = data
+            info = {
+                "dataset_id": dataset_id,
+                "type": "classification",
+                "X_type": type(X).__name__,
+                "y_type": type(y).__name__,
+                "X_shape": str(X.shape) if hasattr(X, 'shape') else str(len(X)),
+                "y_shape": str(y.shape) if hasattr(y, 'shape') else str(len(y)),
+                "classes": list(np.unique(y))
+            }
+        elif hasattr(data, 'shape'):
+            info = {
+                "dataset_id": dataset_id,
+                "type": "series" if len(data.shape) == 1 else "dataframe",
+                "shape": str(data.shape),
+                "dtype": str(data.dtype) if hasattr(data, 'dtype') else None,
+                "index_type": type(data.index).__name__ if hasattr(data, 'index') else None,
+                "head": data.head().to_dict() if hasattr(data, 'head') else str(data[:5])
+            }
+        else:
+            info = {
+                "dataset_id": dataset_id,
+                "type": type(data).__name__,
+                "length": len(data) if hasattr(data, '__len__') else None
+            }
+
+        return {"success": True, "result": info, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="delete_estimator", description="Delete a stored estimator")
+def delete_estimator(estimator_id: str) -> Dict[str, Any]:
+    """
+    Delete a stored estimator.
+
+    Args:
+        estimator_id: ID of the estimator to delete
+
+    Returns:
+        Dictionary with deletion status
+    """
+    try:
+        if estimator_id not in _estimators:
+            return {"success": False, "result": None, "error": f"Estimator '{estimator_id}' not found"}
+
+        del _estimators[estimator_id]
+        return {"success": True, "result": {"deleted": estimator_id}, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="delete_dataset", description="Delete a stored dataset")
+def delete_dataset(dataset_id: str) -> Dict[str, Any]:
+    """
+    Delete a stored dataset.
+
+    Args:
+        dataset_id: ID of the dataset to delete
+
+    Returns:
+        Dictionary with deletion status
+    """
+    try:
+        if dataset_id not in _datasets:
+            return {"success": False, "result": None, "error": f"Dataset '{dataset_id}' not found"}
+
+        del _datasets[dataset_id]
+        return {"success": True, "result": {"deleted": dataset_id}, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
 def create_app() -> FastMCP:
     """
     Create and return the FastMCP service instance.
 
-    :return: An instance of FastMCP configured with the sktime service tools.
+    Returns:
+        An instance of FastMCP configured with the sktime service tools.
     """
-    return mcp
+    return mcp
+
+
+if __name__ == "__main__":
+    mcp.run(transport="http", host="0.0.0.0", port=8000)