API Reference

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.

class sktime_mcp.EstimatorNode(name, task, class_ref, module, tags=<factory>, hyperparameters=<factory>, docstring=None)[source]

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.

name: The class name of the estimator (e.g., “ARIMA”)

task: The task type (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

hyperparameters: List of hyperparameter names with their defaults

docstring: The estimator’s docstring for understanding usage

name: str

task: str

class_ref: type

module: str

tags: dict[str, Any]

hyperparameters: dict[str, Any]

docstring: str | None = None

to_dict()[source]

Convert to dictionary for JSON serialization.

Return type:: dict[str, Any]

to_summary()[source]

Return a minimal summary for list operations.

Return type:: dict[str, Any]

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

class sktime_mcp.RegistryInterface[source]

Bases: object

Interface to sktime’s estimator registry.

This class wraps sktime’s all_estimators function and provides structured access to estimator metadata, tags, and documentation.

The registry is the single source of truth for all estimator information.

TASK_MAP = {'classifier': 'classification', 'clusterer': 'clustering', 'detector': 'detection', 'forecaster': 'forecasting', 'network': 'network', 'param_est': 'parameter_estimation', 'regressor': 'regression', 'splitter': 'splitting', 'transformer': 'transformation'}

__init__()[source]: Initialize the registry interface.

get_all_estimators(task=None, tags=None)[source]

Get all estimators, optionally filtered by task and tags.

Parameters:

task (str | None) – Filter by task type (e.g., “forecasting”, “classification”)
tags (dict[str, Any] | None) – Filter by capability tags (e.g., {“capability:pred_int”: True})

Return type:

list[EstimatorNode]

Returns:

List of matching EstimatorNode objects

get_estimator_by_name(name)[source]

Get a specific estimator by its class name.

Parameters:: name (str) – The class name of the estimator (e.g., “ARIMA”)
Return type:: EstimatorNode | None
Returns:: EstimatorNode if found, None otherwise

get_available_tasks()[source]

Get list of available task types.

Return type:: list[str]

get_available_tags()[source]

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

Returns a list of dicts, each containing: - tag: the tag name (e.g., “scitype:y”) - description: human-readable explanation of what the tag means - value_type: the expected value type (e.g., “bool”, “str”) - applies_to: list of estimator types this tag applies to

Return type:: list[dict[str, Any]]

search_estimators(query)[source]

Search estimators by name, module, or docstring.

Parameters:: query (str) – Search string (case-insensitive)
Return type:: list[EstimatorNode]
Returns:: List of matching EstimatorNode objects

class sktime_mcp.TagResolver[source]

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__()[source]: Initialize the tag resolver.

property tag_definitions: dict[str, TagInfo]: Get tag definitions, loading them if necessary.

get_tag_info(tag_name)[source]

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)[source]

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)[source]

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()[source]

Get list of all tag categories.

Return type:: list[str]

explain_tags(tags)[source]

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)[source]

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)[source]

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)[source]

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)

class sktime_mcp.CompositionValidator[source]

Bases: object

Validator for sktime estimator compositions.

This class encodes the rules for valid estimator compositions in sktime, allowing validation at planning time rather than runtime.

COMPOSITION_RULES: list[CompositionRule] = [CompositionRule(source_task='transformation', target_task='forecasting', composition_type=<CompositionType.FORECASTING_PIPELINE: 'forecasting_pipeline'>, position='before', description='Transformers can be applied before forecasters in a pipeline'), CompositionRule(source_task='transformation', target_task='transformation', composition_type=<CompositionType.TRANSFORMER_PIPELINE: 'transformer_pipeline'>, position='before', description='Transformers can be chained together'), CompositionRule(source_task='forecasting', target_task='forecasting', composition_type=<CompositionType.ENSEMBLE: 'ensemble'>, position='any', description='Forecasters can be combined in an ensemble'), CompositionRule(source_task='classification', target_task='classification', composition_type=<CompositionType.ENSEMBLE: 'ensemble'>, position='any', description='Classifiers can be combined in an ensemble'), CompositionRule(source_task='transformation', target_task='classification', composition_type=<CompositionType.PIPELINE: 'pipeline'>, position='before', description='Transformers can be applied before classifiers'), CompositionRule(source_task='transformation', target_task='regression', composition_type=<CompositionType.PIPELINE: 'pipeline'>, position='before', description='Transformers can be applied before regressors')]

TRANSFORMER_CATEGORIES = {'BoxCoxTransformer': 'power_transform', 'DateTimeFeatures': 'feature_creator', 'Deseasonalize': 'seasonality_remover', 'Detrend': 'trend_remover', 'Differencer': 'differencer', 'Imputer': 'missing_value_handler', 'Lag': 'lag_creator', 'LogTransformer': 'power_transform', 'ScaledLogitTransformer': 'power_transform', 'WindowSummarizer': 'feature_creator'}

__init__()[source]: Initialize the validator.

validate_pipeline(components)[source]

Validate a proposed pipeline composition.

Parameters:: components (list[str]) – List of estimator names in pipeline order
Return type:: ValidationResult
Returns:: ValidationResult with validity status and any issues

get_valid_compositions(estimator_name)[source]

Get valid compositions for an estimator.

Parameters:: estimator_name (str) – Name of the estimator
Return type:: dict[str, list[str]]
Returns:: Dictionary with “can_precede” and “can_follow” lists

suggest_pipeline(task, requirements=None)[source]

Suggest a valid pipeline for a given task.

Parameters:

task (str) – Target task (e.g., “forecasting”)
requirements (dict[str, Any] | None) – Optional requirements (e.g., {“handles_missing”: True})

Return type:

list[str]

Returns:

List of suggested estimator names forming a valid pipeline

class sktime_mcp.Executor[source]

Bases: object

Execution runtime for sktime estimators.

Handles instantiation, fitting, and prediction.

__init__()[source]

instantiate(estimator_name, params=None)[source]

Instantiate an estimator and return a handle.

Return type:: dict[str, Any]

load_dataset(name)[source]

Load a demo dataset.

Return type:: dict[str, Any]

fit(handle_id, y, X=None, fh=None)[source]

Fit an estimator.

Return type:: dict[str, Any]

predict(handle_id, fh=None, X=None)[source]

Generate predictions.

Return type:: dict[str, Any]

fit_predict(handle_id, dataset, horizon=12, data_handle=None)[source]

Convenience method: load data, fit, and predict.

Return type:: dict[str, Any]

async fit_predict_async(handle_id, dataset=None, data_handle=None, horizon=12, job_id=None)[source]

Async version of fit_predict with job tracking.

Runs the training in the background without blocking the MCP server. Accepts either a demo dataset name or a data handle from load_data_source.

Parameters:

handle_id (str) – Estimator handle
dataset (str | None) – Demo dataset name
data_handle (str | None) – Data handle from load_data_source
horizon (int) – Forecast horizon
job_id (str | None) – Optional job ID for tracking (created if not provided)

Return type:

dict[str, Any]

Returns:

Dictionary with success status and job_id

instantiate_pipeline(components, params_list=None)[source]

Instantiate a pipeline from a list of components.

Parameters:

components (list[str]) – List of estimator names in pipeline order
params_list (list[dict[str, Any]] | None) – Optional list of parameter dicts for each component

Return type:

dict[str, Any]

Returns:

Dictionary with success status and handle

list_datasets()[source]

List available demo datasets.

Return type:: list[str]

load_data_source(config)[source]

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)[source]

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)[source]

Format data associated with a handle.

Return type:: dict[str, Any]

list_data_handles()[source]

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)[source]

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

class sktime_mcp.HandleManager(max_handles=100)[source]

Bases: object

Manager for estimator instance handles.

__init__(max_handles=100)[source]

create_handle(estimator_name, instance, params=None, metadata=None)[source]

Return type:: str

get_instance(handle_id)[source]

Return type:: Any

get_info(handle_id)[source]

Return type:: HandleInfo

exists(handle_id)[source]

Return type:: bool

mark_fitted(handle_id)[source]

Return type:: None

is_fitted(handle_id)[source]

Return type:: bool

release_handle(handle_id)[source]

Return type:: bool

list_handles()[source]

Return type:: list[dict[str, Any]]

clear_all()[source]

Return type:: int

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)[source]

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()[source]

List all available MCP tools.

Return type:: list[Tool]

async sktime_mcp.server.call_tool(name, arguments)[source]

Handle tool calls.

Return type:: list[TextContent]

async sktime_mcp.server.run_server()[source]: Run the MCP server.

sktime_mcp.server.main()[source]: Main entry point.

Tools

Tools module for sktime MCP.

sktime_mcp.tools.list_estimators_tool(task=None, tags=None, query=None, limit=50, offset=0)[source]

Discover sktime estimators by task type, capability tags, and/or name search.

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

Parameters:

task (str | None) – Filter by task type. Options: “forecasting”, “classification”, “regression”, “transformation”, “clustering”, “detection”
tags (dict[str, Any] | None) – Filter by capability tags. Example: {“capability:pred_int”: True}
query (str | None) – Search by name or description (substring, case-insensitive).
limit (int) – Maximum number of results to return (default: 50)
offset (int) – Number of results to skip for pagination (default: 0).

Returns:

success: bool
estimators: List of estimator summaries
count: Number of results returned in this page
total: Total matching estimators (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:

Dictionary with

sktime_mcp.tools.get_available_tags()[source]

Get list of all available capability tags.

Return type:: dict[str, Any]

sktime_mcp.tools.describe_estimator_tool(estimator)[source]

Get detailed information about a specific estimator.

Parameters:

estimator (str) – Name of the estimator class (e.g., “ARIMA”, “RandomForest”)

Returns:

success: bool
name: Estimator name
task: Task type
module: Full module path
hyperparameters: 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

Example

>>> describe_estimator_tool("ARIMA")
{
    "success": True,
    "name": "ARIMA",
    "task": "forecasting",
    "hyperparameters": {"order": {"default": [1,0,0], "required": False}, ...},
    "tags": {"capability:pred_int": True, ...},
    ...
}

sktime_mcp.tools.instantiate_estimator_tool(estimator, params=None)[source]

Create an estimator instance and return a handle.

Parameters:

estimator (str) – Name of the estimator class (e.g., “ARIMA”)
params (dict[str, Any] | None) – Optional hyperparameters for the estimator

Returns:

success: bool
handle: Unique handle ID string
estimator: Name of the estimator
params: Parameters used
warnings: List of any validation warnings

Return type:

Dictionary with

Example

>>> instantiate_estimator_tool("ARIMA", {"order": [1, 1, 1]})
{
    "success": True,
    "handle": "est_abc123def456",
    "estimator": "ARIMA",
    "params": {"order": [1, 1, 1]}
}

sktime_mcp.tools.instantiate_pipeline_tool(components, params_list=None)[source]

Create a pipeline from a list of components and return a handle.

Parameters:

components (list[str]) – List of estimator names in pipeline order
params_list (list[dict[str, Any]] | None) – Optional list of parameter dicts for each component

Returns:

success: bool
handle: Unique handle ID string
pipeline: Name of the pipeline
components: List of component names
params_list: Parameters used for each component
warnings: List of any validation warnings

Return type:

Dictionary with

Example

>>> instantiate_pipeline_tool(
...     ["ConditionalDeseasonalizer", "Detrender", "ARIMA"],
...     [{}, {}, {"order": [1, 1, 1]}]
... )
{
    "success": True,
    "handle": "est_xyz789abc123",
    "pipeline": "ConditionalDeseasonalizer → Detrender → ARIMA",
    "components": ["ConditionalDeseasonalizer", "Detrender", "ARIMA"],
    "params_list": [{}, {}, {"order": [1, 1, 1]}]
}

sktime_mcp.tools.list_handles_tool()[source]

List all active estimator handles.

Return type:: dict[str, Any]
Returns:: Dictionary with list of active handles and their info

sktime_mcp.tools.release_handle_tool(handle)[source]

Release an estimator handle and free resources.

Parameters:: handle (str) – The handle ID to release
Return type:: dict[str, Any]
Returns:: Dictionary with success status

sktime_mcp.tools.load_model_tool(path)[source]

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. .. rubric:: Examples

”/tmp/my_arima_model”
”runs:/<run_id>/model”
”mlflow-artifacts:/<run_id>/artifacts/model”
”models:/<model_name>/<version>”

Return type:

dict[str, Any]

Returns:

Dictionary with success status and the new handle.

sktime_mcp.tools.fit_predict_tool(estimator_handle, dataset, horizon=12, data_handle=None)[source]

Execute a complete fit-predict workflow.

Parameters:

estimator_handle (str) – Handle from instantiate_estimator
dataset (str) – Name of demo dataset (e.g., “airline”, “sunspots”)
horizon (int) – Forecast horizon (default: 12)
data_handle (str | None) – Optional handle from load_data_source for custom data

Returns:

success: bool
predictions: Forecast values
horizon: Number of steps predicted

Return type:

Dictionary with

Example

>>> fit_predict_tool("est_abc123", "airline", horizon=12)
{
    "success": True,
    "predictions": {1: 450.2, 2: 460.5, ...},
    "horizon": 12
}

sktime_mcp.tools.fit_predict_async_tool(estimator_handle, dataset=None, data_handle=None, horizon=12)[source]

Execute a fit-predict workflow in the background (non-blocking).

Schedules the training as a background job and returns immediately with a job_id. Use check_job_status to monitor progress.

Accepts either a demo dataset name or a data handle from load_data_source – exactly one must be provided.

Parameters:

estimator_handle (str) – Handle from instantiate_estimator
dataset (str | None) – Name of demo dataset (e.g., “airline”, “sunspots”)
data_handle (str | None) – Handle from load_data_source (e.g., “data_abc123”)
horizon (int) – Forecast horizon (default: 12)

Returns:

success: bool
job_id: Job ID for tracking progress
message: Information about the job

Return type:

Dictionary with

Example

>>> fit_predict_async_tool("est_abc123", dataset="airline", horizon=12)
>>> fit_predict_async_tool("est_abc123", data_handle="data_xyz", horizon=5)

sktime_mcp.tools.evaluate_estimator_tool(estimator_handle, dataset, cv_folds=3)[source]

Evaluate an estimator using cross-validation.

Parameters:

estimator_handle (str) – Handle from instantiate_estimator
dataset (str) – Name of demo dataset
cv_folds (int) – Number of folds for Splitter

Return type:

dict[str, Any]

Returns:

Dictionary with cross-validation results

sktime_mcp.tools.load_data_source_tool(config)[source]

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

Parameters:

config (dict[str, Any]) –

Data source configuration {

”type”: “pandas” | “sql” | “file” | “url”, … (type-specific configuration)

}

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

Examples

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

# SQL Database >>> load_data_source_tool({ … “type”: “sql”, … “connection_string”: “postgresql://user:pass@host:5432/db”, … “query”: “SELECT date, value FROM sales”, … “time_column”: “date”, … “target_column”: “value” … })

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

sktime_mcp.tools.load_data_source_async_tool(config)[source]

Load data from any source in the background (non-blocking).

Schedules the data loading as a background job and returns immediately with a job_id. Use check_job_status to monitor progress and retrieve the data_handle when done.

Parameters:

config (dict[str, Any]) – Data source configuration (same as load_data_source)

Returns:

success: bool
job_id: Job ID for tracking progress
message: Information about the job

Return type:

Dictionary with

Example

>>> load_data_source_async_tool({
...     "type": "file",
...     "path": "/path/to/large_data.csv",
...     "time_column": "date",
...     "target_column": "value"
... })
{
    "success": True,
    "job_id": "abc-123-def-456",
    "message": "Data loading job started..."
}

sktime_mcp.tools.release_data_handle_tool(data_handle)[source]

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.tools.list_available_data_tool(is_demo=None)[source]

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: list of demo dataset name strings (e.g. [“airline”, “lynx”, …])
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)[source]

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.export_code_tool(handle, var_name='model', include_fit_example=False, dataset=None)[source]

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_estimator_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)[source]

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

Parameters:

estimator_handle (str) – Handle ID from instantiate_estimator / instantiate_pipeline
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)[source]

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)[source]

List all background jobs.

Parameters:

status (str | None) – Filter by status (pending, running, completed, failed, cancelled)
limit (int) – Maximum number of jobs to return

Return type:

dict[str, Any]

Returns:

Dictionary with list of jobs

sktime_mcp.tools.cancel_job_tool(job_id, delete=False)[source]

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

Registry

Registry module for sktime MCP.

class sktime_mcp.registry.EstimatorNode(name, task, class_ref, module, tags=<factory>, hyperparameters=<factory>, docstring=None)[source]

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.

name: The class name of the estimator (e.g., “ARIMA”)

task: The task type (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

hyperparameters: List of hyperparameter names with their defaults

docstring: The estimator’s docstring for understanding usage

name: str

task: str

class_ref: type

module: str

tags: dict[str, Any]

hyperparameters: dict[str, Any]

docstring: str | None = None

to_dict()[source]

Convert to dictionary for JSON serialization.

Return type:: dict[str, Any]

to_summary()[source]

Return a minimal summary for list operations.

Return type:: dict[str, Any]

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

class sktime_mcp.registry.RegistryInterface[source]

Bases: object

Interface to sktime’s estimator registry.

This class wraps sktime’s all_estimators function and provides structured access to estimator metadata, tags, and documentation.

The registry is the single source of truth for all estimator information.

TASK_MAP = {'classifier': 'classification', 'clusterer': 'clustering', 'detector': 'detection', 'forecaster': 'forecasting', 'network': 'network', 'param_est': 'parameter_estimation', 'regressor': 'regression', 'splitter': 'splitting', 'transformer': 'transformation'}

__init__()[source]: Initialize the registry interface.

get_all_estimators(task=None, tags=None)[source]

Get all estimators, optionally filtered by task and tags.

Parameters:

task (str | None) – Filter by task type (e.g., “forecasting”, “classification”)
tags (dict[str, Any] | None) – Filter by capability tags (e.g., {“capability:pred_int”: True})

Return type:

list[EstimatorNode]

Returns:

List of matching EstimatorNode objects

get_estimator_by_name(name)[source]

Get a specific estimator by its class name.

Parameters:: name (str) – The class name of the estimator (e.g., “ARIMA”)
Return type:: EstimatorNode | None
Returns:: EstimatorNode if found, None otherwise

get_available_tasks()[source]

Get list of available task types.

Return type:: list[str]

get_available_tags()[source]

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

Returns a list of dicts, each containing: - tag: the tag name (e.g., “scitype:y”) - description: human-readable explanation of what the tag means - value_type: the expected value type (e.g., “bool”, “str”) - applies_to: list of estimator types this tag applies to

Return type:: list[dict[str, Any]]

search_estimators(query)[source]

Search estimators by name, module, or docstring.

Parameters:: query (str) – Search string (case-insensitive)
Return type:: list[EstimatorNode]
Returns:: List of matching EstimatorNode objects

class sktime_mcp.registry.TagResolver[source]

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__()[source]: Initialize the tag resolver.

property tag_definitions: dict[str, TagInfo]: Get tag definitions, loading them if necessary.

get_tag_info(tag_name)[source]

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)[source]

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)[source]

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()[source]

Get list of all tag categories.

Return type:: list[str]

explain_tags(tags)[source]

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)[source]

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)[source]

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)[source]

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)

Runtime

Runtime module for sktime MCP.

class sktime_mcp.runtime.HandleManager(max_handles=100)[source]

Bases: object

Manager for estimator instance handles.

__init__(max_handles=100)[source]

create_handle(estimator_name, instance, params=None, metadata=None)[source]

Return type:: str

get_instance(handle_id)[source]

Return type:: Any

get_info(handle_id)[source]

Return type:: HandleInfo

exists(handle_id)[source]

Return type:: bool

mark_fitted(handle_id)[source]

Return type:: None

is_fitted(handle_id)[source]

Return type:: bool

release_handle(handle_id)[source]

Return type:: bool

list_handles()[source]

Return type:: list[dict[str, Any]]

clear_all()[source]

Return type:: int

class sktime_mcp.runtime.Executor[source]

Bases: object

Execution runtime for sktime estimators.

Handles instantiation, fitting, and prediction.

__init__()[source]

instantiate(estimator_name, params=None)[source]

Instantiate an estimator and return a handle.

Return type:: dict[str, Any]

load_dataset(name)[source]

Load a demo dataset.

Return type:: dict[str, Any]

fit(handle_id, y, X=None, fh=None)[source]

Fit an estimator.

Return type:: dict[str, Any]

predict(handle_id, fh=None, X=None)[source]

Generate predictions.

Return type:: dict[str, Any]

fit_predict(handle_id, dataset, horizon=12, data_handle=None)[source]

Convenience method: load data, fit, and predict.

Return type:: dict[str, Any]

async fit_predict_async(handle_id, dataset=None, data_handle=None, horizon=12, job_id=None)[source]

Async version of fit_predict with job tracking.

Runs the training in the background without blocking the MCP server. Accepts either a demo dataset name or a data handle from load_data_source.

Parameters:

handle_id (str) – Estimator handle
dataset (str | None) – Demo dataset name
data_handle (str | None) – Data handle from load_data_source
horizon (int) – Forecast horizon
job_id (str | None) – Optional job ID for tracking (created if not provided)

Return type:

dict[str, Any]

Returns:

Dictionary with success status and job_id

instantiate_pipeline(components, params_list=None)[source]

Instantiate a pipeline from a list of components.

Parameters:

components (list[str]) – List of estimator names in pipeline order
params_list (list[dict[str, Any]] | None) – Optional list of parameter dicts for each component

Return type:

dict[str, Any]

Returns:

Dictionary with success status and handle

list_datasets()[source]

List available demo datasets.

Return type:: list[str]

load_data_source(config)[source]

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)[source]

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)[source]

Format data associated with a handle.

Return type:: dict[str, Any]

list_data_handles()[source]

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)[source]

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

Composition

Composition module for sktime MCP.

class sktime_mcp.composition.CompositionValidator[source]

Bases: object

Validator for sktime estimator compositions.

This class encodes the rules for valid estimator compositions in sktime, allowing validation at planning time rather than runtime.

COMPOSITION_RULES: list[CompositionRule] = [CompositionRule(source_task='transformation', target_task='forecasting', composition_type=<CompositionType.FORECASTING_PIPELINE: 'forecasting_pipeline'>, position='before', description='Transformers can be applied before forecasters in a pipeline'), CompositionRule(source_task='transformation', target_task='transformation', composition_type=<CompositionType.TRANSFORMER_PIPELINE: 'transformer_pipeline'>, position='before', description='Transformers can be chained together'), CompositionRule(source_task='forecasting', target_task='forecasting', composition_type=<CompositionType.ENSEMBLE: 'ensemble'>, position='any', description='Forecasters can be combined in an ensemble'), CompositionRule(source_task='classification', target_task='classification', composition_type=<CompositionType.ENSEMBLE: 'ensemble'>, position='any', description='Classifiers can be combined in an ensemble'), CompositionRule(source_task='transformation', target_task='classification', composition_type=<CompositionType.PIPELINE: 'pipeline'>, position='before', description='Transformers can be applied before classifiers'), CompositionRule(source_task='transformation', target_task='regression', composition_type=<CompositionType.PIPELINE: 'pipeline'>, position='before', description='Transformers can be applied before regressors')]

TRANSFORMER_CATEGORIES = {'BoxCoxTransformer': 'power_transform', 'DateTimeFeatures': 'feature_creator', 'Deseasonalize': 'seasonality_remover', 'Detrend': 'trend_remover', 'Differencer': 'differencer', 'Imputer': 'missing_value_handler', 'Lag': 'lag_creator', 'LogTransformer': 'power_transform', 'ScaledLogitTransformer': 'power_transform', 'WindowSummarizer': 'feature_creator'}

__init__()[source]: Initialize the validator.

validate_pipeline(components)[source]

Validate a proposed pipeline composition.

Parameters:: components (list[str]) – List of estimator names in pipeline order
Return type:: ValidationResult
Returns:: ValidationResult with validity status and any issues

get_valid_compositions(estimator_name)[source]

Get valid compositions for an estimator.

Parameters:: estimator_name (str) – Name of the estimator
Return type:: dict[str, list[str]]
Returns:: Dictionary with “can_precede” and “can_follow” lists

suggest_pipeline(task, requirements=None)[source]

Suggest a valid pipeline for a given task.

Parameters:

task (str) – Target task (e.g., “forecasting”)
requirements (dict[str, Any] | None) – Optional requirements (e.g., {“handles_missing”: True})

Return type:

list[str]

Returns:

List of suggested estimator names forming a valid pipeline

class sktime_mcp.composition.ValidationResult(valid, errors=<factory>, warnings=<factory>, suggestions=<factory>)[source]

Bases: object

Result of a composition validation.

valid: Whether the composition is valid

errors: List of validation errors

warnings: List of warnings (valid but potentially problematic)

suggestions: Suggested fixes for invalid compositions

valid: bool

errors: list[str]

warnings: list[str]

suggestions: list[str]

to_dict()[source]

Convert to dictionary for JSON serialization.

Return type:: dict[str, Any]

__init__(valid, errors=<factory>, warnings=<factory>, suggestions=<factory>)

class sktime_mcp.composition.CompositionRule(source_task, target_task, composition_type, position, description)[source]

Bases: object

A rule describing valid compositions for an estimator type.

source_task: The task type that can be composed

target_task: The task type it can compose with

composition_type: Type of composition

position: Where in the pipeline (before, after, any)

description: Human-readable description

source_task: str

target_task: str

composition_type: CompositionType

position: str

description: str

__init__(source_task, target_task, composition_type, position, description)