update_column
Modify column values in CSV data using operations like replace, map, apply expressions, or fill missing values.
Instructions
Update values in a column using various operations with discriminated unions.
Returns: ColumnOperationResult with update details
Examples: # Using discriminated union - Replace operation update_column(ctx, "status", { "type": "replace", "pattern": "N/A", "replacement": "Unknown" })
# Using discriminated union - Map operation
update_column(ctx, "code", {
"type": "map",
"mapping": {"A": "Alpha", "B": "Beta"}
})
# Using discriminated union - Fill operation
update_column(ctx, "score", {
"type": "fillna",
"value": 0
})
# Legacy format still supported
update_column(ctx, "score", {
"operation": "fillna",
"value": 0
})Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| column | Yes | Column name to update values in | |
| operation | Yes | Update operation specification (replace, map, apply, fillna) |
Implementation Reference
- Main handler function for the 'update_column' tool. Handles update operations (replace, map, apply, fillna) on a specified DataFrame column using discriminated unions or legacy dict formats. Updates the session DataFrame in-place and returns operation result.
async def update_column( ctx: Annotated[Context, Field(description="FastMCP context for session access")], column: Annotated[str, Field(description="Column name to update values in")], operation: Annotated[ UpdateOperationType | UpdateColumnRequest | dict[str, Any], Field(description="Update operation specification (replace, map, apply, fillna)"), ], ) -> ColumnOperationResult: """Update values in a column using various operations with discriminated unions. Returns: ColumnOperationResult with update details Examples: # Using discriminated union - Replace operation update_column(ctx, "status", { "type": "replace", "pattern": "N/A", "replacement": "Unknown" }) # Using discriminated union - Map operation update_column(ctx, "code", { "type": "map", "mapping": {"A": "Alpha", "B": "Beta"} }) # Using discriminated union - Fill operation update_column(ctx, "score", { "type": "fillna", "value": 0 }) # Legacy format still supported update_column(ctx, "score", { "operation": "fillna", "value": 0 }) """ # Get session_id from FastMCP context session_id = ctx.session_id _session, df = get_session_data(session_id) if column not in df.columns: raise ColumnNotFoundError(column, df.columns.tolist()) # Track initial state # Update column operation (no longer tracking null count changes) operation_type = "unknown" # Handle discriminated union operations if isinstance( operation, ReplaceOperation | MapOperation | ApplyOperation | FillNaOperation, ): if isinstance(operation, ReplaceOperation): operation_type = "replace" df[column] = df[column].replace(operation.pattern, operation.replacement) elif isinstance(operation, MapOperation): operation_type = "map" df[column] = df[column].map(operation.mapping) elif isinstance(operation, ApplyOperation): operation_type = "apply" expr = operation.expression # Use unified secure evaluator for both string and mathematical expressions result = _apply_expression_to_column(expr, column, df, "expression") df[column] = result elif isinstance(operation, FillNaOperation): operation_type = "fillna" df[column] = df[column].fillna(operation.value) # Handle legacy format or dict input elif isinstance(operation, dict): if "type" in operation: # Parse as discriminated union if operation["type"] == "replace": replace_op = ReplaceOperation(**operation) operation_type = "replace" df[column] = df[column].replace( replace_op.pattern, replace_op.replacement, ) elif operation["type"] == "map": map_op = MapOperation(**operation) operation_type = "map" df[column] = df[column].map(map_op.mapping) elif operation["type"] == "apply": apply_op = ApplyOperation(**operation) operation_type = "apply" expr = apply_op.expression # Use unified secure evaluator for both string and mathematical expressions result = _apply_expression_to_column(expr, column, df, "expression") df[column] = result elif operation["type"] == "fillna": fillna_op = FillNaOperation(**operation) operation_type = "fillna" df[column] = df[column].fillna(fillna_op.value) else: msg = "type" raise InvalidParameterError( msg, operation["type"], "Supported types: replace, map, apply, fillna", ) else: # Legacy format with "operation" field update_request = UpdateColumnRequest(**operation) operation_type = update_request.operation if update_request.operation == "replace": if update_request.pattern is None or update_request.replacement is None: msg = "pattern/replacement" raise InvalidParameterError( msg, f"{update_request.pattern}/{update_request.replacement}", "Both pattern and replacement required for replace operation", ) df[column] = df[column].replace( update_request.pattern, update_request.replacement, ) elif update_request.operation == "map": if not isinstance(update_request.value, dict): msg = "value" raise InvalidParameterError( msg, str(update_request.value), "Dictionary mapping required for map operation", ) df[column] = df[column].map(update_request.value) elif update_request.operation == "apply": if update_request.value is None: msg = "value" raise InvalidParameterError( msg, str(update_request.value), "Expression required for apply operation", ) if isinstance(update_request.value, str): expr = update_request.value # Use unified secure evaluator for both string and mathematical expressions result = _apply_expression_to_column(expr, column, df, "value") df[column] = result else: df[column] = df[column].apply(update_request.value) elif update_request.operation == "fillna": if update_request.value is None: msg = "value" raise InvalidParameterError( msg, str(update_request.value), "Fill value required for fillna operation", ) df[column] = df[column].fillna(update_request.value) else: msg = "operation" raise InvalidParameterError( msg, update_request.operation, "Supported operations: replace, map, apply, fillna", ) else: # Handle legacy UpdateColumnRequest object update_request = operation operation_type = update_request.operation # ... (same logic as above legacy handling) # Operation completed # No longer recording operations (simplified MCP architecture) return ColumnOperationResult( operation=f"update_{operation_type}", rows_affected=len(df), columns_affected=[column], ) - Pydantic model defining the input schema for legacy update_column requests, supporting replace, map, apply, fillna operations.
class UpdateColumnRequest(BaseModel): """Request parameters for column update operations.""" model_config = ConfigDict(extra="forbid") operation: Literal["replace", "map", "apply", "fillna"] = Field( description="Type of update operation", ) value: Any | None = Field( # Any justified: operation-dependent type (CellValue|dict|str) None, description="Value for the operation (depends on operation type)", ) pattern: str | None = Field(None, description="Pattern for replace operation") replacement: str | None = Field(None, description="Replacement for replace operation") - Discriminated union type for modern update operations (ReplaceOperation, MapOperation, ApplyOperation, FillNaOperation), used in the tool handler.
UpdateOperationType = Annotated[ ReplaceOperation | MapOperation | ApplyOperation | FillNaOperation, Field(discriminator="type"), ] - src/databeak/servers/column_server.py:658-658 (registration)Registration of the update_column handler as an MCP tool on the column_server FastMCP instance.
column_server.tool(name="update_column")(update_column) - Helper function used in 'apply' operation to safely evaluate expressions on column values, referenced as 'x'.
def _apply_expression_to_column( expression: str, column: str, dataframe: pd.DataFrame, error_context: str = "expression", ) -> pd.Series: """Apply a string or mathematical expression to a column safely. Args: expression: The expression to evaluate (e.g., "x.upper()", "x + 10") column: The column name to use as 'x' in the expression dataframe: The DataFrame containing the column error_context: Context for error messages (e.g., "expression", "value") Returns: pd.Series: Result of the expression evaluation Raises: InvalidParameterError: If the expression is invalid or evaluation fails """ try: # Create column context mapping for 'x' variable column_context = {"x": column} # Use unified expression evaluator that handles both string and math operations return evaluate_string_expression_safely(expression, dataframe, column_context) except Exception as e: msg = error_context raise InvalidParameterError( msg, expression, f"Invalid expression. Use 'x' to reference column values. Error: {e}", ) from e