Update NeuroKit/mcp_output/mcp_plugin/mcp_service.py

NeuroKit/mcp_output/mcp_plugin/mcp_service.py

@@ -1,5 +1,6 @@
 import os
 import sys
+from typing import List, Optional, Dict, Any, Tuple
 
 # 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")
@@ -7,64 +8,1263 @@ if source_path not in sys.path:
     sys.path.insert(0, source_path)
 
 from fastmcp import FastMCP
+import numpy as np
 import neurokit2 as nk
 
 # Create the FastMCP service application
 mcp = FastMCP("neurokit_service")
 
-@mcp.tool(name="ecg_process", description="Process ECG data")
-def ecg_process(ecg_data: list, sampling_rate: int) -> dict:
+
+# =====================================================
+# ECG Tools - 心电图工具
+# =====================================================
+
+@mcp.tool(name="ecg_simulate", description="Simulate an ECG/EKG signal.")
+def ecg_simulate(duration: int = 10, sampling_rate: int = 1000, 
+                 heart_rate: int = 70, noise: float = 0.01,
+                 method: str = "ecgsyn") -> dict:
+    """
+    Simulate an ECG/EKG signal.
+
+    Parameters:
+    - duration (int): Recording length in seconds (default: 10)
+    - sampling_rate (int): Sampling rate in Hz (default: 1000)
+    - heart_rate (int): Heart rate in beats per minute (default: 70)
+    - noise (float): Noise level (default: 0.01)
+    - method (str): Simulation method - "ecgsyn" or "simple" (default: "ecgsyn")
+
+    Returns:
+    - dict: Contains success, ECG signal array, and metadata
+    """
+    try:
+        ecg = nk.ecg_simulate(
+            duration=duration, 
+            sampling_rate=sampling_rate, 
+            heart_rate=heart_rate,
+            noise=noise,
+            method=method
+        )
+        return {
+            "success": True,
+            "signal": ecg.tolist(),
+            "duration": duration,
+            "sampling_rate": sampling_rate,
+            "heart_rate": heart_rate,
+            "num_samples": len(ecg)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="ecg_process", description="Process ECG data including cleaning, peak detection, and feature extraction.")
+def ecg_process(ecg_data: List[float], sampling_rate: int = 1000,
+                method: str = "neurokit") -> dict:
     """
     Process ECG data using NeuroKit2's ecg_process function.
 
     Parameters:
     - ecg_data: list of ECG signal values
-    - sampling_rate: int, the sampling rate of the ECG data
+    - sampling_rate: int, the sampling rate of the ECG data (default: 1000)
+    - method: str, peak detection method (default: "neurokit")
 
     Returns:
-    - dict: Contains success, result, and error fields
+    - dict: Contains processed data columns and R-peaks info
     """
     try:
-        processed_data, info = nk.ecg_process(ecg=ecg_data, sampling_rate=sampling_rate)
-        return {"success": True, "result": {"processed_data": processed_data, "info": info}, "error": None}
+        signals, info = nk.ecg_process(ecg=ecg_data, sampling_rate=sampling_rate, method=method)
+        return {
+            "success": True,
+            "columns": list(signals.columns),
+            "num_samples": len(signals),
+            "r_peaks": info["ECG_R_Peaks"].tolist() if hasattr(info["ECG_R_Peaks"], 'tolist') else list(info["ECG_R_Peaks"]),
+            "sampling_rate": info.get("sampling_rate", sampling_rate),
+            "heart_rate_mean": float(signals["ECG_Rate"].mean()) if "ECG_Rate" in signals.columns else None
+        }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "error": str(e)}
+
 
-@mcp.tool(name="rsp_process", description="Process respiration data")
-def rsp_process(rsp_data: list, sampling_rate: int) -> dict:
+@mcp.tool(name="ecg_clean", description="Clean an ECG signal by removing noise and artifacts.")
+def ecg_clean(ecg_data: List[float], sampling_rate: int = 1000,
+              method: str = "neurokit") -> dict:
     """
-    Process respiration data using NeuroKit2's rsp_process function.
+    Clean an ECG signal by removing noise and artifacts.
 
     Parameters:
-    - rsp_data: list of respiration signal values
-    - sampling_rate: int, the sampling rate of the respiration data
+    - ecg_data: list of ECG signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, cleaning method (default: "neurokit")
+        Options: "neurokit", "biosppy", "pantompkins", "hamilton", "elgendi", "engzeemod"
+
+    Returns:
+    - dict: Contains cleaned ECG signal
+    """
+    try:
+        cleaned = nk.ecg_clean(ecg=ecg_data, sampling_rate=sampling_rate, method=method)
+        return {
+            "success": True,
+            "signal": cleaned.tolist(),
+            "method": method,
+            "num_samples": len(cleaned)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="ecg_findpeaks", description="Find R-peaks in an ECG signal.")
+def ecg_findpeaks(ecg_data: List[float], sampling_rate: int = 1000,
+                  method: str = "neurokit") -> dict:
+    """
+    Find R-peaks in an ECG signal.
+
+    Parameters:
+    - ecg_data: list of ECG signal values (should be cleaned first)
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, peak detection method (default: "neurokit")
 
     Returns:
-    - dict: Contains success, result, and error fields
+    - dict: Contains R-peak indices and count
     """
     try:
-        processed_data, info = nk.rsp_process(rsp=rsp_data, sampling_rate=sampling_rate)
-        return {"success": True, "result": {"processed_data": processed_data, "info": info}, "error": None}
+        peaks, info = nk.ecg_peaks(ecg_cleaned=ecg_data, sampling_rate=sampling_rate, method=method)
+        r_peaks = info["ECG_R_Peaks"]
+        return {
+            "success": True,
+            "r_peaks": r_peaks.tolist() if hasattr(r_peaks, 'tolist') else list(r_peaks),
+            "num_peaks": len(r_peaks),
+            "method": method
+        }
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "error": str(e)}
+
 
-@mcp.tool(name="eda_process", description="Process EDA data")
-def eda_process(eda_data: list, sampling_rate: int) -> dict:
+@mcp.tool(name="ecg_quality", description="Assess the quality of an ECG signal.")
+def ecg_quality(ecg_data: List[float], sampling_rate: int = 1000,
+                method: str = "zhao2018", approach: str = "fuzzy") -> dict:
+    """
+    Assess the quality of an ECG signal.
+
+    Parameters:
+    - ecg_data: list of ECG signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, quality assessment method (default: "zhao2018")
+    - approach: str, approach for quality index (default: "fuzzy")
+
+    Returns:
+    - dict: Contains quality assessment results
+    """
+    try:
+        quality = nk.ecg_quality(ecg_cleaned=ecg_data, sampling_rate=sampling_rate, 
+                                  method=method, approach=approach)
+        return {
+            "success": True,
+            "quality": quality.tolist() if hasattr(quality, 'tolist') else list(quality),
+            "mean_quality": float(np.mean(quality)),
+            "method": method
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+# =====================================================
+# EDA Tools - 皮肤电活动工具
+# =====================================================
+
+@mcp.tool(name="eda_simulate", description="Simulate an Electrodermal Activity (EDA/GSR) signal.")
+def eda_simulate(duration: int = 120, sampling_rate: int = 1000,
+                 scr_number: int = 5, noise: float = 0.01,
+                 drift: float = 0.01) -> dict:
+    """
+    Simulate an EDA/GSR signal.
+
+    Parameters:
+    - duration (int): Recording length in seconds (default: 120)
+    - sampling_rate (int): Sampling rate in Hz (default: 1000)
+    - scr_number (int): Number of skin conductance responses (default: 5)
+    - noise (float): Noise level (default: 0.01)
+    - drift (float): Baseline drift level (default: 0.01)
+
+    Returns:
+    - dict: Contains simulated EDA signal
+    """
+    try:
+        eda = nk.eda_simulate(
+            duration=duration,
+            sampling_rate=sampling_rate,
+            scr_number=scr_number,
+            noise=noise,
+            drift=drift
+        )
+        return {
+            "success": True,
+            "signal": eda.tolist(),
+            "duration": duration,
+            "sampling_rate": sampling_rate,
+            "num_samples": len(eda)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="eda_process", description="Process EDA data including cleaning and decomposition.")
+def eda_process(eda_data: List[float], sampling_rate: int = 1000,
+                method: str = "neurokit") -> dict:
     """
     Process EDA data using NeuroKit2's eda_process function.
 
     Parameters:
     - eda_data: list of EDA signal values
-    - sampling_rate: int, the sampling rate of the EDA data
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, processing method (default: "neurokit")
+
+    Returns:
+    - dict: Contains processed data columns and SCR peaks info
+    """
+    try:
+        signals, info = nk.eda_process(eda=eda_data, sampling_rate=sampling_rate, method=method)
+        return {
+            "success": True,
+            "columns": list(signals.columns),
+            "num_samples": len(signals),
+            "scr_peaks": info.get("SCR_Peaks", []),
+            "scr_onsets": info.get("SCR_Onsets", [])
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="eda_clean", description="Clean an EDA signal.")
+def eda_clean(eda_data: List[float], sampling_rate: int = 1000,
+              method: str = "neurokit") -> dict:
+    """
+    Clean an EDA signal.
+
+    Parameters:
+    - eda_data: list of EDA signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, cleaning method (default: "neurokit")
+
+    Returns:
+    - dict: Contains cleaned EDA signal
+    """
+    try:
+        cleaned = nk.eda_clean(eda=eda_data, sampling_rate=sampling_rate, method=method)
+        return {
+            "success": True,
+            "signal": cleaned.tolist(),
+            "method": method,
+            "num_samples": len(cleaned)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="eda_phasic", description="Decompose EDA signal into phasic and tonic components.")
+def eda_phasic(eda_data: List[float], sampling_rate: int = 1000,
+               method: str = "highpass") -> dict:
+    """
+    Decompose EDA signal into phasic (SCR) and tonic (SCL) components.
+
+    Parameters:
+    - eda_data: list of cleaned EDA signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, decomposition method (default: "highpass")
+        Options: "highpass", "cvxeda", "smoothmedian", "sparse"
+
+    Returns:
+    - dict: Contains phasic and tonic components
+    """
+    try:
+        decomposed = nk.eda_phasic(eda_cleaned=eda_data, sampling_rate=sampling_rate, method=method)
+        return {
+            "success": True,
+            "phasic": decomposed["EDA_Phasic"].tolist(),
+            "tonic": decomposed["EDA_Tonic"].tolist(),
+            "method": method,
+            "num_samples": len(decomposed)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+# =====================================================
+# PPG Tools - 光电容积脉搏波工具
+# =====================================================
+
+@mcp.tool(name="ppg_simulate", description="Simulate a photoplethysmogram (PPG) signal.")
+def ppg_simulate(duration: int = 120, sampling_rate: int = 1000,
+                 heart_rate: int = 70, frequency_modulation: float = 0.2,
+                 drift: float = 0.0) -> dict:
+    """
+    Simulate a PPG signal.
+
+    Parameters:
+    - duration (int): Recording length in seconds (default: 120)
+    - sampling_rate (int): Sampling rate in Hz (default: 1000)
+    - heart_rate (int): Heart rate in beats per minute (default: 70)
+    - frequency_modulation (float): RSA modulation level (default: 0.2)
+    - drift (float): Baseline drift level (default: 0.0)
+
+    Returns:
+    - dict: Contains simulated PPG signal
+    """
+    try:
+        ppg = nk.ppg_simulate(
+            duration=duration,
+            sampling_rate=sampling_rate,
+            heart_rate=heart_rate,
+            frequency_modulation=frequency_modulation,
+            drift=drift
+        )
+        return {
+            "success": True,
+            "signal": ppg.tolist(),
+            "duration": duration,
+            "sampling_rate": sampling_rate,
+            "heart_rate": heart_rate,
+            "num_samples": len(ppg)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="ppg_process", description="Process PPG data including cleaning and peak detection.")
+def ppg_process(ppg_data: List[float], sampling_rate: int = 1000) -> dict:
+    """
+    Process PPG data using NeuroKit2's ppg_process function.
+
+    Parameters:
+    - ppg_data: list of PPG signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+
+    Returns:
+    - dict: Contains processed data columns and peaks info
+    """
+    try:
+        signals, info = nk.ppg_process(ppg=ppg_data, sampling_rate=sampling_rate)
+        return {
+            "success": True,
+            "columns": list(signals.columns),
+            "num_samples": len(signals),
+            "peaks": info.get("PPG_Peaks", []),
+            "heart_rate_mean": float(signals["PPG_Rate"].mean()) if "PPG_Rate" in signals.columns else None
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="ppg_clean", description="Clean a PPG signal.")
+def ppg_clean(ppg_data: List[float], sampling_rate: int = 1000,
+              method: str = "elgendi") -> dict:
+    """
+    Clean a PPG signal.
+
+    Parameters:
+    - ppg_data: list of PPG signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, cleaning method (default: "elgendi")
+
+    Returns:
+    - dict: Contains cleaned PPG signal
+    """
+    try:
+        cleaned = nk.ppg_clean(ppg=ppg_data, sampling_rate=sampling_rate, method=method)
+        return {
+            "success": True,
+            "signal": cleaned.tolist(),
+            "method": method,
+            "num_samples": len(cleaned)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+# =====================================================
+# RSP Tools - 呼吸信号工具
+# =====================================================
+
+@mcp.tool(name="rsp_simulate", description="Simulate a respiration signal.")
+def rsp_simulate(duration: int = 120, sampling_rate: int = 1000,
+                 respiratory_rate: int = 15, noise: float = 0.01) -> dict:
+    """
+    Simulate a respiration signal.
+
+    Parameters:
+    - duration (int): Recording length in seconds (default: 120)
+    - sampling_rate (int): Sampling rate in Hz (default: 1000)
+    - respiratory_rate (int): Respiratory rate in breaths per minute (default: 15)
+    - noise (float): Noise level (default: 0.01)
+
+    Returns:
+    - dict: Contains simulated RSP signal
+    """
+    try:
+        rsp = nk.rsp_simulate(
+            duration=duration,
+            sampling_rate=sampling_rate,
+            respiratory_rate=respiratory_rate,
+            noise=noise
+        )
+        return {
+            "success": True,
+            "signal": rsp.tolist(),
+            "duration": duration,
+            "sampling_rate": sampling_rate,
+            "respiratory_rate": respiratory_rate,
+            "num_samples": len(rsp)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="rsp_process", description="Process respiration data including cleaning and feature extraction.")
+def rsp_process(rsp_data: List[float], sampling_rate: int = 1000,
+                method: str = "khodadad2018") -> dict:
+    """
+    Process respiration data using NeuroKit2's rsp_process function.
+
+    Parameters:
+    - rsp_data: list of respiration signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, peak detection method (default: "khodadad2018")
+
+    Returns:
+    - dict: Contains processed data and peaks info
+    """
+    try:
+        signals, info = nk.rsp_process(rsp=rsp_data, sampling_rate=sampling_rate, method=method)
+        return {
+            "success": True,
+            "columns": list(signals.columns),
+            "num_samples": len(signals),
+            "peaks": list(info.get("RSP_Peaks", [])),
+            "troughs": list(info.get("RSP_Troughs", [])),
+            "rate_mean": float(signals["RSP_Rate"].mean()) if "RSP_Rate" in signals.columns else None
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="rsp_clean", description="Clean a respiration signal.")
+def rsp_clean(rsp_data: List[float], sampling_rate: int = 1000,
+              method: str = "khodadad2018") -> dict:
+    """
+    Clean a respiration signal.
+
+    Parameters:
+    - rsp_data: list of respiration signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, cleaning method (default: "khodadad2018")
+
+    Returns:
+    - dict: Contains cleaned RSP signal
+    """
+    try:
+        cleaned = nk.rsp_clean(rsp=rsp_data, sampling_rate=sampling_rate, method=method)
+        return {
+            "success": True,
+            "signal": cleaned.tolist(),
+            "method": method,
+            "num_samples": len(cleaned)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+# =====================================================
+# EMG Tools - 肌电图工具
+# =====================================================
+
+@mcp.tool(name="emg_simulate", description="Simulate an EMG signal.")
+def emg_simulate(duration: int = 10, sampling_rate: int = 1000,
+                 burst_number: int = 5, burst_duration: float = 1.0,
+                 noise: float = 0.01) -> dict:
+    """
+    Simulate an EMG signal.
+
+    Parameters:
+    - duration (int): Recording length in seconds (default: 10)
+    - sampling_rate (int): Sampling rate in Hz (default: 1000)
+    - burst_number (int): Number of muscle activations (default: 5)
+    - burst_duration (float): Duration of each burst in seconds (default: 1.0)
+    - noise (float): Noise level (default: 0.01)
+
+    Returns:
+    - dict: Contains simulated EMG signal
+    """
+    try:
+        emg = nk.emg_simulate(
+            duration=duration,
+            sampling_rate=sampling_rate,
+            burst_number=burst_number,
+            burst_duration=burst_duration,
+            noise=noise
+        )
+        return {
+            "success": True,
+            "signal": emg.tolist(),
+            "duration": duration,
+            "sampling_rate": sampling_rate,
+            "num_samples": len(emg)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="emg_process", description="Process EMG data including cleaning and activation detection.")
+def emg_process(emg_data: List[float], sampling_rate: int = 1000) -> dict:
+    """
+    Process EMG data using NeuroKit2's emg_process function.
+
+    Parameters:
+    - emg_data: list of EMG signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+
+    Returns:
+    - dict: Contains processed data columns
+    """
+    try:
+        signals, info = nk.emg_process(emg=emg_data, sampling_rate=sampling_rate)
+        return {
+            "success": True,
+            "columns": list(signals.columns),
+            "num_samples": len(signals),
+            "onsets": list(info.get("EMG_Onsets", [])),
+            "offsets": list(info.get("EMG_Offsets", []))
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="emg_clean", description="Clean an EMG signal.")
+def emg_clean(emg_data: List[float], sampling_rate: int = 1000) -> dict:
+    """
+    Clean an EMG signal.
+
+    Parameters:
+    - emg_data: list of EMG signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+
+    Returns:
+    - dict: Contains cleaned EMG signal
+    """
+    try:
+        cleaned = nk.emg_clean(emg=emg_data, sampling_rate=sampling_rate)
+        return {
+            "success": True,
+            "signal": cleaned.tolist(),
+            "num_samples": len(cleaned)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="emg_amplitude", description="Compute EMG amplitude envelope.")
+def emg_amplitude(emg_data: List[float], sampling_rate: int = 1000,
+                  method: str = "rms", size: int = None) -> dict:
+    """
+    Compute EMG amplitude envelope.
+
+    Parameters:
+    - emg_data: list of cleaned EMG signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, method for amplitude computation (default: "rms")
+    - size: int, window size in samples (default: None, auto-calculated)
+
+    Returns:
+    - dict: Contains amplitude envelope
+    """
+    try:
+        amplitude = nk.emg_amplitude(emg_cleaned=emg_data, sampling_rate=sampling_rate)
+        return {
+            "success": True,
+            "amplitude": amplitude.tolist(),
+            "num_samples": len(amplitude)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+# =====================================================
+# HRV Tools - 心率变异性工具
+# =====================================================
+
+@mcp.tool(name="hrv_time", description="Compute time-domain HRV indices.")
+def hrv_time(peaks: List[int], sampling_rate: int = 1000) -> dict:
+    """
+    Compute time-domain HRV indices from R-peaks.
+
+    Parameters:
+    - peaks: list of R-peak sample indices
+    - sampling_rate: int, the sampling rate (default: 1000)
+
+    Returns:
+    - dict: Contains time-domain HRV metrics (SDNN, RMSSD, pNN50, etc.)
+    """
+    try:
+        hrv = nk.hrv_time(peaks=peaks, sampling_rate=sampling_rate)
+        result = {"success": True}
+        for col in hrv.columns:
+            val = hrv[col].values[0]
+            result[col] = float(val) if not np.isnan(val) else None
+        return result
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="hrv_frequency", description="Compute frequency-domain HRV indices.")
+def hrv_frequency(peaks: List[int], sampling_rate: int = 1000,
+                  psd_method: str = "welch") -> dict:
+    """
+    Compute frequency-domain HRV indices from R-peaks.
+
+    Parameters:
+    - peaks: list of R-peak sample indices
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - psd_method: str, PSD estimation method (default: "welch")
 
     Returns:
-    - dict: Contains success, result, and error fields
+    - dict: Contains frequency-domain HRV metrics (LF, HF, LF/HF ratio, etc.)
     """
     try:
-        processed_data, info = nk.eda_process(eda=eda_data, sampling_rate=sampling_rate)
-        return {"success": True, "result": {"processed_data": processed_data, "info": info}, "error": None}
+        hrv = nk.hrv_frequency(peaks=peaks, sampling_rate=sampling_rate, psd_method=psd_method)
+        result = {"success": True}
+        for col in hrv.columns:
+            val = hrv[col].values[0]
+            result[col] = float(val) if not np.isnan(val) else None
+        return result
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="hrv_nonlinear", description="Compute nonlinear HRV indices.")
+def hrv_nonlinear(peaks: List[int], sampling_rate: int = 1000) -> dict:
+    """
+    Compute nonlinear HRV indices from R-peaks.
+
+    Parameters:
+    - peaks: list of R-peak sample indices
+    - sampling_rate: int, the sampling rate (default: 1000)
+
+    Returns:
+    - dict: Contains nonlinear HRV metrics (SD1, SD2, ApEn, SampEn, etc.)
+    """
+    try:
+        hrv = nk.hrv_nonlinear(peaks=peaks, sampling_rate=sampling_rate)
+        result = {"success": True}
+        for col in hrv.columns:
+            val = hrv[col].values[0]
+            result[col] = float(val) if not np.isnan(val) else None
+        return result
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+# =====================================================
+# Signal Processing Tools - 通用信号处理工具
+# =====================================================
+
+@mcp.tool(name="signal_simulate", description="Simulate a continuous signal with specified frequency components.")
+def signal_simulate(duration: float = 10.0, sampling_rate: int = 1000,
+                    frequency: float = 1.0, amplitude: float = 0.5,
+                    noise: float = 0.0) -> dict:
+    """
+    Simulate a continuous signal with specified frequency components.
+
+    Parameters:
+    - duration (float): Length in seconds (default: 10.0)
+    - sampling_rate (int): Sampling rate in Hz (default: 1000)
+    - frequency (float): Oscillatory frequency in Hz (default: 1.0)
+    - amplitude (float): Signal amplitude (default: 0.5)
+    - noise (float): Noise level (default: 0.0)
+
+    Returns:
+    - dict: Contains simulated signal
+    """
+    try:
+        signal = nk.signal_simulate(
+            duration=duration,
+            sampling_rate=sampling_rate,
+            frequency=frequency,
+            amplitude=amplitude,
+            noise=noise
+        )
+        return {
+            "success": True,
+            "signal": signal.tolist(),
+            "duration": duration,
+            "sampling_rate": sampling_rate,
+            "num_samples": len(signal)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="signal_filter", description="Apply digital filter to a signal.")
+def signal_filter(signal_data: List[float], sampling_rate: int = 1000,
+                  lowcut: float = None, highcut: float = None,
+                  method: str = "butterworth", order: int = 2) -> dict:
+    """
+    Apply digital filter to a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - lowcut: float, low cutoff frequency for highpass (default: None)
+    - highcut: float, high cutoff frequency for lowpass (default: None)
+    - method: str, filter method (default: "butterworth")
+        Options: "butterworth", "fir", "bessel", "savgol"
+    - order: int, filter order (default: 2)
+
+    Returns:
+    - dict: Contains filtered signal
+    """
+    try:
+        filtered = nk.signal_filter(
+            signal=signal_data,
+            sampling_rate=sampling_rate,
+            lowcut=lowcut,
+            highcut=highcut,
+            method=method,
+            order=order
+        )
+        return {
+            "success": True,
+            "signal": filtered.tolist(),
+            "lowcut": lowcut,
+            "highcut": highcut,
+            "method": method,
+            "num_samples": len(filtered)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="signal_resample", description="Resample a signal to a different sampling rate.")
+def signal_resample(signal_data: List[float], desired_length: int = None,
+                    sampling_rate: int = None, desired_sampling_rate: int = None,
+                    method: str = "interpolation") -> dict:
+    """
+    Resample a signal to a different sampling rate or length.
+
+    Parameters:
+    - signal_data: list of signal values
+    - desired_length: int, desired number of samples (default: None)
+    - sampling_rate: int, original sampling rate (default: None)
+    - desired_sampling_rate: int, target sampling rate (default: None)
+    - method: str, resampling method (default: "interpolation")
+
+    Returns:
+    - dict: Contains resampled signal
+    """
+    try:
+        resampled = nk.signal_resample(
+            signal=signal_data,
+            desired_length=desired_length,
+            sampling_rate=sampling_rate,
+            desired_sampling_rate=desired_sampling_rate,
+            method=method
+        )
+        return {
+            "success": True,
+            "signal": resampled.tolist(),
+            "original_length": len(signal_data),
+            "new_length": len(resampled),
+            "method": method
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="signal_psd", description="Compute the Power Spectral Density (PSD) of a signal.")
+def signal_psd(signal_data: List[float], sampling_rate: int = 1000,
+               method: str = "welch", min_frequency: float = 0.0,
+               max_frequency: float = None) -> dict:
+    """
+    Compute the Power Spectral Density (PSD) of a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - sampling_rate: int, the sampling rate (default: 1000)
+    - method: str, PSD method (default: "welch")
+        Options: "welch", "fft", "multitapers", "burg"
+    - min_frequency: float, minimum frequency (default: 0.0)
+    - max_frequency: float, maximum frequency (default: None = Nyquist)
+
+    Returns:
+    - dict: Contains power spectral density and frequencies
+    """
+    try:
+        psd = nk.signal_psd(
+            signal=signal_data,
+            sampling_rate=sampling_rate,
+            method=method,
+            min_frequency=min_frequency,
+            max_frequency=max_frequency if max_frequency else np.inf,
+            show=False
+        )
+        return {
+            "success": True,
+            "frequency": psd["Frequency"].tolist(),
+            "power": psd["Power"].tolist(),
+            "method": method
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="signal_findpeaks", description="Find peaks in a signal.")
+def signal_findpeaks(signal_data: List[float], height_min: float = None,
+                     relative_height_min: float = None) -> dict:
+    """
+    Find peaks in a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - height_min: float, minimum peak height (default: None)
+    - relative_height_min: float, minimum relative height (default: None)
+
+    Returns:
+    - dict: Contains peak indices and amplitudes
+    """
+    try:
+        info = nk.signal_findpeaks(
+            signal=signal_data,
+            height_min=height_min,
+            relative_height_min=relative_height_min
+        )
+        return {
+            "success": True,
+            "peaks": list(info.get("Peaks", [])),
+            "num_peaks": len(info.get("Peaks", [])),
+            "heights": list(info.get("Height", [])) if "Height" in info else []
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="signal_detrend", description="Remove trend from a signal.")
+def signal_detrend(signal_data: List[float], method: str = "polynomial",
+                   order: int = 1) -> dict:
+    """
+    Remove trend from a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - method: str, detrending method (default: "polynomial")
+        Options: "polynomial", "tarvainen2002", "loess"
+    - order: int, polynomial order for polynomial detrending (default: 1)
+
+    Returns:
+    - dict: Contains detrended signal
+    """
+    try:
+        detrended = nk.signal_detrend(signal=signal_data, method=method, order=order)
+        return {
+            "success": True,
+            "signal": detrended.tolist(),
+            "method": method,
+            "num_samples": len(detrended)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="signal_smooth", description="Smooth a signal.")
+def signal_smooth(signal_data: List[float], method: str = "convolution",
+                  kernel: str = "boxcar", size: int = 10) -> dict:
+    """
+    Smooth a signal using various methods.
+
+    Parameters:
+    - signal_data: list of signal values
+    - method: str, smoothing method (default: "convolution")
+        Options: "convolution", "loess", "savgol"
+    - kernel: str, kernel type for convolution (default: "boxcar")
+    - size: int, kernel/window size (default: 10)
+
+    Returns:
+    - dict: Contains smoothed signal
+    """
+    try:
+        smoothed = nk.signal_smooth(signal=signal_data, method=method, kernel=kernel, size=size)
+        return {
+            "success": True,
+            "signal": smoothed.tolist(),
+            "method": method,
+            "num_samples": len(smoothed)
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+# =====================================================
+# Complexity Tools - 复杂度分析工具
+# =====================================================
+
+@mcp.tool(name="entropy_sample", description="Compute Sample Entropy of a signal.")
+def entropy_sample(signal_data: List[float], dimension: int = 2,
+                   tolerance: float = None) -> dict:
+    """
+    Compute Sample Entropy (SampEn) of a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - dimension: int, embedding dimension (default: 2)
+    - tolerance: float, tolerance threshold (default: None, 0.2*std)
+
+    Returns:
+    - dict: Contains Sample Entropy value
+    """
+    try:
+        if tolerance is None:
+            tolerance = 0.2 * np.std(signal_data)
+        sampen, info = nk.entropy_sample(signal=signal_data, dimension=dimension, tolerance=tolerance)
+        return {
+            "success": True,
+            "sample_entropy": float(sampen),
+            "dimension": dimension,
+            "tolerance": tolerance
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="entropy_shannon", description="Compute Shannon Entropy of a signal.")
+def entropy_shannon(signal_data: List[float], base: float = 2.0) -> dict:
+    """
+    Compute Shannon Entropy of a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - base: float, logarithm base (default: 2.0)
+
+    Returns:
+    - dict: Contains Shannon Entropy value
+    """
+    try:
+        shanen, info = nk.entropy_shannon(signal=signal_data, base=base)
+        return {
+            "success": True,
+            "shannon_entropy": float(shanen),
+            "base": base
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="entropy_approximate", description="Compute Approximate Entropy of a signal.")
+def entropy_approximate(signal_data: List[float], dimension: int = 2,
+                        tolerance: float = None) -> dict:
+    """
+    Compute Approximate Entropy (ApEn) of a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - dimension: int, embedding dimension (default: 2)
+    - tolerance: float, tolerance threshold (default: None, 0.2*std)
+
+    Returns:
+    - dict: Contains Approximate Entropy value
+    """
+    try:
+        if tolerance is None:
+            tolerance = 0.2 * np.std(signal_data)
+        apen, info = nk.entropy_approximate(signal=signal_data, dimension=dimension, tolerance=tolerance)
+        return {
+            "success": True,
+            "approximate_entropy": float(apen),
+            "dimension": dimension,
+            "tolerance": tolerance
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="entropy_fuzzy", description="Compute Fuzzy Entropy of a signal.")
+def entropy_fuzzy(signal_data: List[float], dimension: int = 2,
+                  tolerance: float = None) -> dict:
+    """
+    Compute Fuzzy Entropy (FuzzyEn) of a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - dimension: int, embedding dimension (default: 2)
+    - tolerance: float, tolerance threshold (default: None, 0.2*std)
+
+    Returns:
+    - dict: Contains Fuzzy Entropy value
+    """
+    try:
+        if tolerance is None:
+            tolerance = 0.2 * np.std(signal_data)
+        fuzzen, info = nk.entropy_fuzzy(signal=signal_data, dimension=dimension, tolerance=tolerance)
+        return {
+            "success": True,
+            "fuzzy_entropy": float(fuzzen),
+            "dimension": dimension,
+            "tolerance": tolerance
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="entropy_permutation", description="Compute Permutation Entropy of a signal.")
+def entropy_permutation(signal_data: List[float], dimension: int = 3,
+                        delay: int = 1) -> dict:
+    """
+    Compute Permutation Entropy (PermEn) of a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - dimension: int, embedding dimension (default: 3)
+    - delay: int, time delay (default: 1)
+
+    Returns:
+    - dict: Contains Permutation Entropy value
+    """
+    try:
+        permen, info = nk.entropy_permutation(signal=signal_data, dimension=dimension, delay=delay)
+        return {
+            "success": True,
+            "permutation_entropy": float(permen),
+            "dimension": dimension,
+            "delay": delay
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="fractal_dfa", description="Compute Detrended Fluctuation Analysis (DFA).")
+def fractal_dfa(signal_data: List[float], multifractal: bool = False) -> dict:
+    """
+    Compute Detrended Fluctuation Analysis (DFA) of a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - multifractal: bool, compute multifractal DFA (default: False)
+
+    Returns:
+    - dict: Contains DFA exponent (alpha)
+    """
+    try:
+        dfa, info = nk.fractal_dfa(signal=signal_data, multifractal=multifractal)
+        return {
+            "success": True,
+            "dfa_alpha": float(dfa),
+            "multifractal": multifractal
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="fractal_higuchi", description="Compute Higuchi Fractal Dimension.")
+def fractal_higuchi(signal_data: List[float], k_max: int = 10) -> dict:
+    """
+    Compute Higuchi Fractal Dimension (HFD) of a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+    - k_max: int, maximum k value (default: 10)
+
+    Returns:
+    - dict: Contains Higuchi Fractal Dimension
+    """
+    try:
+        hfd, info = nk.fractal_higuchi(signal=signal_data, k_max=k_max)
+        return {
+            "success": True,
+            "higuchi_fd": float(hfd),
+            "k_max": k_max
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="complexity_hjorth", description="Compute Hjorth parameters of a signal.")
+def complexity_hjorth(signal_data: List[float]) -> dict:
+    """
+    Compute Hjorth parameters (Activity, Mobility, Complexity) of a signal.
+
+    Parameters:
+    - signal_data: list of signal values
+
+    Returns:
+    - dict: Contains Hjorth Activity, Mobility, and Complexity
+    """
+    try:
+        hjorth = nk.complexity_hjorth(signal=signal_data)
+        return {
+            "success": True,
+            "activity": float(hjorth[0]),
+            "mobility": float(hjorth[1]),
+            "complexity": float(hjorth[2])
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+# =====================================================
+# Information Tools - 信息工具
+# =====================================================
+
+@mcp.tool(name="get_neurokit_info", description="Get NeuroKit2 library information and capabilities.")
+def get_neurokit_info() -> dict:
+    """
+    Get NeuroKit2 library information and capabilities.
+
+    Returns:
+    - dict: Contains library information and main features.
+    """
+    try:
+        return {
+            "success": True,
+            "name": "NeuroKit2",
+            "version": nk.__version__,
+            "description": "A Python toolbox for neurophysiological signal processing",
+            "modules": {
+                "ecg": "ECG (Electrocardiogram) processing - simulate, clean, peaks, process, quality",
+                "eda": "EDA (Electrodermal Activity/GSR) processing - simulate, phasic/tonic decomposition",
+                "ppg": "PPG (Photoplethysmogram) processing - simulate, clean, peaks, process",
+                "rsp": "RSP (Respiration) processing - simulate, clean, rate analysis",
+                "emg": "EMG (Electromyography) processing - simulate, clean, activation detection",
+                "eog": "EOG (Electrooculography) processing",
+                "eeg": "EEG (Electroencephalography) processing",
+                "hrv": "HRV (Heart Rate Variability) - time, frequency, nonlinear domains",
+                "signal": "General signal processing - filter, resample, PSD, peaks",
+                "complexity": "Complexity analysis - entropy, fractal dimensions",
+                "stats": "Statistical analysis functions"
+            },
+            "features": [
+                "Signal simulation for ECG, PPG, EDA, RSP, EMG",
+                "Comprehensive signal cleaning and artifact removal",
+                "Peak detection algorithms for cardiac signals",
+                "Heart Rate Variability (HRV) analysis",
+                "Complexity and entropy measures",
+                "Power spectral density analysis",
+                "Event-related analysis",
+                "Interval-related analysis"
+            ]
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="list_signal_processing_methods", description="List available signal processing methods.")
+def list_signal_processing_methods() -> dict:
+    """
+    List available signal processing methods for various operations.
+
+    Returns:
+    - dict: Contains available methods for different operations.
+    """
+    try:
+        return {
+            "success": True,
+            "ecg_cleaning_methods": ["neurokit", "biosppy", "pantompkins", "hamilton", "elgendi", "engzeemod"],
+            "ecg_peak_methods": ["neurokit", "pantompkins", "hamilton", "christov", "gamboa", "elgendi", "engzee", "kalidas2017", "martinez2003", "nabian2018", "rodrigues2020"],
+            "eda_decomposition_methods": ["highpass", "cvxeda", "smoothmedian", "sparse"],
+            "rsp_cleaning_methods": ["khodadad2018", "biosppy"],
+            "filter_methods": ["butterworth", "butterworth_ba", "fir", "bessel", "savgol"],
+            "psd_methods": ["welch", "fft", "multitapers", "burg", "lombscargle"],
+            "detrend_methods": ["polynomial", "tarvainen2002", "loess"],
+            "resample_methods": ["interpolation", "numpy", "fft", "poly", "pandas"]
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="list_hrv_indices", description="List available HRV indices and their descriptions.")
+def list_hrv_indices() -> dict:
+    """
+    List available HRV indices and their descriptions.
+
+    Returns:
+    - dict: Contains HRV indices organized by domain.
+    """
+    try:
+        return {
+            "success": True,
+            "time_domain": {
+                "MeanNN": "Mean of RR intervals (ms)",
+                "SDNN": "Standard deviation of RR intervals (ms)",
+                "RMSSD": "Root mean square of successive differences (ms)",
+                "SDSD": "Standard deviation of successive differences (ms)",
+                "pNN50": "Percentage of successive RR intervals differing by >50ms (%)",
+                "pNN20": "Percentage of successive RR intervals differing by >20ms (%)",
+                "CVNN": "Coefficient of variation of RR intervals",
+                "CVSD": "RMSSD divided by MeanNN",
+                "MedianNN": "Median of RR intervals (ms)",
+                "MadNN": "Median absolute deviation of RR intervals (ms)",
+                "HTI": "HRV triangular index",
+                "TINN": "Triangular interpolation of RR intervals (ms)"
+            },
+            "frequency_domain": {
+                "ULF": "Ultra low frequency power (<0.0033 Hz)",
+                "VLF": "Very low frequency power (0.0033-0.04 Hz)",
+                "LF": "Low frequency power (0.04-0.15 Hz)",
+                "HF": "High frequency power (0.15-0.4 Hz)",
+                "VHF": "Very high frequency power (0.4-0.5 Hz)",
+                "LFHF": "LF/HF ratio",
+                "LFn": "Normalized LF power",
+                "HFn": "Normalized HF power",
+                "LnHF": "Log-transformed HF"
+            },
+            "nonlinear": {
+                "SD1": "Poincaré plot SD1 (short-term variability)",
+                "SD2": "Poincaré plot SD2 (long-term variability)",
+                "SD1SD2": "SD1/SD2 ratio",
+                "ApEn": "Approximate Entropy",
+                "SampEn": "Sample Entropy",
+                "DFA_alpha1": "Short-term DFA exponent",
+                "DFA_alpha2": "Long-term DFA exponent"
+            }
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@mcp.tool(name="list_entropy_measures", description="List available entropy and complexity measures.")
+def list_entropy_measures() -> dict:
+    """
+    List available entropy and complexity measures.
+
+    Returns:
+    - dict: Contains entropy and complexity measures with descriptions.
+    """
+    try:
+        return {
+            "success": True,
+            "entropy_measures": {
+                "ShannonEntropy": "Measures uncertainty/randomness - higher values = more random",
+                "SampleEntropy": "Regularity measure - higher = more complex, lower = more regular",
+                "ApproximateEntropy": "Similar to SampEn but counts self-matches",
+                "FuzzyEntropy": "Like SampEn but with fuzzy membership functions",
+                "PermutationEntropy": "Complexity measure based on ordinal patterns",
+                "SpectralEntropy": "Entropy of the power spectral density",
+                "SVDEntropy": "Entropy based on singular value decomposition",
+                "DispersionEntropy": "Based on dispersion patterns",
+                "BubbleEntropy": "Based on bubble sort operations"
+            },
+            "fractal_measures": {
+                "DFA": "Detrended Fluctuation Analysis - measures self-similarity",
+                "HiguchisFD": "Higuchi Fractal Dimension - curve complexity",
+                "KatzFD": "Katz Fractal Dimension",
+                "PetrosianFD": "Petrosian Fractal Dimension",
+                "SevcikFD": "Sevcik/Normalized Fractal Dimension",
+                "HurstExponent": "Long-range dependence measure"
+            },
+            "other_measures": {
+                "HjorthActivity": "Signal power/variance",
+                "HjorthMobility": "Mean frequency estimate",
+                "HjorthComplexity": "Bandwidth estimate",
+                "LempelZivComplexity": "Algorithmic complexity",
+                "LyapunovExponent": "Chaos measure - sensitivity to initial conditions"
+            }
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
 
 def create_app() -> FastMCP:
     """