MCP Tabular Data Analysis Server

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"
    )

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,
          }
      }