MCP Tabular Data Analysis Server

Instructions

Filter rows based on a condition.

Args:
    file_path: Path to CSV or SQLite file
    column: Column name to filter on
    operator: Comparison operator - 'eq', 'ne', 'gt', 'gte', 'lt', 'lte', 'contains', 'startswith', 'endswith'
    value: Value to compare against
    limit: Maximum number of rows to return (default 100)

Returns:
    Dictionary containing:
    - filter_applied: Description of the filter
    - original_count: Number of rows before filtering
    - filtered_count: Number of rows after filtering
    - rows: Filtered rows (up to limit)

Implementation Reference

• src/mcp_tabular/server.py:314-378 (handler)

  The filter_rows tool handler function. Loads data using _load_data helper, applies conditional filtering based on operator and value, and returns filtered rows preview with counts.

  def filter_rows(
      file_path: str,
      column: str,
      operator: str,
      value: str | float | int,
      limit: int = 100,
  ) -> dict[str, Any]:
      """
      Filter rows based on a condition.
      
      Args:
          file_path: Path to CSV or SQLite file
          column: Column name to filter on
          operator: Comparison operator - 'eq', 'ne', 'gt', 'gte', 'lt', 'lte', 'contains', 'startswith', 'endswith'
          value: Value to compare against
          limit: Maximum number of rows to return (default 100)
      
      Returns:
          Dictionary containing:
          - filter_applied: Description of the filter
          - original_count: Number of rows before filtering
          - filtered_count: Number of rows after filtering
          - rows: Filtered rows (up to limit)
      """
      df = _load_data(file_path)
      
      if column not in df.columns:
          raise ValueError(f"Column '{column}' not found. Available: {df.columns.tolist()}")
      
      original_count = len(df)
      
      # Apply filter based on operator
      if operator == "eq":
          mask = df[column] == value
      elif operator == "ne":
          mask = df[column] != value
      elif operator == "gt":
          mask = df[column] > float(value)
      elif operator == "gte":
          mask = df[column] >= float(value)
      elif operator == "lt":
          mask = df[column] < float(value)
      elif operator == "lte":
          mask = df[column] <= float(value)
      elif operator == "contains":
          mask = df[column].astype(str).str.contains(str(value), case=False, na=False)
      elif operator == "startswith":
          mask = df[column].astype(str).str.startswith(str(value), na=False)
      elif operator == "endswith":
          mask = df[column].astype(str).str.endswith(str(value), na=False)
      else:
          raise ValueError(
              f"Unknown operator: {operator}. Use: eq, ne, gt, gte, lt, lte, contains, startswith, endswith"
          )
      
      filtered_df = df[mask]
      
      return {
          "filter_applied": f"{column} {operator} {value}",
          "original_count": original_count,
          "filtered_count": len(filtered_df),
          "rows": filtered_df.head(limit).to_dict(orient="records"),
          "truncated": len(filtered_df) > limit,
      }