group_aggregate
Group tabular data by specified columns and compute aggregations like sum, mean, count, or max to analyze patterns and summarize datasets.
Instructions
Group data and compute aggregations.
Args:
file_path: Path to CSV or SQLite file
group_by: Columns to group by
aggregations: Dictionary mapping column names to list of aggregation functions
(e.g., {"sales": ["sum", "mean"], "quantity": ["count", "max"]})
Supported: sum, mean, median, min, max, count, std, var
Returns:
Dictionary containing grouped and aggregated dataInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| file_path | Yes | ||
| group_by | Yes | ||
| aggregations | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/mcp_tabular/server.py:489-533 (handler)The main handler function for the 'group_aggregate' tool. It loads data using _load_data, validates inputs, performs pandas groupby with specified aggregations, flattens multi-level columns, and returns the results as a dictionary with metadata.
@mcp.tool() def group_aggregate( file_path: str, group_by: list[str], aggregations: dict[str, list[str]], ) -> dict[str, Any]: """ Group data and compute aggregations. Args: file_path: Path to CSV or SQLite file group_by: Columns to group by aggregations: Dictionary mapping column names to list of aggregation functions (e.g., {"sales": ["sum", "mean"], "quantity": ["count", "max"]}) Supported: sum, mean, median, min, max, count, std, var Returns: Dictionary containing grouped and aggregated data """ df = _load_data(file_path) # Validate group_by columns invalid = [c for c in group_by if c not in df.columns] if invalid: raise ValueError(f"Group-by columns not found: {invalid}") # Validate aggregation columns for col in aggregations: if col not in df.columns: raise ValueError(f"Aggregation column '{col}' not found") # Perform groupby grouped = df.groupby(group_by).agg(aggregations) # Flatten column names grouped.columns = ["_".join(col).strip() for col in grouped.columns] grouped = grouped.reset_index() return { "group_by": group_by, "aggregations": aggregations, "group_count": len(grouped), "result": grouped.to_dict(orient="records"), }