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)