API Reference

This page documents the Python API of the sktime-mcp package, for contributors and for anyone embedding the server in another application.

If you are looking for the MCP tools exposed to an AI client — their names, arguments, and return values — see MCP Tool Reference instead.

sktime-mcp: MCP (Model Context Protocol) layer for sktime.

A semantic engine that exposes sktime’s native registry and semantics to LLMs, enabling discovery, reasoning, composition, and execution of time series workflows.

Server

MCP Server for sktime.

Main entry point for the Model Context Protocol server that exposes sktime’s registry and execution capabilities to LLMs.

sktime_mcp.server.sanitize_for_json(obj)

Recursively convert objects to JSON-serializable format.

Handles: - Standard Python scalars and containers (dict, list, tuple) - NumPy integer/float scalars and ndarrays - Pandas Timestamp, NaT, NA, and Series/DataFrame - Arbitrary objects (fallback to str repr)

async sktime_mcp.server.list_tools()

List all available MCP tools.

Return type:

list[Tool]

async sktime_mcp.server.call_tool(name, arguments)

Handle tool calls.

Return type:

list[TextContent]

async sktime_mcp.server.run_server()

Run the MCP server.

sktime_mcp.server.build_arg_parser()

Build the command-line parser for the sktime-mcp console script.

Returns:

Parser accepting --version/-V and --help. With no arguments the server starts on stdio.

Return type:

argparse.ArgumentParser

sktime_mcp.server.main()

Main entry point.

Tools

Tools module for sktime MCP.

sktime_mcp.tools.describe_component_tool(name)

Get detailed information about ANY class or component in the sktime ecosystem (estimators, transformers, splitters, metrics, aligners).

Parameters:

name (str) – Name of the component class (e.g., “ARIMA”, “SlidingWindowSplitter”, “MeanAbsolutePercentageError”)

Returns:

• success: bool

• name: Component name

• task: Scitype (e.g., “forecaster”, “transformer”, “splitter”, “metric”)

• module: Full module path

• parameters: Dict of parameter names with defaults

• tags: Dict of capability tags

• tag_explanations: Human-readable tag descriptions

• docstring: First 500 chars of docstring

Return type:

Dictionary with

sktime_mcp.tools.query_registry_tool(task=None, tags=None, query=None, limit=50, offset=0)

Query the sktime registry for estimators,or capability tags.

All filters are combined: query narrows by name/docstring, then task and tags are applied on top.

To retrieve capability tags instead of estimators, set task=”tag” (or “tags”).

Parameters:

• task (str | None) – Filter by scitype (e.g., “forecaster”, “classifier”, “regressor”, “transformer”, “detector”, “metric”). Set to “tag” or “tags” to query available capability tags.

• tags (dict[str, Any] | str | None) – Filter estimators by capability tags. Can be a dictionary or a JSON string. Example JSON string: ‘{“capability:pred_int”: true}’. Ignored when task=”tag”.

• query (str | None) – Search by name/description (substring, case-insensitive).

• limit (int) – Maximum number of results to return (default: 50). Ignored when task=”tag”.

• offset (int) – Number of results to skip for pagination (default: 0). Ignored when task=”tag”.

Returns:

• success: bool

• results: List of matching estimators or tags

• count: Number of results returned in this page

• total: Total matching results (before limit/offset)

• offset: Current offset (for pagination)

• limit: Current limit (for pagination)

• has_more: True if more results exist beyond this page

Return type:

A dictionary with

sktime_mcp.tools.instantiate_tool(spec)

Create an estimator or pipeline instance using craft and return a handle.

Parameters:

spec (str) – A sktime craft specification string (e.g., “ARIMA(order=(1, 1, 1))” or “Detrender() * ARIMA()”).

Returns:

Dictionary containing the success status and the unique handle.

Return type:

dict

sktime_mcp.tools.list_handles_tool()

List all active estimator handles.

Returns:

Dictionary containing details of active handles:

• "success" (bool) – True if the handles were retrieved successfully.

• "handles" (list of dict) – Details of active handles including handle ID, estimator name, and state.

• "count" (int) – The number of active handles.

Return type:

dict

sktime_mcp.tools.release_handle_tool(handle)

Release an estimator handle and free resources.

Parameters:

handle (str) – The handle ID to release.

Returns:

Dictionary containing success status:

• "success" (bool) – True if the handle was successfully released, False otherwise.

• "handle" (str) – The handle ID that was requested for release.

• "message" (str) – Status message indicating outcome.

Return type:

dict

sktime_mcp.tools.load_model_tool(path)

Load a saved model from a local path or MLflow URI and register its handle.

Parameters:

path (str) –

Local directory path or MLflow URI to the saved model. Examples:

• ”/tmp/my_arima_model” (Linux/macOS) or “C:Tempmy_arima_model” (Windows)

• ”runs:/<run_id>/model”

• ”mlflow-artifacts:/<run_id>/artifacts/model”

• ”models:/<model_name>/<version>”

Returns:

Dictionary containing success status and the new handle:

• "success" (bool) – True if the model was loaded successfully.

• "handle" (str, optional) – The registered handle ID for the loaded model.

• "estimator" (str, optional) – Class name of the loaded estimator.

• "path" (str) – The path/URI from which the model was loaded.

• "message" (str) – Status message describing outcome.

• "error" (str, optional) – Error message if “success” is False.

Return type:

dict

sktime_mcp.tools.fit_tool(estimator_handle, X_dataset=None, y_dataset=None, X_handle=None, y_handle=None, fh=None, run_async=False)

Fit an estimator on data.

Return type:

dict[str, Any]

sktime_mcp.tools.predict_tool(estimator_handle, horizon=12, mode='predict', coverage=0.9, alpha=None, X_dataset=None, y_dataset=None, X_handle=None, y_handle=None, run_async=False)

Generate predictions from a fitted estimator.

Set run_async=True to run as a background job and return a job_id.

Return type:

dict[str, Any]

sktime_mcp.tools.update_tool(estimator_handle, X_dataset=None, y_dataset=None, X_handle=None, y_handle=None)

Return type:

dict[str, Any]

sktime_mcp.tools.get_fitted_params_tool(estimator_handle)

Return type:

dict[str, Any]

sktime_mcp.tools.evaluate_tool(estimator_handle, y, X=None, cv_folds=3, metric=None, initial_window=None, run_async=False)

Cross-validate an estimator on a dataset.

y and X accept data_handle ids or built-in demo dataset names. Set run_async=True to run as a background job.

Return type:

dict[str, Any]

sktime_mcp.tools.load_data_source_tool(config, run_async=False)

Load data from any source (pandas, SQL, file, etc.).

Can run synchronously (blocking) or asynchronously in the background.

Parameters:

• config (dict) –

  Data source configuration dictionary. Must contain:

  • "type" (str) – source type: "pandas", "sql", "file", or "url".

  • Additional type-specific configuration keys (e.g. "data", "path", "time_column", "target_column").

• run_async (bool, default=False) – If True, schedules the loading as a background job and returns a job_id immediately. If False, blocks until loaded and returns the data_handle directly.

Returns:

Dictionary containing load results and metadata.

If run_async is False, contains:

• "success" (bool) – True if the data was loaded successfully.

• "data_handle" (str) – the unique handle ID for the loaded data.

• "metadata" (dict) – rich metadata including row count, columns, and data type information.

• "validation" (dict) – results of indexing and format validation checks.

If run_async is True, contains:

• "success" (bool) – True if the background job was scheduled successfully.

• "job_id" (str) – unique job ID to monitor progress via check_job_status.

• "message" (str) – a user-friendly status message.

• "source_type" (str) – the type of the source requested to load.

Return type:

dict

Examples

# Synchronous Pandas DataFrame Loading >>> load_data_source_tool({ … “type”: “pandas”, … “data”: {“date”: […], “value”: […]}, … “time_column”: “date”, … “target_column”: “value” … })

# Asynchronous CSV File Loading >>> load_data_source_tool({ … “type”: “file”, … “path”: “/path/to/data.csv”, … “time_column”: “date”, … “target_column”: “value” … }, run_async=True)

sktime_mcp.tools.release_data_handle_tool(data_handle)

Release a data handle and free memory.

Parameters:

data_handle (str) – Data handle to release.

Returns:

Dictionary containing success status:

• "success" (bool) – True if the handle was successfully released, False otherwise.

• "message" (str, optional) – Detailed status or error message.

Return type:

dict

sktime_mcp.tools.list_available_data_tool(is_demo=None)

List all data available for use — system demo datasets and active data handles.

Aggregates the output of list_datasets and list_data_handles into a single unified response, with clear labelling for each category.

Parameters:

is_demo (bool | None) – Optional boolean filter. - True -> return only system demo datasets - False -> return only active (user-loaded) data handles - None -> return both (default)

Returns:

• success: bool

• system_demos: dict of demo dataset task -> list of names, or {} if is_demo=False

• active_handles: list of dicts with handle id and metadata

• total: int — combined count of items returned

Return type:

Dictionary with

sktime_mcp.tools.format_time_series_tool(data_handle, auto_infer_freq=True, fill_missing=True, remove_duplicates=True)

Automatically format time series data to be sktime-compatible.

This tool fixes common issues: - Missing frequency on DatetimeIndex - Duplicate timestamps - Missing values - Irregular time intervals

Parameters:

• data_handle (str) – Handle from load_data_source

• auto_infer_freq (bool) – Automatically infer and set frequency (default: True)

• fill_missing (bool) – Fill missing values with forward/backward fill (default: True)

• remove_duplicates (bool) – Remove duplicate timestamps (default: True)

Returns:

• success: bool

• data_handle: str (new handle with formatted data)

• changes_made: dict (what was fixed)

• metadata: dict (updated metadata)

Return type:

Dictionary with

Example

>>> format_time_series_tool(
...     data_handle="data_abc123",
...     auto_infer_freq=True,
...     fill_missing=True
... )

sktime_mcp.tools.inspect_data_tool(data_handle)

Inspect a loaded data handle and return rich metadata.

Provides comprehensive information about the data behind a handle, including shape, column types, frequency, cutoff point, missing value counts, a preview (head), and summary statistics.

Parameters:

data_handle (str) – The unique handle ID for the loaded data source (from load_data_source).

Returns:

Dictionary containing detailed metadata and summary statistics:

• "success" (bool) – True if the data handle was found and inspected successfully.

• "data_handle" (str) – The inspected data handle ID.

• "mtype" (str) – The format mtype (e.g. ‘pd.Series’, ‘pd.DataFrame’).

• "scitype" (str) – The sktime scientific type (e.g. ‘Series’, ‘Panel’).

• "shape" (list of int) – Shape list: [rows, columns] or [rows].

• "columns" (list of str) – Names of all variables (including exogenous features if present).

• "dtypes" (dict) – Mapping of column names to string names of their data types.

• "index_names" (list of str) – List of names of the index levels.

• "freq" (str or None) – Inferred or declared frequency of the time index.

• "cutoff" (str or None) – Cutoff timestamp/integer index indicating the end of the history.

• "n_missing" (int) – Total count of missing values across the entire dataset.

• "head" (dict) – Preview of the first 5 rows of data.

• "summary_stats" (dict) – Statistical summary metrics (mean, std, min, max, etc.) per column.

• "error" (str, optional) – Error message if “success” is False.

Return type:

dict

sktime_mcp.tools.split_data_tool(data_handle, test_size=None, fh=None)

Split a time series data handle into train and test sets.

Uses sktime’s temporal_train_test_split() when available, falling back to a pandas-based implementation. Exactly one of test_size or fh must be provided.

Parameters:

• data_handle (str) – Handle ID of the loaded data to split (from load_data_source).

• test_size (float or None, default=None) – Fraction of the data to hold out for testing (0.0–1.0). Mutually exclusive with fh.

• fh (int, list of int, or None, default=None) – Forecast horizon — the number of final time steps to reserve as the test set. Can be a single int or a list of relative step indices. Mutually exclusive with test_size.

Returns:

Dictionary containing the split train/test handles and metadata:

• "success" (bool) – True if the split completed successfully, False otherwise.

• "train_handle" (str) – The new unique data handle ID representing the training set.

• "test_handle" (str) – The new unique data handle ID representing the test set.

• "cutoff" (str) – The cutoff timestamp indicating the last training timestamp.

• "train_size" (int) – Number of observations in the training set.

• "n_test" (int) – Number of observations in the test set.

• "error" (str, optional) – Error message if “success” is False.

Return type:

dict

sktime_mcp.tools.transform_data_tool(data_handle, action='format', auto_infer_freq=True, fill_missing=True, remove_duplicates=True, to_mtype=None)

Transform a data handle — either format it or convert its mtype.

Supports two modes controlled by the action argument.

Parameters:

• data_handle (str) – Handle ID of the loaded data to transform (from load_data_source).

• action (str, default="format") –

  The transformation action to perform. Must be one of:

  • "format" – auto-fix common time series issues such as inferring frequency, removing duplicate timestamps, and filling missing values.

  • "convert" – convert the data to a different sktime machine type (mtype).

• auto_infer_freq (bool, default=True) – (Format mode only) Infer and set frequency.

• fill_missing (bool, default=True) – (Format mode only) Forward/backward fill missing values.

• remove_duplicates (bool, default=True) – (Format mode only) Remove duplicate timestamps.

• to_mtype (str or None, default=None) – (Convert mode only) Target machine type string, e.g. “pd.DataFrame”, “pd.Series”, “np.ndarray”.

Returns:

Dictionary containing the new data handle and a list of applied changes:

• "success" (bool) – True if the transformation succeeded, False otherwise.

• "data_handle" (str) – The new unique data handle ID representing the transformed data.

• "changes_applied" (list of str) – A list of human-readable changes that were applied to the data.

• "metadata" (dict, optional) – updated metadata for the new handle.

• "error" (str, optional) – Error message if “success” is False.

Return type:

dict

sktime_mcp.tools.save_data_tool(data_handle, path, format='csv')

Persist the data behind a handle to a local file.

Supports CSV, Parquet, and JSON output formats. The target directory is created automatically if it does not exist.

Parameters:

• data_handle (str) – Handle ID of the data to save (from load_data_source, split_data, etc.).

• path (str) – Destination file path (e.g. “/tmp/forecast_output.csv”). Note that the file extension is not used to infer the format; use the format argument instead.

• format (str, default="csv") – Output format. Must be one of: “csv”, “parquet”, or “json”.

Returns:

Dictionary containing success status and path information:

• "success" (bool) – True if the data was written successfully, False otherwise.

• "saved_path" (str) – Absolute path to the written file.

• "format" (str) – The format used to write the file.

• "rows" (int) – Number of rows written to the file.

• "error" (str, optional) – Error message if “success” is False.

Return type:

dict

sktime_mcp.tools.export_code_tool(handle, var_name='model', include_fit_example=False, dataset=None)

Export an estimator or pipeline as executable Python code.

Parameters:

• handle (str) – The handle ID of the estimator/pipeline to export

• var_name (str) – Variable name to use in generated code (default: “model”)

• include_fit_example (bool) – Whether to include a fit/predict example (default: False)

• dataset (str | None) – Optional dataset name for the fit example (default: None, falls back to airline)

Returns:

• success: bool

• code: Generated Python code string

• estimator_name: Name of the estimator/pipeline

• is_pipeline: Whether this is a pipeline

Return type:

Dictionary with

Example

>>> # First create an estimator
>>> result = instantiate_tool("ARIMA", {"order": [1, 1, 1]})
>>> handle = result["handle"]
>>>
>>> # Export as code
>>> export_code_tool(handle, var_name="arima_model")
{
    "success": True,
    "code": "from sktime.forecasting.arima import ARIMA\n\narima_model = ARIMA(order=[1, 1, 1])",
    "estimator_name": "ARIMA",
    "is_pipeline": False
}

sktime_mcp.tools.save_model_tool(estimator_handle, path, mlflow_params=None)

Save an instantiated estimator to a local path or URI using sktime+MLflow.

Parameters:

• estimator_handle (str) – Handle ID from instantiate

• path (str) – Local directory or URI where the model should be saved

• mlflow_params (dict[str, Any] | None) – Optional extra keyword arguments for sktime MLflow save_model

Return type:

dict[str, Any]

Returns:

Dictionary with success status and confirmation message/path.

sktime_mcp.tools.check_job_status_tool(job_id)

Check the status of a background job.

Parameters:

job_id (str) – Job ID to check

Return type:

dict[str, Any]

Returns:

Dictionary with job status and progress information

sktime_mcp.tools.list_jobs_tool(status=None, limit=20, offset=0)

List background jobs with offset/limit pagination.

Parameters:

• status (str | None) – Filter by status (pending, running, completed, failed, cancelled)

• limit (int) – Maximum number of jobs to return in this page

• offset (int) – Number of jobs to skip (for paging through results)

Return type:

dict[str, Any]

Returns:

Dictionary with the page of jobs plus pagination metadata (total, offset, limit, has_more).

sktime_mcp.tools.cancel_job_tool(job_id, delete=False)

Cancel a running/pending job, and optionally remove its record.

Parameters:

• job_id (str) – Job ID to cancel

• delete (bool) – Also remove the job record after cancelling (default: False). For jobs that are already completed/failed, set delete=True to remove them.

Return type:

dict[str, Any]

Returns:

Dictionary with success status and message

sktime_mcp.tools.plot_series_tool(data_handles, labels=None, title=None, path=None, figsize=None, dpi=None, markers=None, x_label=None, y_label=None, image_format='png')

Plot one or more time series natively.

Parameters:

• data_handles (list of str) – List of data handle IDs to plot (e.g. train, test, forecasts).

• labels (list of str, optional) – Labels for each series. If the count does not match the number of data handles, it is silently padded or truncated.

• title (str, optional) – Title of the plot.

• path (str, optional) – Path to save the plot. If not provided, returns base64 encoded image.

• figsize (list of float, optional) – Figure size as [width, height] in inches. Default [12, 6].

• dpi (int, optional) – Resolution in dots per inch. Default 150.

• markers (str or list of str, optional) – Marker style(s) for data points (e.g. "o", [".", "x"]). Passed through to sktime.utils.plotting.plot_series.

• x_label (str, optional) – Custom label for the x-axis.

• y_label (str, optional) – Custom label for the y-axis.

• image_format (str, optional) – Image output format: "png" (default), "svg", or "webp".

Returns:

Dictionary containing success status, path or base64 string, and plot metadata (n_series, labels_used, figsize, dpi).

Return type:

dict

sktime_mcp.tools.run_command_tool(command)

Run an arbitrary CLI/bash command inside the sktime-mcp container.

Return type:

dict[str, Any]

Registry

Each class below is documented at its canonical location. sktime_mcp.registry re-exports them for convenience.

Registry Interface for sktime MCP.

Thin adapter over sktime’s estimator registry. Delegates to sktime.registry functions directly.

class sktime_mcp.registry.interface.EstimatorNode(name, task, class_ref, module, tags=<factory>, parameters=<factory>, docstring=None)

Bases: object

Represents a single estimator in the sktime registry.

This is the semantic representation of an estimator that gets exposed to the LLM through the MCP.

Variables:

• name – The class name of the estimator (e.g., “ARIMA”)

• task – The scitype (e.g., “forecaster”, “transformer”, “classifier”)

• class_ref – Reference to the actual Python class

• module – Full module path to the estimator

• tags – Dictionary of capability tags

• parameters – Parameter names with their defaults

• docstring – The estimator’s docstring

name: str

task: str

class_ref: type

module: str

tags: dict[str, Any]

parameters: dict[str, Any]

docstring: str | None = None

to_dict()

Convert to dictionary for JSON serialization.

Return type:

dict[str, Any]

to_summary()

Return a minimal summary for list operations.

Return type:

dict[str, Any]

__init__(name, task, class_ref, module, tags=<factory>, parameters=<factory>, docstring=None)

class sktime_mcp.registry.interface.RegistryInterface

Bases: object

Adapter over sktime’s estimator registry.

Delegates to sktime.registry.all_estimators for discovery and filtering, sktime.registry.craft for name-based lookups, and uses public class methods (get_class_tags, get_param_names, get_param_defaults) for metadata extraction.

__init__()

Initialize the registry interface.

get_all_estimators(task=None, tags=None)

Get all estimators, optionally filtered by scitype and tags.

Delegates filtering to sktime.registry.all_estimators.

Parameters:

• task (str | None) – Filter by scitype (e.g., “forecaster”, “classifier”).

• tags (dict[str, Any] | None) – Filter by capability tags.

Return type:

list[EstimatorNode]

get_estimator_by_name(name)

Get a specific estimator by its class name.

Uses sktime.registry.craft for direct class resolution instead of scanning all estimators.

Parameters:

name (str) – The class name of the estimator (e.g., “ARIMA”).

Return type:

EstimatorNode | None

get_available_tasks()

Get list of available scitypes from sktime’s base class register.

Return type:

list[str]

get_available_tags()

Get rich metadata for all available tags using sktime’s registry.

Returns a list of dicts with: tag, description, value_type, applies_to.

Return type:

list[dict[str, Any]]

search_estimators(query)

Search estimators by name, module, or docstring.

Parameters:

query (str) – Search string (case-insensitive).

Return type:

list[EstimatorNode]

sktime_mcp.registry.interface.get_registry()

Get the singleton registry instance.

Return type:

RegistryInterface

Tag Resolver for sktime MCP.

Tags encode estimator capabilities and constraints. This module provides utilities for working with tags and understanding their meanings.

class sktime_mcp.registry.tag_resolver.TagInfo(name, description, value_type, possible_values=None, category='general')

Bases: object

Information about a specific tag.

name: str

description: str

value_type: str

possible_values: list[Any] | None = None

category: str = 'general'

__init__(name, description, value_type, possible_values=None, category='general')

class sktime_mcp.registry.tag_resolver.TagResolver

Bases: object

Resolver for sktime estimator tags.

Tags encode important semantic information about estimators: - Supported data types - Probabilistic vs deterministic predictions - Composability rules - Missing value handling - And many more…

This class provides utilities for understanding and querying tags.

__init__()

Initialize the tag resolver.

property tag_definitions: dict[str, TagInfo]

Get tag definitions, loading them if necessary.

get_tag_info(tag_name)

Get information about a specific tag.

Parameters:

tag_name (str) – The tag name to look up

Return type:

TagInfo | None

Returns:

TagInfo if known, None otherwise

get_tag_description(tag_name)

Get human-readable description of a tag.

Parameters:

tag_name (str) – The tag name

Return type:

str

Returns:

Description string, or generic message if unknown

get_tags_by_category(category)

Get all known tags in a specific category.

Parameters:

category (str) – Category name (e.g., “capability”, “data”, “behavior”)

Return type:

list[TagInfo]

Returns:

List of TagInfo objects in that category

get_all_categories()

Get list of all tag categories.

Return type:

list[str]

explain_tags(tags)

Get human-readable explanations for a set of tags.

Parameters:

tags (dict[str, Any]) – Dictionary of tag names to values

Return type:

dict[str, str]

Returns:

Dictionary of tag names to explanation strings

filter_estimators_by_capability(task=None, probabilistic=None, handles_missing=None, multivariate=None)

Filter estimators by common capability requirements.

This is a convenience method that translates human-friendly requirements into the appropriate tag queries.

Parameters:

• task (str | None) – Task type filter

• probabilistic (bool | None) – Require probabilistic predictions

• handles_missing (bool | None) – Require missing data handling

• multivariate (bool | None) – Require multivariate support

Return type:

list[EstimatorNode]

Returns:

List of matching EstimatorNode objects

check_compatibility(estimator, requirements)

Check if an estimator meets specific requirements.

Parameters:

• estimator (EstimatorNode) – The estimator to check

• requirements (dict[str, Any]) – Dictionary of required tag values

Return type:

dict[str, bool]

Returns:

Dictionary mapping requirement names to whether they are met

suggest_similar_estimators(estimator, max_results=5)

Find estimators with similar capabilities.

Parameters:

• estimator (EstimatorNode) – Reference estimator

• max_results (int) – Maximum number of results

Return type:

list[EstimatorNode]

Returns:

List of similar estimators (same task, similar tags)

sktime_mcp.registry.tag_resolver.get_tag_resolver()

Get the singleton tag resolver instance.

Return type:

TagResolver

Runtime

Executor for sktime MCP.

Responsible for instantiating estimators, loading datasets, and running fit/predict operations.

class sktime_mcp.runtime.executor.Executor

Bases: object

Execution runtime for sktime estimators.

Handles instantiation, fitting, and prediction.

__init__()

instantiate(spec)

Instantiate an estimator or pipeline from a spec and return a handle.

Return type:

dict[str, Any]

load_dataset(name)

Load a demo dataset.

Return type:

dict[str, Any]

fit(handle_id, y, X=None, fh=None)

Fit an estimator.

Return type:

dict[str, Any]

predict(handle_id, fh=None, X=None, y=None, mode='predict', coverage=0.9, alpha=None)

Generate predictions.

Return type:

dict[str, Any]

async predict_async(handle_id, *, horizon=12, mode='predict', coverage=0.9, alpha=None, X_dataset=None, y_dataset=None, X_handle=None, y_handle=None, job_id=None)

Async version of predict with job tracking.

Return type:

dict[str, Any]

call_method(handle_id, method_name, kwargs=None)

Dynamically call a method on the underlying estimator.

Return type:

dict[str, Any]

update(handle_id, y, X=None, update_params=None)

Update a fitted estimator with new data.

Return type:

dict[str, Any]

get_fitted_params(handle_id)

Get fitted parameters from an estimator.

Return type:

dict[str, Any]

async fit_async(handle_id, X_dataset=None, y_dataset=None, X_handle=None, y_handle=None, fh=None, job_id=None)

Async version of fit with job tracking.

Return type:

dict[str, Any]

async evaluate_async(handle_id, y, *, X=None, cv_folds=3, metric=None, initial_window=None, job_id=None)

Async version of evaluate with job tracking.

Return type:

dict[str, Any]

list_datasets()

List available demo datasets.

Return type:

list[str]

load_data_source(config)

Load data from any source (pandas, SQL, file, etc.).

Parameters:

config (dict[str, Any]) – Data source configuration with ‘type’ key Examples: - {“type”: “pandas”, “data”: df, “time_column”: “date”, “target_column”: “value”} - {“type”: “sql”, “connection_string”: “…”, “query”: “…”, “time_column”: “date”} - {“type”: “file”, “path”: “/path/to/data.csv”, “time_column”: “date”}

Returns:

• success: bool

• data_handle: str (handle ID for the loaded data)

• metadata: dict (information about the data)

• validation: dict (validation results)

Return type:

Dictionary with

async load_data_source_async(config, job_id=None)

Async version of load_data_source with job tracking.

Runs data loading in the background without blocking the MCP server. Progress is tracked via the JobManager.

Parameters:

• config (dict[str, Any]) – Data source configuration

• job_id (str | None) – Optional job ID (created if not provided)

Return type:

dict[str, Any]

Returns:

Dictionary with data_handle and metadata

format_data_handle(data_handle, auto_infer_freq=True, fill_missing=True, remove_duplicates=True)

Format data associated with a handle.

Return type:

dict[str, Any]

list_data_handles()

List all loaded data handles.

Return type:

dict[str, Any]

Returns:

Dictionary with list of data handles and their metadata

release_data_handle(data_handle)

Release a data handle and free memory.

Parameters:

data_handle (str) – Data handle to release

Return type:

dict[str, Any]

Returns:

Dictionary with success status

sktime_mcp.runtime.executor.get_executor()

Return type:

Executor

Handle Manager for sktime MCP.

Manages references to instantiated estimator objects.

class sktime_mcp.runtime.handles.HandleInfo(handle_id, estimator_name, instance, params, created_at, fitted=False, metadata=<factory>)

Bases: object

Information about a managed handle.

handle_id: str

estimator_name: str

instance: Any

params: dict[str, Any]

created_at: datetime

fitted: bool = False

metadata: dict[str, Any]

to_dict()

Return type:

dict[str, Any]

__init__(handle_id, estimator_name, instance, params, created_at, fitted=False, metadata=<factory>)

class sktime_mcp.runtime.handles.HandleManager(max_handles=100)

Bases: object

Manager for estimator instance handles.

__init__(max_handles=100)

create_handle(estimator_name, instance, params=None, metadata=None)

Return type:

str

get_instance(handle_id)

Return type:

Any

get_info(handle_id)

Return type:

HandleInfo

exists(handle_id)

Return type:

bool

mark_fitted(handle_id)

Return type:

None

is_fitted(handle_id)

Return type:

bool

release_handle(handle_id)

Return type:

bool

list_handles()

Return type:

list[dict[str, Any]]

clear_all()

Return type:

int

sktime_mcp.runtime.handles.get_handle_manager()

Return type:

HandleManager

Job management for long-running operations in sktime MCP.

Handles background training jobs with progress tracking and status updates.

class sktime_mcp.runtime.jobs.JobStatus(value)

Bases: Enum

Status of a background job.

PENDING = 'pending'

RUNNING = 'running'

COMPLETED = 'completed'

FAILED = 'failed'

CANCELLED = 'cancelled'

class sktime_mcp.runtime.jobs.JobInfo(job_id, job_type, estimator_handle, status=JobStatus.PENDING, created_at=<factory>, start_time=None, end_time=None, total_steps=0, completed_steps=0, current_step='', result=None, errors=<factory>, dataset_name=None, horizon=None, estimator_name=None)

Bases: object

Information about a background job.

job_id: str

job_type: str

estimator_handle: str

status: JobStatus = 'pending'

created_at: datetime

start_time: datetime | None = None

end_time: datetime | None = None

total_steps: int = 0

completed_steps: int = 0

current_step: str = ''

result: dict[str, Any] | None = None

errors: list[str]

dataset_name: str | None = None

horizon: int | None = None

estimator_name: str | None = None

property progress_percentage: float

Calculate progress percentage.

property elapsed_time: float | None

Calculate elapsed time in seconds.

property estimated_time_remaining: float | None

Estimate remaining time in seconds.

property estimated_time_remaining_human: str | None

Human-readable estimated time remaining.

to_dict()

Convert to dictionary for JSON serialization.

Return type:

dict[str, Any]

__init__(job_id, job_type, estimator_handle, status=JobStatus.PENDING, created_at=<factory>, start_time=None, end_time=None, total_steps=0, completed_steps=0, current_step='', result=None, errors=<factory>, dataset_name=None, horizon=None, estimator_name=None)

class sktime_mcp.runtime.jobs.JobManager

Bases: object

Thread-safe manager for background jobs.

Handles job creation, status updates, and cleanup.

__init__()

jobs: dict[str, JobInfo]

create_job(job_type, estimator_handle, estimator_name=None, dataset_name=None, horizon=None, total_steps=3)

Create a new job and return its ID.

Parameters:

• job_type (str) – Type of job (fit, predict, evaluate, etc.)

• estimator_handle (str) – Handle of the estimator

• estimator_name (str | None) – Name of the estimator

• dataset_name (str | None) – Name of the dataset (if applicable)

• horizon (int | None) – Forecast horizon (if applicable)

• total_steps (int) – Total number of steps in the job

Return type:

str

Returns:

Job ID (UUID)

register_task(job_id, task)

Associate a background asyncio task with a job.

Retaining the task reference lets cancel_job cancel the running coroutine, and prevents the event loop from garbage-collecting a still-pending task. A done callback clears the reference and surfaces any exception that would otherwise be swallowed by fire-and-forget.

Return type:

None

update_job(job_id, status=None, completed_steps=None, current_step=None, result=None, errors=None)

Update job status and progress.

Parameters:

• job_id (str) – Job ID to update

• status (JobStatus | None) – New status

• completed_steps (int | None) – Number of completed steps

• current_step (str | None) – Description of current step

• result (dict[str, Any] | None) – Job result (when completed)

• errors (list[str] | None) – List of errors (when failed)

Return type:

bool

Returns:

True if job was updated, False if not found

get_job(job_id)

Get job information.

Parameters:

job_id (str) – Job ID

Return type:

JobInfo | None

Returns:

JobInfo if found, None otherwise

list_jobs(status=None, limit=None, offset=0)

List jobs, optionally filtered by status, with offset/limit pagination.

Jobs are ordered newest-first, then offset items are skipped and up to limit are returned. Use count_jobs for the total page count.

Parameters:

• status (JobStatus | None) – Filter by status (None = all jobs)

• limit (int | None) – Maximum number of jobs to return (None = no limit)

• offset (int) – Number of jobs to skip from the start of the ordered list

Return type:

list[JobInfo]

Returns:

List of JobInfo objects for the requested page

count_jobs(status=None)

Return the total number of jobs matching an optional status filter.

Return type:

int

cancel_job(job_id)

Cancel a job.

Parameters:

job_id (str) – Job ID to cancel

Return type:

bool

Returns:

True if job was cancelled, False if not found or already completed

Notes

Cancellation is cooperative. The background task is cancelled at its next await point, which stops any remaining job steps. Work already running inside a thread-pool executor (e.g. a single fit call) cannot be force-killed and runs to completion, but its result is discarded because update_job ignores late updates on a CANCELLED job.

cleanup_old_jobs(max_age_hours=24)

Remove jobs older than max_age_hours.

Parameters:

max_age_hours (int) – Maximum age in hours

Return type:

int

Returns:

Number of jobs removed

delete_job(job_id)

Delete a job.

Parameters:

job_id (str) – Job ID to delete

Return type:

bool

Returns:

True if job was deleted, False if not found

sktime_mcp.runtime.jobs.get_job_manager()

Get the singleton JobManager instance.

Return type:

JobManager

Data

Data source layer for sktime-mcp.

Provides adapters for loading data from various sources:

• Pandas DataFrames (in-memory)

• SQL databases (PostgreSQL, MySQL, SQLite, etc.)

• Files (CSV, Excel, Parquet)

• Web URLs

Usage:

from sktime_mcp.data import DataSourceRegistry

# Create adapter from config
config = {
    "type": "pandas",
    "data": df,
    "time_column": "date",
    "target_column": "value"
}

adapter = DataSourceRegistry.create_adapter(config)
data = adapter.load()
is_valid, report = adapter.validate(data)
y, X = adapter.to_sktime_format(data)

class sktime_mcp.data.DataSourceAdapter(config)

Bases: ABC

Abstract base class for all data source adapters.

All adapters must implement: - load(): Fetch data from source - validate(): Check data quality - to_sktime_format(): Convert to sktime-compatible format

__init__(config)

Initialize the adapter.

Parameters:

config (dict[str, Any]) – Configuration dictionary specific to the adapter type

abstractmethod load()

Load data from the source (synchronous).

Return type:

DataFrame

Returns:

DataFrame with time index

async load_async(job_id=None)

Load data from the source (asynchronous).

Default implementation runs the synchronous load() in a separate thread. Adapters should override this for true non-blocking async IO.

Parameters:

job_id (str | None) – Optional job ID for progress reporting

Return type:

DataFrame

Returns:

DataFrame with time index

abstractmethod validate(data)

Validate data quality.

Parameters:

data (DataFrame) – DataFrame to validate

Return type:

tuple[bool, dict[str, Any]]

Returns:

Tuple of (is_valid, validation_report), where validation_report contains the keys valid (bool), errors (list of str), and warnings (list of str).

to_sktime_format(data)

Convert to sktime format (y, X).

Parameters:

data (DataFrame) – DataFrame to convert

Returns:

• y: Target time series (pd.Series with DatetimeIndex)

• X: Exogenous variables (pd.DataFrame, optional)

Return type:

Tuple of (y, X) where

get_metadata()

Return metadata about the data source.

Return type:

dict[str, Any]

Returns:

Dictionary with metadata (rows, columns, frequency, etc.)

class sktime_mcp.data.DataSourceRegistry

Bases: object

Registry for data source adapters.

Provides a central place to register and retrieve adapters.

classmethod register(name, adapter_class)

Register a new adapter.

Parameters:

• name (str) – Name to register the adapter under

• adapter_class (type[DataSourceAdapter]) – Adapter class (must inherit from DataSourceAdapter)

classmethod get_adapter(source_type)

Get adapter class by type.

Parameters:

source_type (str) – Type of data source (e.g., “pandas”, “sql”, “file”)

Return type:

type[DataSourceAdapter]

Returns:

Adapter class

Raises:

ValueError – If source type is not registered

classmethod create_adapter(config)

Create adapter instance from config.

Parameters:

config (dict) – Configuration dictionary with ‘type’ key

Return type:

DataSourceAdapter

Returns:

Adapter instance

Raises:

ValueError – If config is invalid or type is not registered

classmethod list_adapters()

List all registered adapter types.

Return type:

list

Returns:

List of adapter type names

classmethod get_adapter_info(source_type)

Get information about an adapter.

Parameters:

source_type (str) – Type of data source

Return type:

dict

Returns:

Dictionary with adapter information

class sktime_mcp.data.PandasAdapter(config)

Bases: DataSourceAdapter

Adapter for in-memory pandas DataFrames.

Config example:

{
    "type": "pandas",
    "data": <DataFrame object or dict>,
    "time_column": "date",  # optional, will try to detect
    "target_column": "value",  # optional, defaults to first column
    "exog_columns": ["feature1", "feature2"],  # optional
    "frequency": "D"  # optional, will try to infer
}

load()

Load from in-memory DataFrame or dict.

Return type:

DataFrame

validate(data)

Validate pandas DataFrame for time series forecasting.

Return type:

tuple[bool, dict[str, Any]]

class sktime_mcp.data.SQLAdapter(config)

Bases: DataSourceAdapter

Adapter for SQL databases.

Config example:

{
    "type": "sql",
    "connection_string": "postgresql://user:pass@host:5432/db",
    # OR individual components:
    "dialect": "postgresql",  # postgresql, mysql, sqlite, mssql
    "host": "localhost",
    "port": 5432,
    "database": "mydb",
    "username": "user",
    "password": "pass",

    # Query
    "query": "SELECT date, value FROM sales WHERE date >= '2020-01-01'",
    # OR
    "table": "sales",
    "filters": {"date": ">=2020-01-01"},

    # Column mapping
    "time_column": "date",
    "target_column": "value",
    "exog_columns": ["feature1", "feature2"],

    # Optional
    "parse_dates": ["date"],
    "frequency": "D"
}

load()

Load from SQL database.

Return type:

DataFrame

validate(data)

Validate SQL data using pandas adapter validation.

Return type:

tuple[bool, dict[str, Any]]

class sktime_mcp.data.FileAdapter(config)

Bases: DataSourceAdapter

Adapter for file-based data sources.

Config example:

{
    "type": "file",
    "path": "/path/to/data.csv",
    "format": "csv",  # csv, excel, parquet (auto-detected if not specified)

    # Column mapping
    "time_column": "date",
    "target_column": "value",
    "exog_columns": ["feature1", "feature2"],

    # CSV-specific options
    "csv_options": {
        "sep": ",",
        "header": 0,
        "encoding": "utf-8"
    },

    # Excel-specific options
    "excel_options": {
        "sheet_name": 0,
        "header": 0
    },

    # Common options
    "parse_dates": True,
    "frequency": "D"
}

load()

Load from file.

Return type:

DataFrame

validate(data)

Validate file data using pandas adapter validation.

Return type:

tuple[bool, dict[str, Any]]

class sktime_mcp.data.UrlAdapter(config)

Bases: DataSourceAdapter

Adapter for downloading data from Web URLs.

Config example:

{
    "type": "url",
    "url": "https://raw.githubusercontent.com/.../data.csv",
    "format": "csv",  # csv, excel, parquet (auto-detected from URL)

    # Column mapping
    "time_column": "date",
    "target_column": "value",
    "exog_columns": ["feature1", "feature2"],

    # Options are passed identically as FileAdapter
    "csv_options": { ... },
    "parse_dates": True,
    "frequency": "D"
}

async load_async(job_id=None)

Load data from the source (asynchronous).

Default implementation runs the synchronous load() in a separate thread. Adapters should override this for true non-blocking async IO.

Parameters:

job_id (str | None) – Optional job ID for progress reporting

Return type:

DataFrame

Returns:

DataFrame with time index

load()

Load data from the source (synchronous).

Return type:

DataFrame

Returns:

DataFrame with time index

validate(data)

Validate URL data using pandas adapter validation.

Return type:

tuple[bool, dict[str, Any]]

Configuration

Configuration module for sktime-mcp.

Centralizes environment variables and provides sensible defaults.

class sktime_mcp.config.Settings

Bases: object

Server and runtime configuration settings.

property log_level: str

Logging level. Env Var: SKTIME_MCP_LOG_LEVEL Default: “WARNING”

property log_path: str | None

Optional file path to output logs to in addition to stderr. Env Var: SKTIME_MCP_LOG_PATH Default: None

property auto_format: bool

Whether to automatically format time series data upon load. Env Var: SKTIME_MCP_AUTO_FORMAT Default: True

property job_max_age_hours: int

Maximum age in hours before a job is cleaned up. Env Var: SKTIME_MCP_JOB_MAX_AGE_HOURS Default: 24

property job_cleanup_interval_secs: int

Interval in seconds for periodic job cleanup. Env Var: SKTIME_MCP_JOB_CLEANUP_INTERVAL Default: 3600

property max_data_handles: int

Maximum number of active data handles to retain in memory. Env Var: SKTIME_MCP_MAX_DATA_HANDLES Default: 50

property max_response_tokens: int

Maximum tokens allowed per tool response. Env Var: SKTIME_MCP_MAX_RESPONSE_TOKENS Default: 0