CSV Editor

Instructions

Implementation Reference

• src/csv_editor/tools/transformations.py:265-326 (handler)

  Core implementation of the add_column tool. Handles adding a new column with constant value, list of values, or computed via formula using pandas eval. Records the operation for history.

  async def add_column(
      session_id: str, 
      name: str,
      value: Any = None,
      formula: Optional[str] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """
      Add a new column to the dataframe.
      
      Args:
          session_id: Session identifier
          name: Name for the new column
          value: Default value for all rows (scalar or list)
          formula: Python expression to calculate values (e.g., "col1 + col2")
          ctx: FastMCP context
          
      Returns:
          Dict with success status
      """
      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 name in df.columns:
              return {"success": False, "error": f"Column '{name}' already exists"}
          
          if formula:
              # Evaluate formula in the context of the dataframe
              try:
                  session.df[name] = df.eval(formula)
              except Exception as e:
                  return {"success": False, "error": f"Formula evaluation failed: {str(e)}"}
          elif isinstance(value, list):
              if len(value) != len(df):
                  return {"success": False, "error": f"Value list length ({len(value)}) doesn't match row count ({len(df)})"}
              session.df[name] = value
          else:
              # Scalar value or None
              session.df[name] = value
          
          session.record_operation(OperationType.ADD_COLUMN, {
              "name": name,
              "value": str(value) if value is not None else None,
              "formula": formula
          })
          
          return {
              "success": True,
              "column_added": name,
              "columns": session.df.columns.tolist()
          }
          
      except Exception as e:
          logger.error(f"Error adding column: {str(e)}")
          return {"success": False, "error": str(e)}

• src/csv_editor/server.py:239-247 (registration)

  MCP tool registration for 'add_column'. Thin wrapper that delegates to the implementation in transformations.py.

  async def add_column(
      session_id: str,
      name: str,
      value: Any = None,
      formula: Optional[str] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """Add a new column to the dataframe."""
      return await _add_column(session_id, name, value, formula, ctx)

• src/csv_editor/server.py:239-247 (schema)

  Input schema defined by the function signature of the registered tool: session_id (str), name (str), value (Any optional), formula (Optional[str]), ctx (Context optional). Returns Dict[str, Any].

  async def add_column(
      session_id: str,
      name: str,
      value: Any = None,
      formula: Optional[str] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """Add a new column to the dataframe."""
      return await _add_column(session_id, name, value, formula, ctx)

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

  Import of the add_column implementation aliased as _add_column for use in server.py tool wrappers.

  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
  )