pivot_table
Create pivot tables to summarize and analyze tabular data by grouping rows and columns with aggregated values like sums, averages, or counts.
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 |
Input Schema (JSON Schema)
{
"properties": {
"aggfunc": {
"default": "sum",
"description": "Aggregation function",
"enum": [
"sum",
"mean",
"count",
"min",
"max"
],
"type": "string"
},
"columns": {
"description": "Column name for pivot columns",
"type": "string"
},
"context": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Optional annotation to label this calculation (e.g., 'Bond A PV', 'Q2 revenue'). Appears in results for easy identification."
},
"data": {
"description": "List of row dictionaries",
"items": {
"additionalProperties": {
"anyOf": [
{
"type": "string"
},
{
"type": "number"
}
]
},
"type": "object"
},
"type": "array"
},
"index": {
"description": "Column name for row index",
"type": "string"
},
"output_mode": {
"default": "full",
"description": "Output format: full (default), compact, minimal, value, or final. See batch_execute tool for details.",
"enum": [
"full",
"compact",
"minimal",
"value",
"final"
],
"type": "string"
},
"values": {
"description": "Column name to aggregate",
"type": "string"
}
},
"required": [
"values",
"columns",
"index",
"data"
],
"type": "object"
}
Implementation Reference
- The main handler function for the 'pivot_table' tool. It takes tabular data as a list of dictionaries, performs pivoting using Polars' pivot method with specified index, columns, values, and aggregation function (sum, mean, count, min, max), and returns a formatted 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)Registers the 'pivot_table' tool with the MCP server using the @mcp.tool decorator, specifying name, detailed description with examples, and tool annotations.@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 for 'pivot_table' tool inputs defined via Annotated types and Field descriptions in the function signature.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:
- TOOL_CATEGORIES dictionary includes 'pivot_table' under 'Statistics', used by batch_execute to build the tool registry for batch operations.TOOL_CATEGORIES = { "Basic": ["calculate", "percentage", "round", "convert_units"], "Arrays": ["array_operations", "array_statistics", "array_aggregate", "array_transform"], "Statistics": ["statistics", "pivot_table", "correlation"], "Financial": ["financial_calcs", "compound_interest", "perpetuity"], "Linear Algebra": ["matrix_operations", "solve_linear_system", "matrix_decomposition"], "Calculus": ["derivative", "integral", "limits_series"], }