CSV Editor

Instructions

Implementation Reference

• src/csv_editor/tools/transformations.py:535-632 (handler)

  Core handler implementing column updates via pandas operations: replace (regex), extract, split, strip, upper, lower, fillna. Records operation and returns samples.

  async def update_column(
      session_id: str,
      column: str,
      operation: str,
      value: Optional[Any] = None,
      pattern: Optional[str] = None,
      replacement: Optional[str] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """
      Update values in a specific column with simple operations.
      
      Args:
          session_id: Session identifier
          column: Column name to update
          operation: Operation type - 'replace', 'extract', 'split', 'strip', 'upper', 'lower', 'fill'
          value: Value for certain operations (e.g., fill value)
          pattern: Pattern for replace/extract operations (regex supported)
          replacement: Replacement string for replace operation
          ctx: FastMCP context
          
      Returns:
          Dict with success status and update info
      """
      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
          
          if column not in df.columns:
              return {"success": False, "error": f"Column '{column}' not found"}
          
          original_values_sample = df[column].head(5).tolist()
          
          if operation == "replace":
              if pattern is None or replacement is None:
                  return {"success": False, "error": "Pattern and replacement required for replace operation"}
              session.df[column] = df[column].astype(str).str.replace(pattern, replacement, regex=True)
              
          elif operation == "extract":
              if pattern is None:
                  return {"success": False, "error": "Pattern required for extract operation"}
              session.df[column] = df[column].astype(str).str.extract(pattern, expand=False)
              
          elif operation == "split":
              if pattern is None:
                  pattern = " "
              if value is not None and isinstance(value, int):
                  # Extract specific part after split
                  session.df[column] = df[column].astype(str).str.split(pattern).str[value]
              else:
                  # Just do the split, take first part
                  session.df[column] = df[column].astype(str).str.split(pattern).str[0]
                  
          elif operation == "strip":
              session.df[column] = df[column].astype(str).str.strip()
              
          elif operation == "upper":
              session.df[column] = df[column].astype(str).str.upper()
              
          elif operation == "lower":
              session.df[column] = df[column].astype(str).str.lower()
              
          elif operation == "fill":
              if value is None:
                  return {"success": False, "error": "Value required for fill operation"}
              session.df[column] = df[column].fillna(value)
              
          else:
              return {"success": False, "error": f"Unknown operation: {operation}"}
          
          updated_values_sample = session.df[column].head(5).tolist()
          
          session.record_operation(OperationType.UPDATE_COLUMN, {
              "column": column,
              "operation": operation,
              "pattern": pattern,
              "replacement": replacement,
              "value": str(value) if value is not None else None
          })
          
          return {
              "success": True,
              "column": column,
              "operation": operation,
              "original_sample": original_values_sample,
              "updated_sample": updated_values_sample,
              "rows_updated": len(session.df)
          }
          
      except Exception as e:
          logger.error(f"Error updating column: {str(e)}")
          return {"success": False, "error": str(e)}

• src/csv_editor/server.py:290-302 (registration)

  MCP tool registration: @mcp.tool decorator wraps and exposes the update_column handler from transformations.py

  @mcp.tool
  async def update_column(
      session_id: str,
      column: str,
      operation: str,
      value: Optional[Any] = None,
      pattern: Optional[str] = None,
      replacement: Optional[str] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """Update values in a specific column with simple operations like replace, extract, split, etc."""
      return await _update_column(session_id, column, operation, value, pattern, replacement, ctx)

• src/csv_editor/models/data_models.py:31-31 (helper)

  OperationType enum value used for recording update_column operations in session history.

  UPDATE_COLUMN = "update_column"

• src/csv_editor/server.py:198-198 (helper)

  Import of the update_column handler from transformations.py for use in server registration.

  update_column as _update_column