pivot_table
Create pivot tables to reshape tabular data by grouping rows and columns with aggregated values for analysis.
Instructions
Create pivot tables from tabular data using Polars.
Like Excel pivot tables: reshape data with row/column dimensions and aggregated values.
Example:
SALES BY REGION AND PRODUCT: data=[ {"region":"North","product":"A","sales":100}, {"region":"North","product":"B","sales":150}, {"region":"South","product":"A","sales":80}, {"region":"South","product":"B","sales":120} ], index="region", columns="product", values="sales", aggfunc="sum" Result: product | A | B --------|------|------ North | 100 | 150 South | 80 | 120
COUNT AGGREGATION: Same data with aggfunc="count" Result: Count of entries per region-product combination
AVERAGE SCORES: data=[{"dept":"Sales","role":"Manager","score":85}, ...] index="dept", columns="role", values="score", aggfunc="mean" Result: Average scores by department and role
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| context | No | Optional annotation to label this calculation (e.g., 'Bond A PV', 'Q2 revenue'). Appears in results for easy identification. | |
| output_mode | No | Output format: full (default), compact, minimal, value, or final. See batch_execute tool for details. | full |
| data | Yes | List of row dictionaries | |
| index | Yes | Column name for row index | |
| columns | Yes | Column name for pivot columns | |
| values | Yes | Column name to aggregate | |
| aggfunc | No | Aggregation function | sum |
Implementation Reference
- Core handler function implementing pivot table logic with Polars: converts data to DataFrame, maps aggregation functions, performs pivot operation, formats result.async def pivot_table( data: Annotated[List[Dict[str, Union[str, float]]], Field(description="List of row dictionaries")], index: Annotated[str, Field(description="Column name for row index")], columns: Annotated[str, Field(description="Column name for pivot columns")], values: Annotated[str, Field(description="Column name to aggregate")], aggfunc: Annotated[Literal["sum", "mean", "count", "min", "max"], Field(description="Aggregation function")] = "sum", ) -> str: """Create pivot tables.""" try: df = pl.DataFrame(data) # Map aggfunc to Polars-compatible values agg_map = { "sum": "sum", "mean": "mean", "count": "len", # Polars uses "len" for count "min": "min", "max": "max", } if aggfunc not in agg_map: raise ValueError(f"Unknown aggregation function: {aggfunc}") # Polars pivot requires eager mode pivot_df = df.pivot( on=columns, index=index, values=values, aggregate_function=agg_map[aggfunc], # type: ignore[arg-type] ) # Convert to dict for JSON response result = pivot_df.to_dicts() return format_result( result, {"index": index, "columns": columns, "values": values, "aggfunc": aggfunc} ) except Exception as e: raise ValueError( f"Pivot table creation failed: {str(e)}. " f"Ensure data contains columns: {index}, {columns}, {values}" )
- src/vibe_math_mcp/tools/statistics.py:102-137 (registration)@mcp.tool decorator registering the pivot_table tool, providing name, detailed description with examples, and ToolAnnotations.@mcp.tool( name="pivot_table", description="""Create pivot tables from tabular data using Polars. Like Excel pivot tables: reshape data with row/column dimensions and aggregated values. Example: SALES BY REGION AND PRODUCT: data=[ {"region":"North","product":"A","sales":100}, {"region":"North","product":"B","sales":150}, {"region":"South","product":"A","sales":80}, {"region":"South","product":"B","sales":120} ], index="region", columns="product", values="sales", aggfunc="sum" Result: product | A | B --------|------|------ North | 100 | 150 South | 80 | 120 COUNT AGGREGATION: Same data with aggfunc="count" Result: Count of entries per region-product combination AVERAGE SCORES: data=[{"dept":"Sales","role":"Manager","score":85}, ...] index="dept", columns="role", values="score", aggfunc="mean" Result: Average scores by department and role""", annotations=ToolAnnotations( title="Pivot Table", readOnlyHint=True, idempotentHint=True, ), )
- Pydantic schema definitions for tool inputs using Annotated and Field for validation and descriptions.data: Annotated[List[Dict[str, Union[str, float]]], Field(description="List of row dictionaries")], index: Annotated[str, Field(description="Column name for row index")], columns: Annotated[str, Field(description="Column name for pivot columns")], values: Annotated[str, Field(description="Column name to aggregate")], aggfunc: Annotated[Literal["sum", "mean", "count", "min", "max"], Field(description="Aggregation function")] = "sum", ) -> str: