CSV Editor

Instructions

Implementation Reference

• src/csv_editor/tools/transformations.py:15-107 (handler)

  The primary handler function implementing the filter_rows tool logic. It retrieves the session DataFrame, constructs boolean masks for each condition using various operators (==, >, contains, etc.), combines them with AND/OR logic, filters the DataFrame in-place, records the operation, and returns filtering statistics.

  async def filter_rows(
      session_id: str, 
      conditions: List[Dict[str, Any]], 
      mode: str = "and",
      ctx: Context = None
  ) -> Dict[str, Any]:
      """
      Filter rows based on conditions.
      
      Args:
          session_id: Session identifier
          conditions: List of filter conditions, each with:
              - column: Column name
              - operator: One of '==', '!=', '>', '<', '>=', '<=', 'contains', 'starts_with', 'ends_with', 'in', 'not_in', 'is_null', 'not_null'
              - value: Value to compare (not needed for is_null/not_null)
          mode: 'and' or 'or' to combine multiple conditions
          ctx: FastMCP context
          
      Returns:
          Dict with success status and filtered row count
      """
      try:
          manager = get_session_manager()
          session = manager.get_session(session_id)
          
          if not session or session.df is None:
              return {"success": False, "error": "Invalid session or no data loaded"}
          
          df = session.df
          mask = pd.Series([True] * len(df))
          
          for condition in conditions:
              column = condition.get("column")
              operator = condition.get("operator")
              value = condition.get("value")
              
              if column not in df.columns:
                  return {"success": False, "error": f"Column '{column}' not found"}
              
              col_data = df[column]
              
              if operator == "==":
                  condition_mask = col_data == value
              elif operator == "!=":
                  condition_mask = col_data != value
              elif operator == ">":
                  condition_mask = col_data > value
              elif operator == "<":
                  condition_mask = col_data < value
              elif operator == ">=":
                  condition_mask = col_data >= value
              elif operator == "<=":
                  condition_mask = col_data <= value
              elif operator == "contains":
                  condition_mask = col_data.astype(str).str.contains(str(value), na=False)
              elif operator == "starts_with":
                  condition_mask = col_data.astype(str).str.startswith(str(value), na=False)
              elif operator == "ends_with":
                  condition_mask = col_data.astype(str).str.endswith(str(value), na=False)
              elif operator == "in":
                  condition_mask = col_data.isin(value if isinstance(value, list) else [value])
              elif operator == "not_in":
                  condition_mask = ~col_data.isin(value if isinstance(value, list) else [value])
              elif operator == "is_null":
                  condition_mask = col_data.isna()
              elif operator == "not_null":
                  condition_mask = col_data.notna()
              else:
                  return {"success": False, "error": f"Unknown operator: {operator}"}
              
              if mode == "and":
                  mask = mask & condition_mask
              else:
                  mask = mask | condition_mask
          
          session.df = df[mask].reset_index(drop=True)
          session.record_operation(OperationType.FILTER, {
              "conditions": conditions,
              "mode": mode,
              "rows_before": len(df),
              "rows_after": len(session.df)
          })
          
          return {
              "success": True,
              "rows_before": len(df),
              "rows_after": len(session.df),
              "rows_filtered": len(df) - len(session.df)
          }
          
      except Exception as e:
          logger.error(f"Error filtering rows: {str(e)}")
          return {"success": False, "error": str(e)}

• src/csv_editor/server.py:188-210 (registration)

  Tool registration in the main MCP server. Imports the handler from transformations.py as _filter_rows and registers a wrapper function with @mcp.tool decorator, which delegates to the core implementation.

  from .tools.transformations import (
      filter_rows as _filter_rows,
      sort_data as _sort_data,
      select_columns as _select_columns,
      rename_columns as _rename_columns,
      add_column as _add_column,
      remove_columns as _remove_columns,
      change_column_type as _change_column_type,
      fill_missing_values as _fill_missing_values,
      remove_duplicates as _remove_duplicates,
      update_column as _update_column
  )
  
  @mcp.tool
  async def filter_rows(
      session_id: str,
      conditions: List[Dict[str, Any]],
      mode: str = "and",
      ctx: Context = None
  ) -> Dict[str, Any]:
      """Filter rows based on conditions."""
      return await _filter_rows(session_id, conditions, mode, ctx)