Upload 14 files

Dockerfile

@@ -1,18 +1,23 @@
-FROM python:3.10
+FROM python:3.11-slim
 
-RUN useradd -m -u 1000 user && python -m pip install --upgrade pip
-USER user
-ENV PATH="/home/user/.local/bin:$PATH"
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
 
 WORKDIR /app
 
-COPY --chown=user ./requirements.txt requirements.txt
-RUN pip install --no-cache-dir --upgrade -r requirements.txt
+RUN useradd -m -u 1000 appuser
+
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+
+COPY mne-python /app/mne-python
+COPY app.py /app/app.py
 
-COPY --chown=user . /app
 ENV MCP_TRANSPORT=http
 ENV MCP_PORT=7860
 
 EXPOSE 7860
 
-CMD ["python", "mne-python/mcp_output/start_mcp.py"]
+USER appuser
+
+ENTRYPOINT ["python", "mne-python/mcp_output/start_mcp.py"]

README.md

@@ -1,10 +1,65 @@
 ---
-title: Mne Python
-emoji: 🏃
-colorFrom: pink
+title: mne-python MCP Service
+emoji: 🔧
+colorFrom: blue
 colorTo: indigo
 sdk: docker
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# mne-python MCP Service
+
+This deployment package exposes core MNE-Python capabilities through a FastMCP service.
+
+## Available MCP Tools
+
+- `health_check`
+- `get_mne_version`
+- `list_builtin_montages`
+- `create_info`
+- `create_raw_array`
+- `filter_data_array`
+- `compute_psd_welch`
+- `list_loaded_modules`
+
+## Local stdio usage (Claude Desktop / CLI)
+
+From `mne-python/mcp_output/mcp_plugin/`:
+
+```bash
+python main.py
+```
+
+Or from `mne-python/mcp_output/`:
+
+```bash
+python start_mcp.py
+```
+
+Default transport is `stdio`.
+
+## HTTP client usage
+
+Run with HTTP transport:
+
+```bash
+MCP_TRANSPORT=http MCP_PORT=8000 python mne-python/mcp_output/start_mcp.py
+```
+
+Then connect MCP clients to:
+
+- `http://127.0.0.1:8000/mcp`
+
+## Docker / HuggingFace Spaces
+
+This package is configured for HuggingFace Spaces Docker deployment:
+
+- fixed port `7860`
+- non-root runtime user UID `1000`
+- MCP HTTP endpoint served directly by FastMCP
+- container entrypoint: `python mne-python/mcp_output/start_mcp.py`
+
+Clients should connect to:
+
+- `https://<your-space>.hf.space/mcp`

app.py

@@ -1,45 +1,74 @@
-from fastapi import FastAPI
+from __future__ import annotations
+
 import os
+from pathlib import Path
 import sys
+import importlib
+
+try:
+    FastAPI = importlib.import_module("fastapi").FastAPI
+except Exception:  # pragma: no cover
+    FastAPI = None
+
+
+BASE_DIR = Path(__file__).resolve().parent
+MCP_OUTPUT_DIR = BASE_DIR / "mne-python" / "mcp_output"
+output_str = str(MCP_OUTPUT_DIR)
+if output_str not in sys.path:
+    sys.path.insert(0, output_str)
 
-mcp_plugin_path = os.path.join(os.path.dirname(__file__), "mne-python", "mcp_output", "mcp_plugin")
-sys.path.insert(0, mcp_plugin_path)
 
-app = FastAPI(
-    title="Mne-Python MCP Service",
-    description="Auto-generated MCP service for mne-python",
-    version="1.0.0"
-)
+if FastAPI is None:
+    raise RuntimeError("fastapi is required to run app.py")
+
+
+app = FastAPI(title="mne-python MCP Companion API", version="1.0.0")
+
 
 @app.get("/")
-def root():
+def root() -> dict[str, object]:
     return {
-        "service": "Mne-Python MCP Service",
-        "version": "1.0.0",
-        "status": "running",
-        "transport": os.environ.get("MCP_TRANSPORT", "http")
+        "service": "mne-python-mcp",
+        "description": "Supplementary API for MCP deployment info.",
+        "mcp_transport_env": os.getenv("MCP_TRANSPORT", "stdio"),
+        "mcp_port_env": int(os.getenv("MCP_PORT", "7860")),
+        "note": "This FastAPI app is informational and does not run MCP itself.",
     }
 
+
 @app.get("/health")
-def health_check():
-    return {"status": "healthy", "service": "mne-python MCP"}
+def health() -> dict[str, str]:
+    return {"status": "healthy"}
+
 
 @app.get("/tools")
-def list_tools():
+def list_tools() -> dict[str, object]:
     try:
-        from mcp_service import create_app
-        mcp_app = create_app()
-        tools = []
-        for tool_name, tool_func in mcp_app.tools.items():
-            tools.append({
-                "name": tool_name,
-                "description": tool_func.__doc__ or "No description available"
-            })
-        return {"tools": tools}
-    except Exception as e:
-        return {"error": f"Failed to load tools: {str(e)}"}
+        mcp_module = importlib.import_module("mcp_plugin.mcp_service")
+        create_app = getattr(mcp_module, "create_app")
+        mcp = create_app()
+        raw_tools = getattr(mcp, "tools", [])
+        tools_list: list[dict[str, str]] = []
+
+        if isinstance(raw_tools, dict):
+            iterable = raw_tools.values()
+        else:
+            iterable = raw_tools
+
+        for tool in iterable:
+            name = getattr(tool, "name", None) or getattr(tool, "__name__", "unknown")
+            description = getattr(tool, "description", None) or getattr(tool, "__doc__", "")
+            tools_list.append({"name": str(name), "description": str(description).strip()})
+
+        return {"count": len(tools_list), "tools": tools_list}
+    except Exception as exc:
+        return {"count": 0, "tools": [], "error": str(exc)}
+
+
+PORT = int(os.getenv("PORT", "7860"))
+
 
 if __name__ == "__main__":
     import uvicorn
-    port = int(os.environ.get("PORT", 7860))
-    uvicorn.run(app, host="0.0.0.0", port=port)
+
+    uvicorn.run(app, host="0.0.0.0", port=PORT)

mne-python/mcp_output/README_MCP.md

@@ -1,129 +1,158 @@
-# MNE-Python MCP (Model Context Protocol) Service README
-
-## 1) Project Introduction
-
-This service wraps core **MNE-Python** capabilities for EEG/MEG workflows in an MCP (Model Context Protocol)-friendly interface.
-
-Primary goals:
-- Load raw neurophysiology data (FIF, EDF, BDF, BrainVision).
-- Perform event handling and channel picking.
-- Run common preprocessing (EOG/ECG detection, ICA-related preparation).
-- Compute PSD and time-frequency features.
-- Optionally render diagnostic visualizations (when GUI/backends are available).
-
-Recommended integration path:
-- **Primary:** in-process Python imports from `mne` modules.
-- **Fallback:** call the `mne` CLI for heavier or isolated execution contexts.
-
----
-
-## 2) Installation Method
-
-### Requirements
-Core dependencies:
-- `numpy`
-- `scipy`
-- `matplotlib`
-- `packaging`
-- `pooch`
-- `tqdm`
-
-Common optional dependencies (feature-dependent):
-- `scikit-learn`, `pandas`, `h5py`, `nibabel`
-- `pyvista`, `vtk`, `mne-qt-browser`
-- `numba`
-
-### Install
-- Install MNE-Python and common scientific stack via pip:
-  - `pip install mne`
-- For broader functionality:
-  - `pip install mne[full]` (if supported by your environment/version)
-- If your service uses project-local dependency files, prefer:
-  - `pyproject.toml` / `environment.yml` in your deployment workflow.
-
----
-
-## 3) Quick Start
-
-### Minimal service flow (Python-side)
-1. Read raw data:
-   - `mne.io.read_raw_fif(...)` / `read_raw_edf(...)` / `read_raw_bdf(...)` / `read_raw_brainvision(...)`
-2. Extract events:
-   - `mne.find_events(raw)` or `mne.read_events(path)`
-3. Pick channels:
-   - `mne.pick_types(raw.info, meg=True, eeg=True, eog=True, exclude="bads")`
-4. Preprocess (optional):
-   - `mne.preprocessing.find_eog_events(raw)`, `find_ecg_events(raw)`, ICA workflows
-5. Spectral analysis:
-   - `mne.time_frequency.psd_array_welch(...)` or `psd_array_multitaper(...)`
-6. Return structured results from your MCP (Model Context Protocol) service endpoint.
-
-### CLI fallback
-- Use `mne` command wrapper when import-time overhead or environment isolation is needed.
-
----
-
-## 4) Available Tools and Endpoints List
-
-Suggested MCP (Model Context Protocol) service endpoints:
-
-- `load_raw`
-  - Load raw recordings from FIF/EDF/BDF/BrainVision.
-  - Maps to `mne.io.read_raw_*`.
-
-- `read_info`
-  - Read metadata/header info without full processing.
-  - Maps to `mne.io.read_info`.
-
-- `events_detect`
-  - Detect or load event markers.
-  - Maps to `mne.find_events`, `mne.read_events`, `mne.merge_events`.
-
-- `channels_pick`
-  - Build channel selections by modality/type.
-  - Maps to `mne.pick_types`.
-
-- `preprocess_eog_ecg`
-  - EOG/ECG event detection and projection helpers.
-  - Maps to `mne.preprocessing.find_eog_events`, `find_ecg_events`, `compute_proj_eog`, `compute_proj_ecg`.
-
-- `ica_workflow`
-  - Artifact decomposition/removal pipeline.
-  - Maps to `mne.preprocessing.ICA`, `EOGRegression`.
-
-- `psd_compute`
-  - Power spectral density calculations.
-  - Maps to `mne.time_frequency.psd_array_welch`, `psd_array_multitaper`.
-
-- `tfr_compute`
-  - Time-frequency decomposition.
-  - Maps to `tfr_morlet`, `tfr_multitaper`, `csd_multitaper`.
-
-- `viz_diagnostics` (optional/headless-sensitive)
-  - Event/covariance/alignment diagnostic plotting.
-  - Maps to `mne.viz.plot_events`, `plot_cov`, `plot_bem`, `plot_alignment`.
-
-- `cli_exec` (fallback)
-  - Run `mne` subcommands for isolated or heavyweight tasks.
-
----
-
-## 5) Common Issues and Notes
-
-- **Complexity:** MNE-Python is feature-rich; keep endpoint contracts narrow and typed.
-- **Import overhead:** First import can be non-trivial; consider lazy-loading per endpoint.
-- **GUI dependencies:** Visualization endpoints may fail in headless servers unless backend is configured.
-- **Optional packages:** Some analyses silently require extras (`sklearn`, `h5py`, `nibabel`, etc.).
-- **Performance:** Large FIF files and TFR/ICA jobs are memory/CPU intensive; set resource limits.
-- **Reproducibility:** Pin MNE + NumPy/SciPy versions in deployment.
-- **Risk profile:** Import feasibility is moderate (~0.72); keep CLI fallback available.
-- **Data handling:** Validate paths, file formats, and channel metadata before heavy processing.
-
----
-
-## 6) Reference Links / Documentation
-
-- MNE-Python repository: https://github.com/mne-tools/mne-python
-- MNE official docs: https://mne.tools/stable/index.html
-- MNE API reference: https://mne.tools/stable/python_reference.html
-- MNE command-line tools: https://mne.tools/stable/overview/command_line.html
+# MNE-Python MCP Layer
+
+This directory contains the MCP plugin and launch scripts for exposing selected MNE-Python capabilities.
+
+## Exposed MCP Tools
+
+### 1) `health_check`
+- **Description**: Reports dependency availability and adapter health.
+- **Parameters**: none
+- **Example**:
+```json
+{"tool": "health_check", "arguments": {}}
+```
+
+### 2) `get_mne_version`
+- **Description**: Returns installed MNE version and module location.
+- **Parameters**: none
+- **Example**:
+```json
+{"tool": "get_mne_version", "arguments": {}}
+```
+
+### 3) `list_builtin_montages`
+- **Description**: Lists available built-in electrode montages.
+- **Parameters**: none
+- **Example**:
+```json
+{"tool": "list_builtin_montages", "arguments": {}}
+```
+
+### 4) `create_info`
+- **Description**: Creates an MNE `Info` object and returns a summary.
+- **Parameters**:
+  - `ch_names: list[str]`
+  - `sfreq: float`
+  - `ch_types: str = "eeg"`
+- **Example**:
+```json
+{
+  "tool": "create_info",
+  "arguments": {
+    "ch_names": ["Fz", "Cz", "Pz"],
+    "sfreq": 250.0,
+    "ch_types": "eeg"
+  }
+}
+```
+
+### 5) `create_raw_array`
+- **Description**: Builds an in-memory `RawArray` from matrix data and returns metadata.
+- **Parameters**:
+  - `data: list[list[float]]` (shape `[n_channels, n_times]`)
+  - `sfreq: float`
+  - `ch_names: list[str]`
+  - `ch_types: str = "eeg"`
+- **Example**:
+```json
+{
+  "tool": "create_raw_array",
+  "arguments": {
+    "data": [[0.1, 0.2, 0.3], [0.0, -0.1, 0.1]],
+    "sfreq": 1000.0,
+    "ch_names": ["EEG001", "EEG002"],
+    "ch_types": "eeg"
+  }
+}
+```
+
+### 6) `filter_data_array`
+- **Description**: Applies MNE filtering to array data.
+- **Parameters**:
+  - `data: list[list[float]]`
+  - `sfreq: float`
+  - `l_freq: float | null`
+  - `h_freq: float | null`
+- **Example**:
+```json
+{
+  "tool": "filter_data_array",
+  "arguments": {
+    "data": [[0.1, 0.2, 0.3, 0.2], [0.0, -0.1, 0.1, 0.0]],
+    "sfreq": 250.0,
+    "l_freq": 1.0,
+    "h_freq": 40.0
+  }
+}
+```
+
+### 7) `compute_psd_welch`
+- **Description**: Computes Welch PSD from array data.
+- **Parameters**:
+  - `data: list[list[float]]`
+  - `sfreq: float`
+  - `fmin: float = 0.0`
+  - `fmax: float = 120.0`
+  - `n_fft: int = 256`
+- **Example**:
+```json
+{
+  "tool": "compute_psd_welch",
+  "arguments": {
+    "data": [[0.1, 0.2, 0.3, 0.2], [0.0, -0.1, 0.1, 0.0]],
+    "sfreq": 250.0,
+    "fmin": 1.0,
+    "fmax": 60.0,
+    "n_fft": 128
+  }
+}
+```
+
+### 8) `list_loaded_modules`
+- **Description**: Lists modules loaded by the adapter.
+- **Parameters**:
+  - `limit: int = 100`
+- **Example**:
+```json
+{"tool": "list_loaded_modules", "arguments": {"limit": 50}}
+```
+
+## Running Locally (stdio)
+
+From `mcp_output/`:
+
+```bash
+python start_mcp.py
+```
+
+Or directly:
+
+```bash
+python mcp_plugin/main.py
+```
+
+Both default to stdio transport for Claude Desktop / CLI style MCP clients.
+
+## Running as HTTP MCP Service
+
+```bash
+MCP_TRANSPORT=http MCP_PORT=8000 python start_mcp.py
+```
+
+The MCP HTTP endpoint is served at:
+
+- `http://127.0.0.1:8000/mcp`
+
+## Response Shape
+
+All MCP tools return:
+
+```json
+{
+  "success": true,
+  "result": {},
+  "error": null
+}
+```
+
+On errors, `success` is `false`, `result` is `null`, and `error` contains a message.

mne-python/mcp_output/mcp_plugin/adapter.py

@@ -1,358 +1,227 @@
-import os
+from __future__ import annotations
+
+from pathlib import Path
 import sys
-import importlib
-import traceback
-from typing import Any, Dict, List, Optional
 
-source_path = os.path.join(
-    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
-    "source",
-)
-sys.path.insert(0, source_path)
+try:
+    import importlib
+except Exception:
+    importlib = None
 
+try:
+    import inspect
+except Exception:
+    inspect = None
 
-class Adapter:
-    """
-    Import-mode adapter for the mne-python MCP plugin integration.
+try:
+    import pkgutil
+except Exception:
+    pkgutil = None
 
-    This adapter prioritizes Python imports and provides a graceful CLI fallback
-    pathway when import-based execution is unavailable.
-    """
 
-    # -------------------------------------------------------------------------
-    # Lifecycle / Initialization
-    # -------------------------------------------------------------------------
-    def __init__(self) -> None:
-        """
-        Initialize adapter state, import registry, and module availability.
+CURRENT_DIR = Path(__file__).resolve().parent
+SOURCE_DIR = CURRENT_DIR.parent.parent / "source"
+if SOURCE_DIR.exists():
+    source_str = str(SOURCE_DIR)
+    if source_str not in sys.path:
+        sys.path.insert(0, source_str)
 
-        Attributes:
-            mode: Adapter execution mode, fixed to "import".
-            package_root: Root import path for repository package.
-            modules: Loaded module cache keyed by logical name.
-            available: Boolean import readiness state.
-            warnings: List of non-fatal issues detected during initialization.
-        """
-        self.mode = "import"
-        self.package_root = "mne"
-        self.modules: Dict[str, Any] = {}
-        self.available = False
-        self.warnings: List[str] = []
-        self._initialize_imports()
 
-    def _initialize_imports(self) -> None:
-        """
-        Attempt to import core and command modules identified by analysis.
+class Adapter:
+    def __init__(self, root_package: str = "mne") -> None:
+        self.root_package = root_package
+        self.loaded_modules: dict[str, object] = {}
+        self.failed_modules: dict[str, str] = {}
+        self.mode = "normal"
+        self._load_all_submodules()
+
+    def _load_all_submodules(self) -> dict[str, object]:
+        if importlib is None:
+            self.mode = "blackbox"
+            return {
+                "status": "fallback",
+                "mode": self.mode,
+                "reason": "importlib unavailable",
+            }
 
-        This method captures and stores import exceptions so the adapter can
-        continue operating in graceful fallback mode.
-        """
-        targets = {
-            "mne": "mne",
-            "commands": "mne.commands",
+        try:
+            root_mod = importlib.import_module(self.root_package)
+            self.loaded_modules[self.root_package] = root_mod
+        except Exception as exc:
+            self.mode = "blackbox"
+            self.failed_modules[self.root_package] = str(exc)
+            return {
+                "status": "fallback",
+                "mode": self.mode,
+                "reason": f"failed to import root package {self.root_package}: {exc}",
+            }
+
+        if pkgutil is not None and hasattr(root_mod, "__path__"):
+            for module_info in pkgutil.walk_packages(
+                root_mod.__path__,
+                prefix=f"{self.root_package}.",
+            ):
+                name = module_info.name
+                try:
+                    mod = importlib.import_module(name)
+                    self.loaded_modules[name] = mod
+                except Exception as exc:
+                    self.failed_modules[name] = str(exc)
+
+        if not self.loaded_modules:
+            self.mode = "blackbox"
+            return {
+                "status": "fallback",
+                "mode": self.mode,
+                "reason": "no modules loaded",
+            }
+
+        return {
+            "status": "ok",
+            "mode": self.mode,
+            "loaded_count": len(self.loaded_modules),
+            "failed_count": len(self.failed_modules),
         }
 
-        loaded = 0
-        for key, mod_path in targets.items():
-            try:
-                self.modules[key] = importlib.import_module(mod_path)
-                loaded += 1
-            except Exception as exc:
-                self.modules[key] = None
-                self.warnings.append(
-                    f"Failed to import '{mod_path}'. Verify local source checkout and dependencies. Detail: {exc}"
-                )
-        self.available = loaded > 0
+    def health(self) -> dict[str, object]:
+        status = "fallback" if self.mode == "blackbox" else "ok"
+        return {
+            "status": status,
+            "mode": self.mode,
+            "root_package": self.root_package,
+            "loaded_count": len(self.loaded_modules),
+            "failed_count": len(self.failed_modules),
+        }
 
-    # -------------------------------------------------------------------------
-    # Unified response helpers
-    # -------------------------------------------------------------------------
-    def _ok(self, message: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
-        return {"status": "success", "message": message, "data": data or {}}
+    def list_modules(self, include_failed: bool = True) -> dict[str, object]:
+        status = "fallback" if self.mode == "blackbox" else "ok"
+        result: dict[str, object] = {
+            "status": status,
+            "mode": self.mode,
+            "loaded_modules": sorted(self.loaded_modules.keys()),
+        }
+        if include_failed:
+            result["failed_modules"] = self.failed_modules
+        return result
 
-    def _fail(
+    def list_symbols(
         self,
-        message: str,
-        error: Optional[str] = None,
-        guidance: Optional[str] = None,
-        data: Optional[Dict[str, Any]] = None,
-    ) -> Dict[str, Any]:
-        payload = {"status": "error", "message": message, "data": data or {}}
-        if error:
-            payload["error"] = error
-        if guidance:
-            payload["guidance"] = guidance
-        return payload
+        module_name: str,
+        include_private: bool = False,
+        only_callables: bool = False,
+    ) -> dict[str, object]:
+        module = self.loaded_modules.get(module_name)
+        if module is None:
+            return {
+                "status": "error",
+                "error": f"module not loaded: {module_name}",
+            }
+
+        names = dir(module)
+        if not include_private:
+            names = [name for name in names if not name.startswith("_")]
+
+        if only_callables:
+            symbols = []
+            for name in names:
+                try:
+                    value = getattr(module, name)
+                    if callable(value):
+                        symbols.append(name)
+                except Exception:
+                    continue
+        else:
+            symbols = names
 
-    def _fallback(
-        self,
-        action: str,
-        reason: str,
-        extra: Optional[Dict[str, Any]] = None,
-    ) -> Dict[str, Any]:
         return {
-            "status": "fallback",
-            "message": f"Import mode unavailable for '{action}'.",
-            "reason": reason,
-            "guidance": (
-                "Ensure repository source is present under the configured 'source' directory and "
-                "install required dependencies (numpy, scipy, matplotlib, packaging, pooch, tqdm). "
-                "If import still fails, use the CLI fallback via the 'mne' command."
-            ),
-            "data": extra or {},
+            "status": "ok",
+            "module": module_name,
+            "count": len(symbols),
+            "symbols": sorted(symbols),
         }
 
-    # -------------------------------------------------------------------------
-    # Health / Introspection
-    # -------------------------------------------------------------------------
-    def health_check(self) -> Dict[str, Any]:
-        """
-        Report adapter readiness and import diagnostics.
-
-        Returns:
-            Dict with status, mode, import availability, and warning details.
-        """
-        return self._ok(
-            "Adapter health check completed.",
-            {
+    def call_function(
+        self,
+        module_name: str,
+        function_name: str,
+        args: list[object] | None = None,
+        kwargs: dict[str, object] | None = None,
+    ) -> dict[str, object]:
+        module = self.loaded_modules.get(module_name)
+        if module is None:
+            return {
+                "status": "error",
+                "error": f"module not loaded: {module_name}",
+            }
+
+        if inspect is None:
+            return {
+                "status": "fallback",
                 "mode": self.mode,
-                "available": self.available,
-                "loaded_modules": {k: bool(v) for k, v in self.modules.items()},
-                "warnings": self.warnings,
-            },
-        )
-
-    def list_known_packages(self) -> Dict[str, Any]:
-        """
-        Return package namespaces discovered during analysis.
-
-        Returns:
-            Dict containing analyzed package names for discovery/debugging.
-        """
-        packages = [
-            "deployment.mne-python.source",
-            "mcp_output.mcp_plugin",
-            "source.mne",
-            "source.mne._fiff",
-            "source.mne.beamformer",
-            "source.mne.channels",
-            "source.mne.commands",
-            "source.mne.data",
-            "source.mne.datasets",
-            "source.mne.decoding",
-            "source.mne.export",
-            "source.mne.forward",
-            "source.mne.gui",
-            "source.mne.html_templates",
-            "source.mne.inverse_sparse",
-            "source.mne.io",
-            "source.mne.minimum_norm",
-            "source.mne.preprocessing",
-            "source.mne.report",
-            "source.mne.simulation",
-            "source.mne.source_space",
-            "source.mne.stats",
-            "source.mne.tests",
-            "source.mne.time_frequency",
-            "source.mne.utils",
-            "source.mne.viz",
-        ]
-        return self._ok("Known package list prepared.", {"packages": packages})
-
-    # -------------------------------------------------------------------------
-    # Module management
-    # -------------------------------------------------------------------------
-    def import_module(self, module_path: str) -> Dict[str, Any]:
-        """
-        Dynamically import an MNE module using full package path.
-
-        Args:
-            module_path: Absolute module path (e.g., 'mne.io', 'mne.preprocessing').
-
-        Returns:
-            Unified status dictionary with module import result.
-        """
+                "error": "inspect unavailable",
+            }
+
+        fn = getattr(module, function_name, None)
+        if fn is None or not callable(fn):
+            return {
+                "status": "error",
+                "error": f"callable not found: {module_name}.{function_name}",
+            }
+
+        local_args = args or []
+        local_kwargs = kwargs or {}
         try:
-            mod = importlib.import_module(module_path)
-            self.modules[module_path] = mod
-            return self._ok("Module imported successfully.", {"module_path": module_path})
+            output = fn(*local_args, **local_kwargs)
+            return {
+                "status": "ok",
+                "module": module_name,
+                "function": function_name,
+                "result": output,
+            }
         except Exception as exc:
-            return self._fail(
-                "Module import failed.",
-                error=str(exc),
-                guidance="Confirm module path and dependency installation.",
-                data={"module_path": module_path},
-            )
-
-    def get_module_attributes(self, module_path: str, limit: int = 200) -> Dict[str, Any]:
-        """
-        Enumerate public attributes for a module to support function discovery.
+            return {
+                "status": "error",
+                "module": module_name,
+                "function": function_name,
+                "error": str(exc),
+            }
 
-        Args:
-            module_path: Full module path.
-            limit: Maximum number of attributes to return.
-
-        Returns:
-            Unified status dictionary containing exported attribute names.
-        """
-        try:
-            mod = self.modules.get(module_path) or importlib.import_module(module_path)
-            attrs = [a for a in dir(mod) if not a.startswith("_")]
-            return self._ok(
-                "Module attributes fetched.",
-                {"module_path": module_path, "attributes": attrs[: max(1, limit)]},
-            )
-        except Exception as exc:
-            return self._fail(
-                "Could not inspect module attributes.",
-                error=str(exc),
-                guidance="Import the module first and verify it is available in local source.",
-                data={"module_path": module_path},
-            )
-
-    # -------------------------------------------------------------------------
-    # Core MNE call surface
-    # -------------------------------------------------------------------------
-    def call_mne_function(self, function_name: str, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        """
-        Call a function from the top-level mne module by name.
-
-        Args:
-            function_name: Name of the function in module 'mne'.
-            *args: Positional arguments forwarded to the target function.
-            **kwargs: Keyword arguments forwarded to the target function.
-
-        Returns:
-            Unified status dictionary with execution result or actionable failure.
-        """
-        if not self.modules.get("mne"):
-            return self._fallback("call_mne_function", "Top-level module 'mne' is not importable.")
-
-        try:
-            target = getattr(self.modules["mne"], function_name, None)
-            if target is None or not callable(target):
-                return self._fail(
-                    "Requested MNE function was not found.",
-                    guidance="Use get_module_attributes('mne') to discover available callables.",
-                    data={"function_name": function_name},
-                )
-            result = target(*args, **kwargs)
-            return self._ok(
-                "MNE function executed successfully.",
-                {"function_name": function_name, "result": result},
-            )
-        except Exception as exc:
-            return self._fail(
-                "MNE function execution failed.",
-                error=str(exc),
-                guidance="Validate function arguments and data formats expected by MNE.",
-                data={"function_name": function_name, "traceback": traceback.format_exc()},
-            )
-
-    # -------------------------------------------------------------------------
-    # CLI wrapper (fallback-friendly)
-    # -------------------------------------------------------------------------
-    def call_mne_cli(self, argv: Optional[List[str]] = None) -> Dict[str, Any]:
-        """
-        Execute the primary MNE command-line wrapper from imported command module.
-
-        Args:
-            argv: Optional list of CLI-like arguments. If omitted, attempts default call.
-
-        Returns:
-            Unified status dictionary with invocation status and hints.
-        """
-        commands_mod = self.modules.get("commands")
-        if not commands_mod:
-            return self._fallback("call_mne_cli", "Module 'mne.commands' is not importable.")
-
-        try:
-            entry_candidates = ["main", "run", "command_main"]
-            entry = None
-            for name in entry_candidates:
-                fn = getattr(commands_mod, name, None)
-                if callable(fn):
-                    entry = fn
-                    break
-
-            if entry is None:
-                return self._fail(
-                    "No callable CLI entry point found in mne.commands.",
-                    guidance="Inspect mne.commands module attributes and map the correct callable.",
-                    data={"checked_candidates": entry_candidates},
-                )
-
-            if argv is None:
-                out = entry()
-            else:
-                out = entry(argv)
-
-            return self._ok(
-                "MNE CLI wrapper executed.",
-                {"argv": argv or [], "result": out, "entry_point": entry.__name__},
-            )
-        except TypeError:
-            try:
-                out = entry()  # type: ignore[misc]
-                return self._ok(
-                    "MNE CLI wrapper executed with default signature.",
-                    {"argv": argv or [], "result": out, "entry_point": entry.__name__},  # type: ignore[union-attr]
-                )
-            except Exception as exc:
-                return self._fail(
-                    "MNE CLI invocation failed.",
-                    error=str(exc),
-                    guidance="Pass valid subcommands/arguments or verify command module compatibility.",
-                    data={"traceback": traceback.format_exc()},
-                )
-        except Exception as exc:
-            return self._fail(
-                "MNE CLI invocation failed.",
-                error=str(exc),
-                guidance="Check installed optional dependencies and command arguments.",
-                data={"traceback": traceback.format_exc()},
-            )
-
-    # -------------------------------------------------------------------------
-    # Class instantiation helper (generic, analysis-driven)
-    # -------------------------------------------------------------------------
     def create_instance(
         self,
-        module_path: str,
+        module_name: str,
         class_name: str,
-        *args: Any,
-        **kwargs: Any,
-    ) -> Dict[str, Any]:
-        """
-        Instantiate a class from a target module path.
-
-        Args:
-            module_path: Full module path containing the class.
-            class_name: Class name to instantiate.
-            *args: Positional constructor arguments.
-            **kwargs: Keyword constructor arguments.
-
-        Returns:
-            Unified status dictionary with instantiated object handle.
-        """
+        args: list[object] | None = None,
+        kwargs: dict[str, object] | None = None,
+    ) -> dict[str, object]:
+        module = self.loaded_modules.get(module_name)
+        if module is None:
+            return {
+                "status": "error",
+                "error": f"module not loaded: {module_name}",
+            }
+
+        cls = getattr(module, class_name, None)
+        if cls is None:
+            return {
+                "status": "error",
+                "error": f"class not found: {module_name}.{class_name}",
+            }
+
+        local_args = args or []
+        local_kwargs = kwargs or {}
         try:
-            mod = self.modules.get(module_path) or importlib.import_module(module_path)
-            cls = getattr(mod, class_name, None)
-            if cls is None:
-                return self._fail(
-                    "Class not found in module.",
-                    guidance="Use get_module_attributes(module_path) to verify class exports.",
-                    data={"module_path": module_path, "class_name": class_name},
-                )
-            instance = cls(*args, **kwargs)
-            return self._ok(
-                "Class instantiated successfully.",
-                {"module_path": module_path, "class_name": class_name, "instance": instance},
-            )
+            instance = cls(*local_args, **local_kwargs)
+            return {
+                "status": "ok",
+                "module": module_name,
+                "class": class_name,
+                "instance_type": type(instance).__name__,
+                "instance_repr": repr(instance),
+            }
         except Exception as exc:
-            return self._fail(
-                "Class instantiation failed.",
-                error=str(exc),
-                guidance="Verify constructor arguments and required dependencies for this class.",
-                data={"module_path": module_path, "class_name": class_name, "traceback": traceback.format_exc()},
-            )
+            return {
+                "status": "error",
+                "module": module_name,
+                "class": class_name,
+                "error": str(exc),
+            }

mne-python/mcp_output/mcp_plugin/main.py

@@ -1,13 +1,7 @@
-"""
-MCP Service Auto-Wrapper - Auto-generated
-"""
 from mcp_service import create_app
 
-def main():
-    """Main entry point"""
-    app = create_app()
-    return app
 
 if __name__ == "__main__":
-    app = main()
-    app.run()
+    # Local stdio entry point only (Claude Desktop / CLI); not for Docker/web deployment.
+    app = create_app()
+    app.run()

mne-python/mcp_output/mcp_plugin/mcp_service.py

@@ -1,215 +1,319 @@
-import os
+from __future__ import annotations
+
+from pathlib import Path
 import sys
-from typing import Optional, Dict, Any, List
 
-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)
+try:
+    from fastmcp import FastMCP
+except Exception:
+    FastMCP = None
 
-from fastmcp import FastMCP
+try:
+    import importlib
+except Exception:
+    importlib = None
 
-mcp = FastMCP("mne_python_service")
+try:
+    np = importlib.import_module("numpy") if importlib is not None else None
+except Exception:
+    np = None
 
+try:
+    mne = importlib.import_module("mne") if importlib is not None else None
+except Exception:
+    mne = None
 
-def _safe_import_mne():
-    try:
-        import mne  # type: ignore
-        return True, mne, None
-    except Exception as exc:
-        return False, None, str(exc)
+try:
+    mne_channels = importlib.import_module("mne.channels") if importlib is not None else None
+except Exception:
+    mne_channels = None
 
+try:
+    mne_io = importlib.import_module("mne.io") if importlib is not None else None
+except Exception:
+    mne_io = None
 
-@mcp.tool(name="mne_get_version", description="Get installed MNE version information.")
-def mne_get_version() -> Dict[str, Any]:
-    """
-    Return MNE package version.
+try:
+    mne_time_frequency = importlib.import_module("mne.time_frequency") if importlib is not None else None
+    psd_array_welch = (
+        getattr(mne_time_frequency, "psd_array_welch", None)
+        if mne_time_frequency is not None
+        else None
+    )
+except Exception:
+    psd_array_welch = None
 
-    Returns:
-        Dict with:
-        - success: bool indicating operation status
-        - result: version string when successful
-        - error: error string when failed
-    """
-    ok, mne_mod, err = _safe_import_mne()
-    if not ok:
-        return {"success": False, "result": None, "error": err}
-    try:
-        version = getattr(mne_mod, "__version__", "unknown")
-        return {"success": True, "result": version, "error": None}
-    except Exception as exc:
-        return {"success": False, "result": None, "error": str(exc)}
+from adapter import Adapter
 
 
-@mcp.tool(name="mne_get_config", description="Read MNE configuration value by key.")
-def mne_get_config(key: str, default: Optional[str] = None) -> Dict[str, Any]:
-    """
-    Read a single MNE config key.
+CURRENT_DIR = Path(__file__).resolve().parent
+SOURCE_DIR = CURRENT_DIR.parent.parent / "source"
+if SOURCE_DIR.exists():
+    source_str = str(SOURCE_DIR)
+    if source_str not in sys.path:
+        sys.path.insert(0, source_str)
 
-    Parameters:
-        key: Configuration key to lookup.
-        default: Optional fallback if key is not found.
 
-    Returns:
-        Dict with:
-        - success: bool
-        - result: config value
-        - error: error message when failed
-    """
-    ok, mne_mod, err = _safe_import_mne()
-    if not ok:
-        return {"success": False, "result": None, "error": err}
-    try:
-        value = mne_mod.get_config(key=key, default=default)
-        return {"success": True, "result": value, "error": None}
-    except Exception as exc:
-        return {"success": False, "result": None, "error": str(exc)}
+class _FallbackMCP:
+    def __init__(self) -> None:
+        self.tools: list[object] = []
 
+    def tool(self, name: str, description: str):
+        def decorator(func):
+            setattr(func, "name", name)
+            setattr(func, "description", description)
+            self.tools.append(func)
+            return func
 
-@mcp.tool(name="mne_set_config", description="Set an MNE configuration key to a value.")
-def mne_set_config(key: str, value: str, set_env: bool = False) -> Dict[str, Any]:
-    """
-    Set a single MNE config key.
-
-    Parameters:
-        key: Configuration key.
-        value: Configuration value.
-        set_env: If True, also set environment variable for current process.
-
-    Returns:
-        Dict with:
-        - success: bool
-        - result: True when set successfully
-        - error: error message when failed
-    """
-    ok, mne_mod, err = _safe_import_mne()
-    if not ok:
-        return {"success": False, "result": None, "error": err}
-    try:
-        mne_mod.set_config(key=key, value=value, set_env=set_env)
-        return {"success": True, "result": True, "error": None}
-    except Exception as exc:
-        return {"success": False, "result": None, "error": str(exc)}
+        return decorator
+
+    def run(self, *args, **kwargs) -> None:
+        raise RuntimeError("fastmcp is not available")
+
+
+mcp = FastMCP("mne_python_mcp_service") if FastMCP is not None else _FallbackMCP()
+adapter = Adapter(root_package="mne")
+
+
+def _ok(result: object) -> dict[str, object]:
+    return {"success": True, "result": result, "error": None}
+
+
+def _err(message: str) -> dict[str, object]:
+    return {"success": False, "result": None, "error": message}
 
 
-@mcp.tool(name="mne_create_info", description="Create an MNE Info object from channels and sampling frequency.")
-def mne_create_info(ch_names: List[str], sfreq: float, ch_types: Optional[List[str]] = None) -> Dict[str, Any]:
+@mcp.tool(name="health_check", description="Report dependency and adapter health status")
+def health_check() -> dict[str, object]:
+    """Return availability status for key MCP dependencies and mne imports."""
+    deps = {
+        "fastmcp": FastMCP is not None,
+        "numpy": np is not None,
+        "mne": mne is not None,
+        "mne.channels": mne_channels is not None,
+        "mne.io": mne_io is not None,
+        "mne.time_frequency.psd_array_welch": psd_array_welch is not None,
+    }
+    return _ok({"dependencies": deps, "adapter": adapter.health()})
+
+
+@mcp.tool(name="get_mne_version", description="Get installed MNE version information")
+def get_mne_version() -> dict[str, object]:
+    """Return installed MNE version and module file path."""
+    if mne is None:
+        return _err("mne is not available")
+    return _ok({"version": mne.__version__, "module": getattr(mne, "__file__", None)})
+
+
+@mcp.tool(name="list_builtin_montages", description="List available built-in electrode montages")
+def list_builtin_montages() -> dict[str, object]:
+    """Return names of built-in montages from mne.channels.
+
+    Returns
+    -------
+    dict
+        Standard MCP response with montage names and count.
     """
-    Create a lightweight channel metadata structure.
-
-    Parameters:
-        ch_names: List of channel names.
-        sfreq: Sampling frequency in Hz.
-        ch_types: Optional list of channel types aligned with ch_names.
-
-    Returns:
-        Dict with:
-        - success: bool
-        - result: serializable summary of created Info
-        - error: error message when failed
+    if mne_channels is None:
+        return _err("mne.channels is not available")
+    try:
+        names = mne_channels.get_builtin_montages(descriptions=False)
+        return _ok({"count": len(names), "montages": names})
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="create_info", description="Create and summarize an MNE Info object")
+def create_info(
+    ch_names: list[str],
+    sfreq: float,
+    ch_types: str = "eeg",
+) -> dict[str, object]:
+    """Create an MNE Info object.
+
+    Parameters
+    ----------
+    ch_names : list[str]
+        Channel names.
+    sfreq : float
+        Sampling frequency in Hz.
+    ch_types : str
+        Channel type applied to all channels.
     """
-    ok, mne_mod, err = _safe_import_mne()
-    if not ok:
-        return {"success": False, "result": None, "error": err}
+    if mne is None:
+        return _err("mne is not available")
     try:
-        info = mne_mod.create_info(ch_names=ch_names, sfreq=sfreq, ch_types=ch_types)
-        result = {
+        info = mne.create_info(ch_names=ch_names, sfreq=sfreq, ch_types=ch_types)
+        summary = {
             "nchan": int(info["nchan"]),
             "sfreq": float(info["sfreq"]),
             "ch_names": list(info["ch_names"]),
-            "highpass": float(info.get("highpass", 0.0) or 0.0),
-            "lowpass": float(info.get("lowpass", 0.0) or 0.0),
+            "ch_types": list(info.get_channel_types()),
         }
-        return {"success": True, "result": result, "error": None}
+        return _ok(summary)
     except Exception as exc:
-        return {"success": False, "result": None, "error": str(exc)}
-
-
-@mcp.tool(name="mne_compute_events", description="Detect events from a stim channel in a raw FIF file.")
-def mne_compute_events(
-    raw_fif_path: str,
-    stim_channel: Optional[str] = None,
-    shortest_event: int = 1,
-    min_duration: float = 0.0,
-) -> Dict[str, Any]:
+        return _err(str(exc))
+
+
+@mcp.tool(name="create_raw_array", description="Create RawArray metadata from matrix-like data")
+def create_raw_array(
+    data: list[list[float]],
+    sfreq: float,
+    ch_names: list[str],
+    ch_types: str = "eeg",
+) -> dict[str, object]:
+    """Create an in-memory RawArray and return metadata.
+
+    Parameters
+    ----------
+    data : list[list[float]]
+        Data shaped as [n_channels, n_times].
+    sfreq : float
+        Sampling frequency in Hz.
+    ch_names : list[str]
+        Channel names, length must equal number of rows in data.
+    ch_types : str
+        Channel type applied to all channels.
+    """
+    if np is None or mne is None or mne_io is None:
+        return _err("numpy or mne.io is not available")
+    try:
+        arr = np.asarray(data, dtype=float)
+        if arr.ndim != 2:
+            return _err("data must be a 2D array [n_channels, n_times]")
+        if arr.shape[0] != len(ch_names):
+            return _err("len(ch_names) must equal n_channels")
+        info = mne.create_info(ch_names=ch_names, sfreq=sfreq, ch_types=ch_types)
+        raw = mne_io.RawArray(arr, info, verbose=False)
+        return _ok(
+            {
+                "n_channels": int(raw.info["nchan"]),
+                "n_times": int(raw.n_times),
+                "duration_sec": float(raw.n_times / raw.info["sfreq"]),
+                "sfreq": float(raw.info["sfreq"]),
+            }
+        )
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="filter_data_array", description="Band-pass or high/low-pass filter array data")
+def filter_data_array(
+    data: list[list[float]],
+    sfreq: float,
+    l_freq: float | None,
+    h_freq: float | None,
+) -> dict[str, object]:
+    """Filter array data with MNE filtering utilities.
+
+    Parameters
+    ----------
+    data : list[list[float]]
+        Input array shaped as [n_channels, n_times].
+    sfreq : float
+        Sampling frequency in Hz.
+    l_freq : float | None
+        Lower passband edge in Hz. Use None for low-pass only.
+    h_freq : float | None
+        Upper passband edge in Hz. Use None for high-pass only.
     """
-    Load raw FIF and compute events from stimulation channel.
-
-    Parameters:
-        raw_fif_path: Path to a readable raw FIF file.
-        stim_channel: Optional stim channel name. If None, MNE default detection is used.
-        shortest_event: Minimum number of samples for an event.
-        min_duration: Minimum event duration in seconds.
-
-    Returns:
-        Dict with:
-        - success: bool
-        - result: event count and preview rows
-        - error: error message when failed
+    if np is None or mne is None:
+        return _err("numpy or mne is not available")
+    try:
+        arr = np.asarray(data, dtype=float)
+        if arr.ndim != 2:
+            return _err("data must be a 2D array [n_channels, n_times]")
+        filtered = mne.filter.filter_data(
+            data=arr,
+            sfreq=sfreq,
+            l_freq=l_freq,
+            h_freq=h_freq,
+            verbose=False,
+        )
+        return _ok(
+            {
+                "shape": [int(filtered.shape[0]), int(filtered.shape[1])],
+                "mean": float(np.mean(filtered)),
+                "std": float(np.std(filtered)),
+                "data": filtered.tolist(),
+            }
+        )
+    except Exception as exc:
+        return _err(str(exc))
+
+
+@mcp.tool(name="compute_psd_welch", description="Compute PSD using Welch on array data")
+def compute_psd_welch(
+    data: list[list[float]],
+    sfreq: float,
+    fmin: float = 0.0,
+    fmax: float = 120.0,
+    n_fft: int = 256,
+) -> dict[str, object]:
+    """Compute power spectral density with Welch method.
+
+    Parameters
+    ----------
+    data : list[list[float]]
+        Input array shaped as [n_channels, n_times].
+    sfreq : float
+        Sampling frequency in Hz.
+    fmin : float
+        Minimum frequency of interest.
+    fmax : float
+        Maximum frequency of interest.
+    n_fft : int
+        FFT length.
     """
-    ok, mne_mod, err = _safe_import_mne()
-    if not ok:
-        return {"success": False, "result": None, "error": err}
+    if np is None or psd_array_welch is None:
+        return _err("numpy or psd_array_welch is not available")
     try:
-        raw = mne_mod.io.read_raw_fif(raw_fif_path, preload=False, verbose=False)
-        events = mne_mod.find_events(
-            raw,
-            stim_channel=stim_channel,
-            shortest_event=shortest_event,
-            min_duration=min_duration,
+        arr = np.asarray(data, dtype=float)
+        if arr.ndim != 2:
+            return _err("data must be a 2D array [n_channels, n_times]")
+        psds, freqs = psd_array_welch(
+            x=arr,
+            sfreq=sfreq,
+            fmin=fmin,
+            fmax=fmax,
+            n_fft=n_fft,
             verbose=False,
         )
-        preview = events[:20].tolist() if len(events) > 0 else []
-        return {
-            "success": True,
-            "result": {"count": int(len(events)), "preview": preview},
-            "error": None,
-        }
+        mean_psd = np.mean(psds, axis=0)
+        return _ok(
+            {
+                "n_channels": int(arr.shape[0]),
+                "n_freqs": int(len(freqs)),
+                "freqs": freqs.tolist(),
+                "mean_psd": mean_psd.tolist(),
+            }
+        )
     except Exception as exc:
-        return {"success": False, "result": None, "error": str(exc)}
+        return _err(str(exc))
 
 
-@mcp.tool(name="mne_estimate_rank", description="Estimate data rank from an epochs or raw FIF file.")
-def mne_estimate_rank(fif_path: str, tol: Optional[float] = None) -> Dict[str, Any]:
-    """
-    Estimate numerical rank from MNE data object loaded from FIF.
-
-    Parameters:
-        fif_path: Path to raw or epochs FIF file.
-        tol: Optional tolerance passed to rank estimator.
+@mcp.tool(name="list_loaded_modules", description="List modules loaded by adapter")
+def list_loaded_modules(limit: int = 100) -> dict[str, object]:
+    """List loaded and failed modules discovered by the Adapter.
 
-    Returns:
-        Dict with:
-        - success: bool
-        - result: estimated rank dictionary or scalar
-        - error: error message when failed
+    Parameters
+    ----------
+    limit : int
+        Maximum number of loaded modules returned.
     """
-    ok, mne_mod, err = _safe_import_mne()
-    if not ok:
-        return {"success": False, "result": None, "error": err}
     try:
-        result_obj: Any
-        try:
-            raw = mne_mod.io.read_raw_fif(fif_path, preload=False, verbose=False)
-            result_obj = mne_mod.compute_rank(raw, tol=tol, verbose=False)
-        except Exception:
-            epochs = mne_mod.read_epochs(fif_path, preload=False, verbose=False)
-            result_obj = mne_mod.compute_rank(epochs, tol=tol, verbose=False)
-
-        if isinstance(result_obj, dict):
-            serializable = {str(k): int(v) for k, v in result_obj.items()}
-        else:
-            serializable = int(result_obj)
-        return {"success": True, "result": serializable, "error": None}
+        modules = adapter.list_modules(include_failed=True)
+        loaded = modules.get("loaded_modules", [])
+        if isinstance(loaded, list):
+            modules["loaded_modules"] = loaded[: max(0, limit)]
+        return _ok(modules)
     except Exception as exc:
-        return {"success": False, "result": None, "error": str(exc)}
+        return _err(str(exc))
 
 
-def create_app() -> FastMCP:
+def create_app():
     return mcp
 
 
 if __name__ == "__main__":
-    mcp.run()
+    mcp.run()

mne-python/mcp_output/requirements.txt

@@ -1,16 +1,4 @@
 fastmcp
-fastapi
-uvicorn[standard]
-pydantic>=2.0.0
-decorator >= 5.1
-jinja2 >= 3.1
-lazy_loader >= 0.3
-matplotlib >= 3.8
-numpy >= 1.26, < 3
-packaging
-pooch >= 1.5
-scipy >= 1.12
-tqdm >= 4.66
-nest-asyncio2
-pymef
-pyobjc-framework-Cocoa >=5.2.0;platform_system=='Darwin'
+mne
+numpy
+scipy

mne-python/mcp_output/start_mcp.py

@@ -1,30 +1,37 @@
+from __future__ import annotations
 
-"""
-MCP Service Startup Entry
-"""
-import sys
 import os
+from pathlib import Path
+import sys
+
 
-project_root = os.path.dirname(os.path.abspath(__file__))
-mcp_plugin_dir = os.path.join(project_root, "mcp_plugin")
-if mcp_plugin_dir not in sys.path:
-    sys.path.insert(0, mcp_plugin_dir)
+CURRENT_DIR = Path(__file__).resolve().parent
+PLUGIN_DIR = CURRENT_DIR / "mcp_plugin"
+plugin_str = str(PLUGIN_DIR)
+if plugin_str not in sys.path:
+    sys.path.insert(0, plugin_str)
 
 from mcp_service import create_app
 
-def main():
-    """Start FastMCP service"""
+
+def main() -> None:
+    transport = os.getenv("MCP_TRANSPORT", "stdio").strip().lower()
+    port = int(os.getenv("MCP_PORT", "8000"))
     app = create_app()
-    # Use environment variable to configure port, default 8000
-    port = int(os.environ.get("MCP_PORT", "8000"))
-    
-    # Choose transport mode based on environment variable
-    transport = os.environ.get("MCP_TRANSPORT", "stdio")
-    if transport == "http":
-        app.run(transport="http", host="0.0.0.0", port=port)
-    else:
-        # Default to STDIO mode
+
+    if transport == "stdio":
         app.run()
+        return
+
+    if transport == "http":
+        try:
+            app.run(transport="http", host="0.0.0.0", port=port)
+        except TypeError:
+            app.run(transport="http", port=port)
+        return
+
+    raise ValueError(f"Unsupported MCP_TRANSPORT: {transport}")
+
 
 if __name__ == "__main__":
     main()

port.json

@@ -1,5 +1 @@
-{
-  "repo": "mne-python",
-  "port": 7938,
-  "timestamp": 1773462397
-}
+{"port": 7860}

requirements.txt

@@ -1,16 +1,6 @@
 fastmcp
+mne
+numpy
+scipy
 fastapi
-uvicorn[standard]
-pydantic>=2.0.0
-decorator >= 5.1
-jinja2 >= 3.1
-lazy_loader >= 0.3
-matplotlib >= 3.8
-numpy >= 1.26, < 3
-packaging
-pooch >= 1.5
-scipy >= 1.12
-tqdm >= 4.66
-nest-asyncio2
-pymef
-pyobjc-framework-Cocoa >=5.2.0;platform_system=='Darwin'
+uvicorn

run_docker.ps1

@@ -1,26 +1,8 @@
-cd $PSScriptRoot
-$ErrorActionPreference = "Stop"
-$entryName = if ($env:MCP_ENTRY_NAME) { $env:MCP_ENTRY_NAME } else { "mne-python" }
-$entryUrl  = if ($env:MCP_ENTRY_URL)  { $env:MCP_ENTRY_URL  } else { "http://localhost:7938/mcp" }
-$imageName = if ($env:MCP_IMAGE_NAME) { $env:MCP_IMAGE_NAME } else { "mne-python-mcp" }
-$mcpDir = Join-Path $env:USERPROFILE ".cursor"
-$mcpPath = Join-Path $mcpDir "mcp.json"
-if (!(Test-Path $mcpDir)) { New-Item -ItemType Directory -Path $mcpDir | Out-Null }
-$config = @{}
-if (Test-Path $mcpPath) {
-  try { $config = Get-Content $mcpPath -Raw | ConvertFrom-Json } catch { $config = @{} }
-}
-$serversOrdered = [ordered]@{}
-if ($config -and ($config.PSObject.Properties.Name -contains "mcpServers") -and $config.mcpServers) {
-  $existing = $config.mcpServers
-  if ($existing -is [pscustomobject]) {
-    foreach ($p in $existing.PSObject.Properties) { if ($p.Name -ne $entryName) { $serversOrdered[$p.Name] = $p.Value } }
-  } elseif ($existing -is [System.Collections.IDictionary]) {
-    foreach ($k in $existing.Keys) { if ($k -ne $entryName) { $serversOrdered[$k] = $existing[$k] } }
-  }
-}
-$serversOrdered[$entryName] = @{ url = $entryUrl }
-$config = @{ mcpServers = $serversOrdered }
-$config | ConvertTo-Json -Depth 10 | Set-Content -Path $mcpPath -Encoding UTF8
+$port = (Get-Content -Raw -Path "port.json" | ConvertFrom-Json).port
+$imageName = "mne-python-mcp"
+
+Write-Host "Building Docker image: $imageName"
 docker build -t $imageName .
-docker run --rm -p 7938:7860 $imageName
+
+Write-Host "Running container on port $port"
+docker run --rm -it -p "${port}:${port}" -e MCP_PORT=$port $imageName

run_docker.sh

@@ -1,75 +1,12 @@
 #!/usr/bin/env bash
 set -euo pipefail
-cd "$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
-mcp_entry_name="${MCP_ENTRY_NAME:-mne-python}"
-mcp_entry_url="${MCP_ENTRY_URL:-http://localhost:7938/mcp}"
-mcp_dir="${HOME}/.cursor"
-mcp_path="${mcp_dir}/mcp.json"
-mkdir -p "${mcp_dir}"
-if command -v python3 >/dev/null 2>&1; then
-python3 - "${mcp_path}" "${mcp_entry_name}" "${mcp_entry_url}" <<'PY'
-import json, os, sys
-path, name, url = sys.argv[1:4]
-cfg = {"mcpServers": {}}
-if os.path.exists(path):
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            cfg = json.load(f)
-    except Exception:
-        cfg = {"mcpServers": {}}
-if not isinstance(cfg, dict):
-    cfg = {"mcpServers": {}}
-servers = cfg.get("mcpServers")
-if not isinstance(servers, dict):
-    servers = {}
-ordered = {}
-for k, v in servers.items():
-    if k != name:
-        ordered[k] = v
-ordered[name] = {"url": url}
-cfg = {"mcpServers": ordered}
-with open(path, "w", encoding="utf-8") as f:
-    json.dump(cfg, f, indent=2, ensure_ascii=False)
-PY
-elif command -v python >/dev/null 2>&1; then
-python - "${mcp_path}" "${mcp_entry_name}" "${mcp_entry_url}" <<'PY'
-import json, os, sys
-path, name, url = sys.argv[1:4]
-cfg = {"mcpServers": {}}
-if os.path.exists(path):
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            cfg = json.load(f)
-    except Exception:
-        cfg = {"mcpServers": {}}
-if not isinstance(cfg, dict):
-    cfg = {"mcpServers": {}}
-servers = cfg.get("mcpServers")
-if not isinstance(servers, dict):
-    servers = {}
-ordered = {}
-for k, v in servers.items():
-    if k != name:
-        ordered[k] = v
-ordered[name] = {"url": url}
-cfg = {"mcpServers": ordered}
-with open(path, "w", encoding="utf-8") as f:
-    json.dump(cfg, f, indent=2, ensure_ascii=False)
-PY
-elif command -v jq >/dev/null 2>&1; then
-  name="${mcp_entry_name}"; url="${mcp_entry_url}"
-  if [ -f "${mcp_path}" ]; then
-    tmp="$(mktemp)"
-    jq --arg name "$name" --arg url "$url" '
-      .mcpServers = (.mcpServers // {})
-      | .mcpServers as $s
-      | ($s | with_entries(select(.key != $name))) as $base
-      | .mcpServers = ($base + {($name): {"url": $url}})
-    ' "${mcp_path}" > "${tmp}" && mv "${tmp}" "${mcp_path}"
-  else
-    printf '{ "mcpServers": { "%s": { "url": "%s" } } }
-' "$name" "$url" > "${mcp_path}"
-  fi
-fi
-docker build -t mne-python-mcp .
-docker run --rm -p 7938:7860 mne-python-mcp
+
+PORT=$(python3 -c "import json; print(json.load(open('port.json'))['port'])")
+IMAGE_NAME="mne-python-mcp"
+
+
+echo "Building Docker image: ${IMAGE_NAME}"
+docker build -t "${IMAGE_NAME}" .
+
+echo "Running container on port ${PORT}"
+docker run --rm -it -p "${PORT}:${PORT}" -e MCP_PORT="${PORT}" "${IMAGE_NAME}"