CSV Editor

Instructions

Implementation Reference

• src/csv_editor/server.py:370-379 (registration)

  Registration of the 'detect_outliers' tool using @mcp.tool decorator. This defines the tool interface and delegates to the implementation in analytics.py.

  @mcp.tool
  async def detect_outliers(
      session_id: str,
      columns: Optional[List[str]] = None,
      method: str = "iqr",
      threshold: float = 1.5,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """Detect outliers in numeric columns."""
      return await _detect_outliers(session_id, columns, method, threshold, ctx)

• src/csv_editor/tools/analytics.py:463-570 (handler)

  Core implementation of the detect_outliers tool logic, handling IQR and Z-score methods for outlier detection in numeric columns of the CSV data.

  async def detect_outliers(
      session_id: str,
      columns: Optional[List[str]] = None,
      method: str = "iqr",
      threshold: float = 1.5,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """
      Detect outliers in numeric columns.
      
      Args:
          session_id: Session identifier
          columns: Columns to check (None for all numeric)
          method: Detection method ('iqr', 'zscore', 'isolation_forest')
          threshold: Threshold for outlier detection (1.5 for IQR, 3 for z-score)
          ctx: FastMCP context
          
      Returns:
          Dict with outlier information
      """
      try:
          manager = get_session_manager()
          session = manager.get_session(session_id)
          
          if not session or session.df is None:
              return {"success": False, "error": "Invalid session or no data loaded"}
          
          df = session.df
          
          # Select numeric columns
          if columns:
              missing_cols = [col for col in columns if col not in df.columns]
              if missing_cols:
                  return {"success": False, "error": f"Columns not found: {missing_cols}"}
              numeric_df = df[columns].select_dtypes(include=[np.number])
          else:
              numeric_df = df.select_dtypes(include=[np.number])
          
          if numeric_df.empty:
              return {"success": False, "error": "No numeric columns found"}
          
          outliers = {}
          
          if method == "iqr":
              for col in numeric_df.columns:
                  Q1 = numeric_df[col].quantile(0.25)
                  Q3 = numeric_df[col].quantile(0.75)
                  IQR = Q3 - Q1
                  
                  lower_bound = Q1 - threshold * IQR
                  upper_bound = Q3 + threshold * IQR
                  
                  outlier_mask = (numeric_df[col] < lower_bound) | (numeric_df[col] > upper_bound)
                  outlier_indices = df.index[outlier_mask].tolist()
                  
                  outliers[col] = {
                      "method": "IQR",
                      "lower_bound": float(lower_bound),
                      "upper_bound": float(upper_bound),
                      "outlier_count": len(outlier_indices),
                      "outlier_percentage": round(len(outlier_indices) / len(df) * 100, 2),
                      "outlier_indices": outlier_indices[:100],  # Limit to first 100
                      "q1": float(Q1),
                      "q3": float(Q3),
                      "iqr": float(IQR)
                  }
                  
          elif method == "zscore":
              for col in numeric_df.columns:
                  z_scores = np.abs((numeric_df[col] - numeric_df[col].mean()) / numeric_df[col].std())
                  outlier_mask = z_scores > threshold
                  outlier_indices = df.index[outlier_mask].tolist()
                  
                  outliers[col] = {
                      "method": "Z-Score",
                      "threshold": threshold,
                      "outlier_count": len(outlier_indices),
                      "outlier_percentage": round(len(outlier_indices) / len(df) * 100, 2),
                      "outlier_indices": outlier_indices[:100],  # Limit to first 100
                      "mean": float(numeric_df[col].mean()),
                      "std": float(numeric_df[col].std())
                  }
                  
          else:
              return {"success": False, "error": f"Unknown method: {method}"}
          
          # Summary statistics
          total_outliers = sum(info["outlier_count"] for info in outliers.values())
          
          session.record_operation(OperationType.ANALYZE, {
              "type": "outlier_detection",
              "method": method,
              "threshold": threshold,
              "columns": list(outliers.keys())
          })
          
          return {
              "success": True,
              "method": method,
              "threshold": threshold,
              "outliers": outliers,
              "total_outliers": total_outliers,
              "columns_analyzed": list(outliers.keys())
          }
          
      except Exception as e:
          logger.error(f"Error detecting outliers: {str(e)}")
          return {"success": False, "error": str(e)}

• src/csv_editor/server.py:371-376 (schema)

  Input schema and parameters for the detect_outliers tool, defining expected arguments and their types/defaults.

  async def detect_outliers(
      session_id: str,
      columns: Optional[List[str]] = None,
      method: str = "iqr",
      threshold: float = 1.5,
      ctx: Context = None