create_pivot_table
Generate pivot tables from CSV or SQLite data to summarize and analyze business metrics by grouping rows and columns with customizable aggregation functions.
Instructions
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"
)Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| file_path | Yes | ||
| index | Yes | ||
| columns | No | ||
| values | No | ||
| aggfunc | No | mean | |
| fill_value | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/mcp_tabular/server.py:594-685 (handler)The primary handler function for the 'create_pivot_table' MCP tool. It loads data using _load_data, validates parameters, creates a pandas pivot table, and returns structured results including the pivot data and summary statistics.
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, } }