CSV Editor

Instructions

Implementation Reference

• src/csv_editor/tools/analytics.py:100-200 (handler)

  The primary handler function that implements the logic for computing detailed statistics (numeric, categorical, string) for a specific column in a CSV session's dataframe.
  async def get_column_statistics( session_id: str, column: str, ctx: Context = None ) -> Dict[str, Any]: """ Get detailed statistics for a specific column. Args: session_id: Session identifier column: Column name to analyze ctx: FastMCP context Returns: Dict with detailed column statistics """ 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 if column not in df.columns: return {"success": False, "error": f"Column '{column}' not found"} col_data = df[column] result = { "column": column, "dtype": str(col_data.dtype), "total_count": len(col_data), "null_count": int(col_data.isna().sum()), "null_percentage": round(col_data.isna().sum() / len(col_data) * 100, 2), "unique_count": int(col_data.nunique()), "unique_percentage": round(col_data.nunique() / len(col_data) * 100, 2) } # Numeric column statistics if pd.api.types.is_numeric_dtype(col_data): non_null = col_data.dropna() result.update({ "type": "numeric", "mean": float(non_null.mean()), "median": float(non_null.median()), "mode": float(non_null.mode()[0]) if len(non_null.mode()) > 0 else None, "std": float(non_null.std()), "variance": float(non_null.var()), "min": float(non_null.min()), "max": float(non_null.max()), "range": float(non_null.max() - non_null.min()), "sum": float(non_null.sum()), "skewness": float(non_null.skew()), "kurtosis": float(non_null.kurt()), "25%": float(non_null.quantile(0.25)), "50%": float(non_null.quantile(0.50)), "75%": float(non_null.quantile(0.75)), "iqr": float(non_null.quantile(0.75) - non_null.quantile(0.25)), "zero_count": int((col_data == 0).sum()), "positive_count": int((col_data > 0).sum()), "negative_count": int((col_data < 0).sum()) }) # Categorical column statistics else: value_counts = col_data.value_counts() top_values = value_counts.head(10).to_dict() result.update({ "type": "categorical", "most_frequent": str(value_counts.index[0]) if len(value_counts) > 0 else None, "most_frequent_count": int(value_counts.iloc[0]) if len(value_counts) > 0 else 0, "top_10_values": {str(k): int(v) for k, v in top_values.items()} }) # String-specific stats if col_data.dtype == 'object': str_data = col_data.dropna().astype(str) if len(str_data) > 0: str_lengths = str_data.str.len() result["string_stats"] = { "min_length": int(str_lengths.min()), "max_length": int(str_lengths.max()), "mean_length": round(str_lengths.mean(), 2), "empty_string_count": int((str_data == "").sum()) } session.record_operation(OperationType.ANALYZE, { "type": "column_statistics", "column": column }) return { "success": True, "statistics": result } except Exception as e: logger.error(f"Error getting column statistics: {str(e)}") return {"success": False, "error": str(e)}
• src/csv_editor/server.py:327-334 (registration)

  Registers the 'get_column_statistics' tool with the FastMCP server using the @mcp.tool decorator. This wrapper function delegates execution to the actual implementation imported as _get_column_statistics from analytics.py.
  @mcp.tool async def get_column_statistics( session_id: str, column: str, ctx: Context = None ) -> Dict[str, Any]: """Get detailed statistics for a specific column.""" return await _get_column_statistics(session_id, column, ctx)