Update nilearn/mcp_output/mcp_plugin/mcp_service.py

nilearn/mcp_output/mcp_plugin/mcp_service.py

@@ -1,97 +1,274 @@
 import os
 import sys
-
-# Path settings to include the local source directory
-source_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "source")
-if source_path not in sys.path:
-    sys.path.insert(0, source_path)
+from typing import List, Optional
 
 from fastmcp import FastMCP
+import numpy as np
 
-# Import core modules from the local source directory
-from nilearn.image import image
-from nilearn.glm import first_level
-from nilearn.decoding import decoder
-from nilearn.plotting import plotting
+# Import nilearn modules (from PyPI installed package)
+from nilearn import image, datasets, plotting
+from nilearn.maskers import NiftiMasker
+from nilearn.connectome import ConnectivityMeasure
+from nilearn.decoding import Decoder
+from nilearn.glm.first_level import FirstLevelModel, make_first_level_design_matrix
 
 # Create the FastMCP service application
 mcp = FastMCP("nilearn_service")
 
-@mcp.tool(name="load_image", description="Load and process a neuroimaging file")
-def load_image(file_path: str) -> dict:
+
+@mcp.tool(name="get_nilearn_version", description="Get nilearn library version")
+def get_nilearn_version() -> dict:
+    """
+    Get the nilearn library version.
+
+    Returns:
+    - dict: Version information.
+    """
+    try:
+        import nilearn
+        return {
+            "success": True,
+            "result": {
+                "version": nilearn.__version__,
+                "description": "Machine learning for neuroimaging in Python"
+            },
+            "error": None
+        }
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="list_available_atlases", description="List available brain atlases in nilearn")
+def list_available_atlases() -> dict:
     """
-    Load and process a neuroimaging file.
+    List available brain atlases in nilearn datasets.
+
+    Returns:
+    - dict: Available atlases and their descriptions.
+    """
+    try:
+        atlases = {
+            "aal": "Automated Anatomical Labeling atlas",
+            "destrieux": "Destrieux cortical atlas (aparc.a2009s)",
+            "harvard_oxford": "Harvard-Oxford cortical/subcortical atlas",
+            "msdl": "MSDL probabilistic atlas (39 regions)",
+            "pauli_2017": "Subcortical atlas (Pauli et al. 2017)",
+            "schaefer_2018": "Schaefer cortical parcellation",
+            "yeo_2011": "Yeo intrinsic functional networks (7 or 17 networks)",
+            "difumo": "DiFuMo atlas (64, 128, 256, 512, or 1024 dimensions)",
+            "talairach": "Talairach atlas"
+        }
+        return {"success": True, "result": atlases, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="fetch_atlas", description="Fetch a brain atlas from nilearn datasets")
+def fetch_atlas(atlas_name: str, n_rois: int = None) -> dict:
+    """
+    Fetch a brain atlas from nilearn datasets.
 
     Parameters:
-    - file_path (str): Path to the neuroimaging file.
+    - atlas_name: Name of the atlas ('aal', 'msdl', 'schaefer_2018', 'yeo_2011', etc.)
+    - n_rois: Number of ROIs (for atlases that support it)
 
     Returns:
-    - dict: A dictionary containing success, result, or error.
+    - dict: Atlas information including number of regions.
     """
     try:
-        img = image.load_img(file_path)
-        return {"success": True, "result": img}
+        if atlas_name == "aal":
+            atlas = datasets.fetch_atlas_aal()
+            return {
+                "success": True,
+                "result": {
+                    "atlas_name": "AAL",
+                    "n_regions": len(atlas.labels),
+                    "labels": atlas.labels[:10],  # First 10 labels as sample
+                    "description": "Automated Anatomical Labeling atlas"
+                },
+                "error": None
+            }
+        elif atlas_name == "msdl":
+            atlas = datasets.fetch_atlas_msdl()
+            return {
+                "success": True,
+                "result": {
+                    "atlas_name": "MSDL",
+                    "n_regions": len(atlas.labels),
+                    "labels": atlas.labels,
+                    "description": "Multi-Subject Dictionary Learning atlas"
+                },
+                "error": None
+            }
+        elif atlas_name == "schaefer_2018":
+            n_rois = n_rois or 100
+            atlas = datasets.fetch_atlas_schaefer_2018(n_rois=n_rois)
+            return {
+                "success": True,
+                "result": {
+                    "atlas_name": f"Schaefer {n_rois}",
+                    "n_regions": n_rois,
+                    "labels": list(atlas.labels[:10]),
+                    "description": "Schaefer cortical parcellation"
+                },
+                "error": None
+            }
+        else:
+            return {"success": False, "result": None, "error": f"Unknown atlas: {atlas_name}"}
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-@mcp.tool(name="run_glm", description="Run a General Linear Model on fMRI data")
-def run_glm(data: str, design_matrix: str) -> dict:
+@mcp.tool(name="compute_connectivity_matrix", description="Compute functional connectivity matrix from time series")
+def compute_connectivity_matrix(
+    time_series: List[List[float]],
+    connectivity_kind: str = "correlation"
+) -> dict:
     """
-    Run a General Linear Model on fMRI data.
+    Compute functional connectivity matrix from ROI time series.
 
     Parameters:
-    - data (str): Path to the fMRI data.
-    - design_matrix (str): Path to the design matrix file.
+    - time_series: 2D list of shape (n_timepoints, n_regions)
+    - connectivity_kind: Type of connectivity ('correlation', 'partial correlation', 'covariance')
 
     Returns:
-    - dict: A dictionary containing success, result, or error.
+    - dict: Connectivity matrix and statistics.
     """
     try:
-        fmri_img = image.load_img(data)
-        design = first_level.make_first_level_design_matrix(fmri_img, design_matrix)
-        model = first_level.FirstLevelModel()
-        model = model.fit(fmri_img, design_matrices=design)
-        return {"success": True, "result": model}
+        ts_array = np.array(time_series)
+        
+        conn_measure = ConnectivityMeasure(kind=connectivity_kind)
+        connectivity = conn_measure.fit_transform([ts_array])[0]
+        
+        # Get lower triangle values (excluding diagonal)
+        lower_tri = connectivity[np.tril_indices_from(connectivity, k=-1)]
+        
+        return {
+            "success": True,
+            "result": {
+                "shape": list(connectivity.shape),
+                "connectivity_kind": connectivity_kind,
+                "mean_connectivity": float(np.mean(lower_tri)),
+                "std_connectivity": float(np.std(lower_tri)),
+                "min_connectivity": float(np.min(lower_tri)),
+                "max_connectivity": float(np.max(lower_tri)),
+                "matrix_preview": connectivity[:5, :5].tolist() if connectivity.shape[0] >= 5 else connectivity.tolist()
+            },
+            "error": None
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="decode_brain_data", description="Decode brain data using a machine learning model")
-def decode_brain_data(data: str, labels: str) -> dict:
+
+@mcp.tool(name="list_decoding_estimators", description="List available decoding estimators")
+def list_decoding_estimators() -> dict:
+    """
+    List available machine learning estimators for brain decoding.
+
+    Returns:
+    - dict: Available estimators and their descriptions.
+    """
+    try:
+        estimators = {
+            "svc": "Support Vector Classification (linear kernel)",
+            "svc_l1": "SVC with L1 penalty for feature selection",
+            "svc_l2": "SVC with L2 penalty",
+            "logistic": "Logistic Regression",
+            "logistic_l1": "Logistic Regression with L1 penalty",
+            "logistic_l2": "Logistic Regression with L2 penalty",
+            "ridge": "Ridge Regression (for continuous targets)",
+            "ridge_classifier": "Ridge Classifier"
+        }
+        return {"success": True, "result": estimators, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="resample_image", description="Resample a neuroimaging volume to new resolution")
+def resample_image(
+    target_affine_diag: List[float] = None,
+    target_shape: List[int] = None,
+    interpolation: str = "continuous"
+) -> dict:
     """
-    Decode brain data using a machine learning model.
+    Get parameters for resampling neuroimaging data.
 
     Parameters:
-    - data (str): Path to the brain data.
-    - labels (str): Path to the labels file.
+    - target_affine_diag: Diagonal elements of target affine (voxel sizes)
+    - target_shape: Target shape [x, y, z]
+    - interpolation: 'continuous', 'linear', or 'nearest'
 
     Returns:
-    - dict: A dictionary containing success, result, or error.
+    - dict: Resampling configuration.
     """
     try:
-        brain_data = image.load_img(data)
-        model = decoder.Decoder()
-        model.fit(brain_data, labels)
-        return {"success": True, "result": model}
+        if target_affine_diag is None:
+            target_affine_diag = [3.0, 3.0, 3.0]  # 3mm isotropic
+        if target_shape is None:
+            target_shape = [61, 73, 61]  # MNI152 standard shape at 3mm
+        
+        config = {
+            "target_affine_diagonal": target_affine_diag,
+            "target_shape": target_shape,
+            "interpolation": interpolation,
+            "description": f"Resample to {target_affine_diag[0]}mm isotropic resolution"
+        }
+        return {"success": True, "result": config, "error": None}
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-@mcp.tool(name="plot_stat_map", description="Plot a statistical map")
-def plot_stat_map(stat_map: str, bg_img: str) -> dict:
+@mcp.tool(name="get_glm_contrast_types", description="List available GLM contrast types")
+def get_glm_contrast_types() -> dict:
     """
-    Plot a statistical map.
+    List available contrast types for first-level GLM analysis.
+
+    Returns:
+    - dict: Contrast types and descriptions.
+    """
+    try:
+        contrasts = {
+            "t": "T-contrast: tests linear combination of parameters",
+            "F": "F-contrast: tests multiple linear hypotheses simultaneously",
+            "effects": {
+                "condition_effect": "Main effect of a single condition",
+                "differential": "Difference between two conditions (A - B)",
+                "conjunction": "Common activation across conditions",
+                "interaction": "Interaction between factors"
+            },
+            "examples": {
+                "condition_a": "[1, 0, 0, ...]",
+                "condition_a_vs_b": "[1, -1, 0, ...]",
+                "any_condition": "[[1, 0, 0], [0, 1, 0], [0, 0, 1]]"
+            }
+        }
+        return {"success": True, "result": contrasts, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="smooth_image", description="Smooth a neuroimaging volume")
+def smooth_image(fwhm: float = 6.0) -> dict:
+    """
+    Get configuration for smoothing neuroimaging data.
 
     Parameters:
-    - stat_map (str): Path to the statistical map.
-    - bg_img (str): Path to the background image.
+    - fwhm: Full Width at Half Maximum in mm (default: 6.0)
 
     Returns:
-    - dict: A dictionary containing success, result, or error.
+    - dict: Smoothing configuration.
     """
     try:
-        display = plotting.plot_stat_map(stat_map, bg_img=bg_img)
-        return {"success": True, "result": display}
+        config = {
+            "fwhm": fwhm,
+            "description": f"Gaussian smoothing with {fwhm}mm FWHM kernel",
+            "function": "nilearn.image.smooth_img",
+            "usage": f"smoothed_img = smooth_img(img, fwhm={fwhm})"
+        }
+        return {"success": True, "result": config, "error": None}
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
 def create_app() -> FastMCP:
     """