Update sktime/mcp_output/mcp_plugin/mcp_service.py

sktime/mcp_output/mcp_plugin/mcp_service.py

@@ -1,586 +1,157 @@
-import os
-import sys
-from typing import List, Dict, Any, Optional
-import json
-
-# 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 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")
 
-# 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]:
-    """
-    List all available transformers in sktime.
-
-    Args:
-        filter_tags: Optional dictionary of tags to filter transformers
-
-    Returns:
-        Dictionary with list of transformer names
-    """
-    try:
-        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="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]:
-    """
-    Fit a forecaster on time series data.
-
-    Args:
-        estimator_id: ID of the forecaster to fit
-        dataset_id: ID of the dataset to fit on
-
-    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]
-
-        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="predict_forecaster", description="Make predictions with a fitted forecaster")
-def predict_forecaster(
-    estimator_id: str,
-    forecast_horizon: List[int]
-) -> Dict[str, Any]:
-    """
-    Make predictions with a fitted forecaster.
-
-    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:
-        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.
-
-    Returns:
-        An instance of FastMCP configured with the sktime service tools.
-    """
-    return mcp
+@mcp.tool(name="list_available_datasets", description="List all available sktime datasets")
+def list_available_datasets():
+    """List all available datasets in sktime."""
+    from sktime.datasets import get_dataset_names
+    return get_dataset_names()
+
+@mcp.tool(name="load_dataset", description="Load a dataset from sktime")
+def load_dataset(dataset_name: str):
+    """Load a dataset by name from sktime.
+
+    Parameters
+    ----------
+    dataset_name : str
+        The name of the dataset to load.
+
+    Returns
+    -------
+    Dataset object.
+    """
+    from sktime.datasets import load_from_tsfile
+    return load_from_tsfile(dataset_name)
+
+@mcp.tool(name="time_series_classification", description="Perform time series classification using sktime")
+def time_series_classification(data, classifier_name: str):
+    """Perform time series classification.
+
+    Parameters
+    ----------
+    data : pd.DataFrame
+        Input time series data.
+    classifier_name : str
+        Name of the classifier to use.
+
+    Returns
+    -------
+    Classification results.
+    """
+    from sktime.classification.all import all_estimators
+    from sktime.classification.compose import make_pipeline
+
+    classifiers = dict(all_estimators())
+    if classifier_name not in classifiers:
+        raise ValueError(f"Classifier {classifier_name} not found in sktime.")
+
+    classifier = classifiers[classifier_name]()
+    pipeline = make_pipeline(classifier)
+    return pipeline.fit_predict(data)
+
+@mcp.tool(name="time_series_forecasting", description="Perform time series forecasting using sktime")
+def time_series_forecasting(data, forecaster_name: str, fh):
+    """Perform time series forecasting.
+
+    Parameters
+    ----------
+    data : pd.DataFrame
+        Input time series data.
+    forecaster_name : str
+        Name of the forecaster to use.
+    fh : ForecastingHorizon
+        Forecasting horizon.
+
+    Returns
+    -------
+    Forecast results.
+    """
+    from sktime.forecasting.all import all_estimators
+    from sktime.forecasting.compose import make_reduction
+
+    forecasters = dict(all_estimators())
+    if forecaster_name not in forecasters:
+        raise ValueError(f"Forecaster {forecaster_name} not found in sktime.")
+
+    forecaster = forecasters[forecaster_name]()
+    pipeline = make_reduction(forecaster)
+    return pipeline.fit_predict(data, fh)
+
+@mcp.tool(name="time_series_transformation", description="Apply transformations to time series data using sktime")
+def time_series_transformation(data, transformer_name: str):
+    """Apply transformations to time series data.
+
+    Parameters
+    ----------
+    data : pd.DataFrame
+        Input time series data.
+    transformer_name : str
+        Name of the transformer to use.
+
+    Returns
+    -------
+    Transformed data.
+    """
+    from sktime.transformations.all import all_estimators
+
+    transformers = dict(all_estimators())
+    if transformer_name not in transformers:
+        raise ValueError(f"Transformer {transformer_name} not found in sktime.")
+
+    transformer = transformers[transformer_name]()
+    return transformer.fit_transform(data)
+
+@mcp.tool(name="time_series_clustering", description="Perform time series clustering using sktime")
+def time_series_clustering(data, clusterer_name: str):
+    """Perform time series clustering.
+
+    Parameters
+    ----------
+    data : pd.DataFrame
+        Input time series data.
+    clusterer_name : str
+        Name of the clusterer to use.
+
+    Returns
+    -------
+    Clustering results.
+    """
+    from sktime.clustering.all import all_estimators
+
+    clusterers = dict(all_estimators())
+    if clusterer_name not in clusterers:
+        raise ValueError(f"Clusterer {clusterer_name} not found in sktime.")
+
+    clusterer = clusterers[clusterer_name]()
+    return clusterer.fit_predict(data)
+
+@mcp.tool(name="time_series_distance", description="Compute distances between time series using sktime")
+def time_series_distance(data1, data2, distance_name: str):
+    """Compute distances between time series.
+
+    Parameters
+    ----------
+    data1 : pd.DataFrame
+        First time series data.
+    data2 : pd.DataFrame
+        Second time series data.
+    distance_name : str
+        Name of the distance metric to use.
 
+    Returns
+    -------
+    Distance value.
+    """
+    from sktime.dists_kernels.all import all_estimators
+
+    distances = dict(all_estimators())
+    if distance_name not in distances:
+        raise ValueError(f"Distance metric {distance_name} not found in sktime.")
+
+    distance = distances[distance_name]()
+    return distance(data1, data2)
 
 if __name__ == "__main__":
-    mcp.run(transport="http", host="0.0.0.0", port=8000)
+    mcp.run()