MCP Tabular Data Analysis Server

Instructions

Export filtered/transformed data to a new CSV file.

Args:
    file_path: Path to source CSV or SQLite file
    output_name: Name for output file (without extension, saved to data/ folder)
    filter_column: Optional column to filter on
    filter_operator: Filter operator - 'eq', 'ne', 'gt', 'gte', 'lt', 'lte', 'contains'
    filter_value: Value to filter by
    columns: List of columns to include (default: all)
    sort_by: Column to sort by
    sort_ascending: Sort direction (default: ascending)
    limit: Maximum rows to export

Returns:
    Dictionary containing export details and file path

Implementation Reference

• src/mcp_tabular/server.py:1550-1641 (handler)

  The primary handler function for the 'export_data' tool. It loads data from a file, optionally applies filters, column selection, sorting, and row limits, then exports the processed data to a timestamped CSV file in the data/ directory. Returns details about the export operation.

  def export_data(
      file_path: str,
      output_name: str,
      filter_column: str | None = None,
      filter_operator: str | None = None,
      filter_value: str | float | None = None,
      columns: list[str] | None = None,
      sort_by: str | None = None,
      sort_ascending: bool = True,
      limit: int | None = None,
  ) -> dict[str, Any]:
      """
      Export filtered/transformed data to a new CSV file.
      
      Args:
          file_path: Path to source CSV or SQLite file
          output_name: Name for output file (without extension, saved to data/ folder)
          filter_column: Optional column to filter on
          filter_operator: Filter operator - 'eq', 'ne', 'gt', 'gte', 'lt', 'lte', 'contains'
          filter_value: Value to filter by
          columns: List of columns to include (default: all)
          sort_by: Column to sort by
          sort_ascending: Sort direction (default: ascending)
          limit: Maximum rows to export
      
      Returns:
          Dictionary containing export details and file path
      """
      df = _load_data(file_path)
      original_count = len(df)
      
      # Apply filter if specified
      if filter_column and filter_operator and filter_value is not None:
          if filter_column not in df.columns:
              raise ValueError(f"Filter column '{filter_column}' not found")
          
          if filter_operator == "eq":
              df = df[df[filter_column] == filter_value]
          elif filter_operator == "ne":
              df = df[df[filter_column] != filter_value]
          elif filter_operator == "gt":
              df = df[df[filter_column] > float(filter_value)]
          elif filter_operator == "gte":
              df = df[df[filter_column] >= float(filter_value)]
          elif filter_operator == "lt":
              df = df[df[filter_column] < float(filter_value)]
          elif filter_operator == "lte":
              df = df[df[filter_column] <= float(filter_value)]
          elif filter_operator == "contains":
              df = df[df[filter_column].astype(str).str.contains(str(filter_value), case=False, na=False)]
          else:
              raise ValueError(f"Unknown operator: {filter_operator}")
      
      # Select columns
      if columns:
          invalid = [c for c in columns if c not in df.columns]
          if invalid:
              raise ValueError(f"Columns not found: {invalid}")
          df = df[columns]
      
      # Sort
      if sort_by:
          if sort_by not in df.columns:
              raise ValueError(f"Sort column '{sort_by}' not found")
          df = df.sort_values(sort_by, ascending=sort_ascending)
      
      # Limit rows
      if limit:
          df = df.head(limit)
      
      # Save file
      output_dir = _PROJECT_ROOT / 'data'
      output_dir.mkdir(parents=True, exist_ok=True)
      
      # Clean output name and add timestamp
      clean_name = "".join(c for c in output_name if c.isalnum() or c in ('-', '_'))
      timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
      output_file = output_dir / f"{clean_name}_{timestamp}.csv"
      
      df.to_csv(output_file, index=False)
      
      return {
          "success": True,
          "source_file": file_path,
          "output_file": str(output_file.relative_to(_PROJECT_ROOT)),
          "absolute_path": str(output_file),
          "original_rows": original_count,
          "exported_rows": len(df),
          "exported_columns": df.columns.tolist(),
          "filter_applied": f"{filter_column} {filter_operator} {filter_value}" if filter_column else None,
      }