Update scvelo/mcp_output/mcp_plugin/mcp_service.py

scvelo/mcp_output/mcp_plugin/mcp_service.py

@@ -1,103 +1,428 @@
+import os
+import sys
+from typing import Any, List, Optional, Dict
+
+# Add the local source directory to sys.path
+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)
+
+import numpy as np
 from fastmcp import FastMCP
 
+# Import scvelo core modules
+from scvelo.core._arithmetic import clipped_log, invert, multiply, prod_sum, sum as scv_sum
+from scvelo.core._metrics import l2_norm
+from scvelo.core._models import SplicingDynamics
+from scvelo.core._linear_models import LinearRegression
+from scvelo.core._parallelize import get_n_jobs
+
 # Create the FastMCP service application
 mcp = FastMCP("scvelo_service")
 
-@mcp.tool(name="read_data", description="Read and load data into scvelo")
-def read_data(file_path: str) -> dict:
+
+# ===================== Arithmetic Tools =====================
+
+@mcp.tool(name="arithmetic_tool", description="Tool for arithmetic operations")
+def arithmetic_tool(a: float, b: float) -> dict:
+    """
+    Performs basic arithmetic operations on two numbers.
+
+    Parameters:
+    - a (float): First number.
+    - b (float): Second number.
+
+    Returns:
+    - dict: Result containing sum, product, and other calculations.
+    """
+    try:
+        result = {
+            "sum": float(a + b),
+            "product": float(a * b),
+            "difference": float(a - b),
+            "division": float(a / b) if b != 0 else None
+        }
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="clipped_log_tool", description="Calculate clipped logarithm of values")
+def clipped_log_tool(values: List[float], lb: float = 0.0, ub: float = 1.0, eps: float = 1e-6) -> dict:
+    """
+    Calculate logarithm of values clipped to [lb + eps, ub - eps].
+
+    Parameters:
+    - values (List[float]): Input values to calculate clipped log.
+    - lb (float): Lower bound for clipping. Default 0.0.
+    - ub (float): Upper bound for clipping. Default 1.0.
+    - eps (float): Epsilon offset. Default 1e-6.
+
+    Returns:
+    - dict: Clipped logarithm values.
+    """
+    try:
+        arr = np.array(values)
+        result = clipped_log(arr, lb=lb, ub=ub, eps=eps)
+        return {"success": True, "result": result.tolist(), "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+# ===================== Metrics Tools =====================
+
+@mcp.tool(name="metrics_tool", description="Tool for metrics calculations")
+def metrics_tool(values: List[float]) -> dict:
+    """
+    Calculates various metrics including l2 norm, mean, std, etc.
+
+    Parameters:
+    - values (List[float]): List of values for metrics calculation.
+
+    Returns:
+    - dict: Result of the metrics calculation with success status.
+    """
+    try:
+        arr = np.array(values)
+        result = {
+            "l2_norm": float(np.sqrt(np.sum(arr ** 2))),
+            "mean": float(np.mean(arr)),
+            "std": float(np.std(arr)),
+            "min": float(np.min(arr)),
+            "max": float(np.max(arr)),
+            "sum": float(np.sum(arr))
+        }
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="l2_norm_tool", description="Calculate L2 norm of a matrix along specified axis")
+def l2_norm_tool(matrix: List[List[float]], axis: int = 1) -> dict:
+    """
+    Calculate L2 norm along a given axis.
+
+    Parameters:
+    - matrix (List[List[float]]): 2D matrix of values.
+    - axis (int): Axis along which to calculate l2 norm (0 or 1). Default 1.
+
+    Returns:
+    - dict: L2 norm values along the specified axis.
+    """
+    try:
+        arr = np.array(matrix)
+        result = l2_norm(arr, axis=axis)
+        return {"success": True, "result": result.tolist(), "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+# ===================== Models Tools =====================
+
+@mcp.tool(name="models_tool", description="Tool for model operations")
+def models_tool(model_data: dict) -> dict:
+    """
+    Performs operations related to SplicingDynamics model.
+
+    Parameters:
+    - model_data (dict): Dictionary containing model parameters:
+        - alpha (float): Transcription rate
+        - beta (float): Translation rate  
+        - gamma (float): Splicing degradation rate
+        - initial_state (list, optional): Initial [u0, s0] state
+        - time_points (list, optional): Time points for solution
+
+    Returns:
+    - dict: Result of the model operation with success status.
+    """
+    try:
+        alpha = model_data.get("alpha", 1.0)
+        beta = model_data.get("beta", 1.0)
+        gamma = model_data.get("gamma", 0.5)
+        initial_state = model_data.get("initial_state", [0, 0])
+        time_points = model_data.get("time_points", [0, 1, 2, 3, 4, 5])
+        
+        dynamics = SplicingDynamics(
+            alpha=alpha,
+            beta=beta,
+            gamma=gamma,
+            initial_state=initial_state
+        )
+        
+        t = np.array(time_points)
+        solution = dynamics.get_solution(t, with_keys=True)
+        steady_states = dynamics.get_steady_states(with_keys=True)
+        
+        result = {
+            "solution": {
+                "unspliced": solution["u"].tolist() if isinstance(solution["u"], np.ndarray) else [solution["u"]],
+                "spliced": solution["s"].tolist() if isinstance(solution["s"], np.ndarray) else [solution["s"]]
+            },
+            "steady_states": {
+                "unspliced": float(steady_states["u"]),
+                "spliced": float(steady_states["s"])
+            },
+            "parameters": {
+                "alpha": alpha,
+                "beta": beta,
+                "gamma": gamma
+            }
+        }
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="splicing_dynamics_solution", description="Calculate RNA splicing dynamics solution over time")
+def splicing_dynamics_solution(
+    alpha: float,
+    beta: float,
+    gamma: float,
+    time_points: List[float],
+    u0: float = 0.0,
+    s0: float = 0.0
+) -> dict:
     """
-    Read and load data into scvelo.
+    Calculate RNA splicing dynamics solution.
 
     Parameters:
-    - file_path: Path to the data file.
+    - alpha (float): Transcription rate.
+    - beta (float): Translation/splicing rate.
+    - gamma (float): Degradation rate.
+    - time_points (List[float]): Time points to evaluate.
+    - u0 (float): Initial unspliced RNA abundance. Default 0.0.
+    - s0 (float): Initial spliced RNA abundance. Default 0.0.
 
     Returns:
-    - dict: Success status and loaded data information.
+    - dict: Time-course solution of unspliced and spliced RNA.
     """
     try:
-        from scvelo.read_load import read
-        data = read(file_path)
-        return {
-            "success": True,
-            "data_summary": str(data)
+        dynamics = SplicingDynamics(
+            alpha=alpha,
+            beta=beta,
+            gamma=gamma,
+            initial_state=[u0, s0]
+        )
+        
+        t = np.array(time_points)
+        solution = dynamics.get_solution(t, with_keys=True)
+        
+        result = {
+            "time": time_points,
+            "unspliced": solution["u"].tolist() if isinstance(solution["u"], np.ndarray) else [solution["u"]],
+            "spliced": solution["s"].tolist() if isinstance(solution["s"], np.ndarray) else [solution["s"]]
         }
+        return {"success": True, "result": result, "error": None}
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="compute_velocity", description="Compute RNA velocity")
-def compute_velocity(data: dict) -> dict:
+
+@mcp.tool(name="splicing_steady_state", description="Calculate steady state of RNA splicing dynamics")
+def splicing_steady_state(alpha: float, beta: float, gamma: float) -> dict:
     """
-    Compute RNA velocity using scvelo.
+    Calculate steady state of RNA splicing system.
 
     Parameters:
-    - data: The loaded data object.
+    - alpha (float): Transcription rate.
+    - beta (float): Translation/splicing rate.
+    - gamma (float): Degradation rate.
 
     Returns:
-    - dict: Success status and velocity computation results.
+    - dict: Steady state values for unspliced and spliced RNA.
     """
     try:
-        from scvelo.core import velocity
-        velocity(data)
-        return {
-            "success": True,
-            "message": "Velocity computed successfully."
+        dynamics = SplicingDynamics(alpha=alpha, beta=beta, gamma=gamma)
+        steady_states = dynamics.get_steady_states(with_keys=True)
+        
+        result = {
+            "unspliced_steady_state": float(steady_states["u"]),
+            "spliced_steady_state": float(steady_states["s"]),
+            "ratio_u_to_s": float(steady_states["u"] / steady_states["s"]) if steady_states["s"] != 0 else None
         }
+        return {"success": True, "result": result, "error": None}
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
+
+# ===================== Linear Models Tools =====================
 
-@mcp.tool(name="plot_velocity", description="Plot RNA velocity")
-def plot_velocity(data: dict, output_path: str) -> dict:
+@mcp.tool(name="linear_models_tool", description="Tool for linear model operations")
+def linear_models_tool(x: List[float], y: List[float], percentile: Optional[float] = None, fit_intercept: bool = False) -> dict:
     """
-    Plot RNA velocity results.
+    Performs linear regression fitting.
 
     Parameters:
-    - data: The data object with computed velocity.
-    - output_path: Path to save the plot.
+    - x (List[float]): Independent variable values.
+    - y (List[float]): Dependent variable values.
+    - percentile (float, optional): Percentile for extreme quantile regression.
+    - fit_intercept (bool): Whether to fit intercept. Default False.
 
     Returns:
-    - dict: Success status and plot information.
+    - dict: Regression coefficients and intercept.
     """
     try:
-        from scvelo.plotting import velocity_embedding_stream
-        velocity_embedding_stream(data, save=output_path)
-        return {
-            "success": True,
-            "message": f"Velocity plot saved to {output_path}."
+        x_arr = np.array(x).reshape(-1, 1)
+        y_arr = np.array(y)
+        
+        model = LinearRegression(
+            percentile=percentile,
+            fit_intercept=fit_intercept
+        )
+        model.fit(x_arr, y_arr)
+        
+        result = {
+            "coefficient": float(model.coef_[0]) if hasattr(model, 'coef_') else None,
+            "intercept": float(model.intercept_) if hasattr(model, 'intercept_') else 0.0
         }
+        return {"success": True, "result": result, "error": None}
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="run_pipeline", description="Run a full scvelo pipeline")
-def run_pipeline(file_path: str, output_path: str) -> dict:
+
+# ===================== Utility Tools =====================
+
+@mcp.tool(name="base_tool", description="Tool for base operations")
+def base_tool(param: str) -> dict:
     """
-    Run a full scvelo pipeline from data loading to plotting.
+    Performs base utility operations.
 
     Parameters:
-    - file_path: Path to the data file.
-    - output_path: Path to save the final plot.
+    - param (str): Input parameter for base operations.
 
     Returns:
-    - dict: Success status and pipeline results.
+    - dict: Result of the base operation.
     """
     try:
-        from scvelo.read_load import read
-        from scvelo.core import velocity
-        from scvelo.plotting import velocity_embedding_stream
+        result = {
+            "input": param,
+            "length": len(param),
+            "upper": param.upper(),
+            "lower": param.lower()
+        }
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="parallelize_tool", description="Tool for parallelization operations")
+def parallelize_tool(tasks: List[Any]) -> dict:
+    """
+    Get information about parallelization capabilities.
 
-        data = read(file_path)
-        velocity(data)
-        velocity_embedding_stream(data, save=output_path)
+    Parameters:
+    - tasks (List): List of tasks (used to determine optimal job count).
 
-        return {
-            "success": True,
-            "message": f"Pipeline completed successfully. Plot saved to {output_path}."
+    Returns:
+    - dict: Parallelization configuration info.
+    """
+    try:
+        n_tasks = len(tasks)
+        optimal_jobs = get_n_jobs(None)
+        
+        result = {
+            "n_tasks": n_tasks,
+            "available_cpus": os.cpu_count(),
+            "recommended_n_jobs": min(n_tasks, optimal_jobs),
+            "tasks_preview": tasks[:5] if len(tasks) > 5 else tasks
         }
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
+
+@mcp.tool(name="utils_tool", description="Tool for utility operations")
+def utils_tool(input_data: Any) -> dict:
+    """
+    Performs utility operations on input data.
+
+    Parameters:
+    - input_data: Input data for utility operations (can be any type).
+
+    Returns:
+    - dict: Information about the input data.
+    """
+    try:
+        if isinstance(input_data, list):
+            result = {
+                "type": "list",
+                "length": len(input_data),
+                "preview": input_data[:10] if len(input_data) > 10 else input_data
+            }
+        elif isinstance(input_data, dict):
+            result = {
+                "type": "dict",
+                "keys": list(input_data.keys()),
+                "n_keys": len(input_data)
+            }
+        elif isinstance(input_data, (int, float)):
+            result = {
+                "type": "number",
+                "value": input_data,
+                "is_integer": isinstance(input_data, int)
+            }
+        else:
+            result = {
+                "type": str(type(input_data).__name__),
+                "value": str(input_data)
+            }
+        return {"success": True, "result": result, "error": None}
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
+
+# ===================== AnnData Tools =====================
+
+@mcp.tool(name="anndata_tool", description="Tool for handling AnnData operations")
+def anndata_tool(data: dict) -> dict:
+    """
+    Provides information about AnnData operations available in scvelo.
+
+    Parameters:
+    - data (dict): Configuration for AnnData operations.
+        - operation (str): Operation type ('info', 'functions')
+
+    Returns:
+    - dict: Information about available AnnData operations.
+    """
+    try:
+        operation = data.get("operation", "info")
+        
+        if operation == "functions":
+            result = {
+                "available_functions": [
+                    "clean_obs_names - Clean up observation names",
+                    "cleanup - Delete not needed attributes",
+                    "get_df - Get DataFrame from AnnData",
+                    "get_initial_size - Get initial size",
+                    "get_modality - Get modality",
+                    "get_size - Get size",
+                    "make_dense - Convert sparse to dense",
+                    "make_sparse - Convert dense to sparse",
+                    "merge - Merge AnnData objects",
+                    "set_initial_size - Set initial size",
+                    "set_modality - Set modality",
+                    "show_proportions - Show proportions"
+                ]
+            }
+        else:
+            result = {
+                "description": "AnnData is the primary data structure in scvelo for storing single-cell RNA velocity data",
+                "main_components": [
+                    "X - Expression matrix",
+                    "layers - Additional matrices (spliced, unspliced)",
+                    "obs - Cell annotations",
+                    "var - Gene annotations",
+                    "uns - Unstructured data"
+                ]
+            }
+        return {"success": True, "result": result, "error": None}
+    except Exception as e:
+        return {"success": False, "result": None, "error": str(e)}
+
 
 def create_app() -> FastMCP:
     """
-    Create and return the FastMCP application instance.
+    Creates and returns the FastMCP application instance.
 
     Returns:
     - FastMCP: The FastMCP application instance.