add_column

Instructions

Implementation Reference

• src/csv_editor/tools/transformations.py:247-301 (handler)

  Core implementation of the add_column tool. Adds a new column to the dataframe, supporting scalar values, lists, and formula-based calculations via df.eval(). Records the operation in session history.

  async def add_column(
      session_id: str, name: str, value: Any = None, formula: str | None = 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: {e!s}"}
          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: {e!s}")
          return {"success": False, "error": str(e)}

• src/csv_editor/server.py:227-232 (registration)

  MCP tool registration decorator (@mcp.tool) for add_column. Defines the tool's interface (session_id, name, value, formula, ctx) and delegates to the handler in transformations.py.

  @mcp.tool
  async def add_column(
      session_id: str, name: str, value: Any = None, formula: str | None = 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/models/data_models.py:33-33 (schema)

  Enum entry ADD_COLUMN = 'add_column' in OperationType, used for operation history tracking.

  ADD_COLUMN = "add_column"

• src/csv_editor/tools/transformations.py:292-295 (helper)

  Records the add_column operation in session history with the operation name, column name, value, and formula.

  session.record_operation(
      OperationType.ADD_COLUMN,
      {"name": name, "value": str(value) if value is not None else None, "formula": formula},
  )