MCP Tabular Data Analysis Server

Instructions

Implementation Reference

• src/mcp_tabular/server.py:592-684 (handler)

  The core handler function for the 'create_pivot_table' MCP tool. It loads the dataset using _load_data, validates parameters, defaults to first numeric column if needed, creates a pandas pivot_table, and returns structured results including the pivot data as list of records and summary statistics.
  @mcp.tool() def create_pivot_table( file_path: str, index: list[str], columns: list[str] | None = None, values: str | None = None, aggfunc: str = "mean", fill_value: float | None = None, ) -> dict[str, Any]: """ Create a pivot table from tabular data - the most common business analysis operation. Args: file_path: Path to CSV or SQLite file index: Column(s) to use as row labels (grouping) columns: Column(s) to use as column headers (optional) values: Column to aggregate (default: first numeric column) aggfunc: Aggregation function - 'sum', 'mean', 'count', 'min', 'max', 'median', 'std' fill_value: Value to replace missing entries (default: None = show as null) Returns: Dictionary containing the pivot table data and metadata Example: create_pivot_table( file_path="data/sales.csv", index=["region"], columns=["category"], values="revenue", aggfunc="sum" ) """ df = _load_data(file_path) # Validate index columns invalid = [c for c in index if c not in df.columns] if invalid: raise ValueError(f"Index columns not found: {invalid}. Available: {df.columns.tolist()}") # Validate columns if provided if columns: invalid = [c for c in columns if c not in df.columns] if invalid: raise ValueError(f"Column headers not found: {invalid}") # Default to first numeric column if values not specified if values is None: numeric_cols = _get_numeric_columns(df) if not numeric_cols: raise ValueError("No numeric columns found for aggregation") values = numeric_cols[0] elif values not in df.columns: raise ValueError(f"Values column '{values}' not found") # Map aggfunc string to function agg_map = { "sum": "sum", "mean": "mean", "count": "count", "min": "min", "max": "max", "median": "median", "std": "std", } if aggfunc not in agg_map: raise ValueError(f"Unknown aggfunc: {aggfunc}. Use: {list(agg_map.keys())}") # Create pivot table pivot = pd.pivot_table( df, values=values, index=index, columns=columns, aggfunc=agg_map[aggfunc], fill_value=fill_value, ) # Reset index for cleaner output pivot_reset = pivot.reset_index() return { "index": index, "columns": columns, "values": values, "aggfunc": aggfunc, "shape": {"rows": len(pivot), "columns": len(pivot.columns)}, "pivot_table": pivot_reset.to_dict(orient="records"), "summary": { "total": float(pivot.values.sum()) if np.issubdtype(pivot.values.dtype, np.number) else None, "grand_mean": float(pivot.values.mean()) if np.issubdtype(pivot.values.dtype, np.number) else None, } }