MCP Tabular Data Analysis Server

Instructions

Implementation Reference

• src/mcp_tabular/server.py:163-234 (handler)

  The detect_anomalies tool handler function. It loads tabular data from a CSV or SQLite file, detects anomalies in a specified numeric column using either z-score or IQR methods, and returns detailed information about the detected anomalies including indices, rows, and statistics.
  @mcp.tool() def detect_anomalies( file_path: str, column: str, method: str = "zscore", threshold: float = 3.0, ) -> dict[str, Any]: """ Detect anomalies/outliers in a numeric column. Args: file_path: Path to CSV or SQLite file column: Name of the numeric column to analyze method: Detection method - 'zscore' (default), 'iqr', or 'isolation_forest' threshold: Threshold for anomaly detection (default 3.0 for zscore, 1.5 for IQR) Returns: Dictionary containing: - method: Detection method used - anomaly_count: Number of anomalies found - anomaly_indices: Row indices of anomalies - anomalies: The anomalous rows - statistics: Column statistics """ df = _load_data(file_path) if column not in df.columns: raise ValueError(f"Column '{column}' not found. Available: {df.columns.tolist()}") if not np.issubdtype(df[column].dtype, np.number): raise ValueError(f"Column '{column}' is not numeric") col_data = df[column].dropna() if method == "zscore": # Z-score method z_scores = np.abs(stats.zscore(col_data)) anomaly_mask = z_scores > threshold anomaly_indices = col_data[anomaly_mask].index.tolist() elif method == "iqr": # Interquartile Range method q1 = col_data.quantile(0.25) q3 = col_data.quantile(0.75) iqr = q3 - q1 lower_bound = q1 - threshold * iqr upper_bound = q3 + threshold * iqr anomaly_mask = (col_data < lower_bound) | (col_data > upper_bound) anomaly_indices = col_data[anomaly_mask].index.tolist() else: raise ValueError(f"Unknown method: {method}. Use 'zscore' or 'iqr'") anomalies_df = df.loc[anomaly_indices] return { "method": method, "threshold": threshold, "column": column, "anomaly_count": len(anomaly_indices), "anomaly_percentage": round(len(anomaly_indices) / len(col_data) * 100, 2), "anomaly_indices": anomaly_indices, "anomalies": anomalies_df.to_dict(orient="records"), "statistics": { "mean": float(col_data.mean()), "std": float(col_data.std()), "min": float(col_data.min()), "max": float(col_data.max()), "median": float(col_data.median()), } }