change_column_type
Convert column data types in CSV files to int, float, str, bool, or datetime formats with error handling options.
Instructions
Change the data type of a column.
Returns: ColumnOperationResult with conversion details
Examples: # Convert string numbers to integers change_column_type(ctx, "age", "int")
# Convert to float, replacing errors with NaN
change_column_type(ctx, "price", "float", errors="coerce")
# Convert to datetime
change_column_type(ctx, "date", "datetime")
# Convert to boolean
change_column_type(ctx, "is_active", "bool")Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| column | Yes | Column name to change data type for | |
| dtype | Yes | Target data type (int, float, str, bool, datetime) | |
| errors | No | Error handling: 'raise' for errors, 'coerce' to replace invalid values with NaN | coerce |
Implementation Reference
- Main handler function for the 'change_column_type' tool. Validates column existence, maps string dtype to pandas dtype, performs type conversion using pd.to_datetime or astype/to_numeric with configurable error handling ('raise' or 'coerce'), and returns a ColumnOperationResult.
async def change_column_type( ctx: Annotated[Context, Field(description="FastMCP context for session access")], column: Annotated[str, Field(description="Column name to change data type for")], dtype: Annotated[ Literal["int", "float", "str", "bool", "datetime"], Field(description="Target data type (int, float, str, bool, datetime)"), ], errors: Annotated[ Literal["raise", "coerce"], Field( description="Error handling: 'raise' for errors, 'coerce' to replace invalid values with NaN", ), ] = "coerce", ) -> ColumnOperationResult: """Change the data type of a column. Returns: ColumnOperationResult with conversion details Examples: # Convert string numbers to integers change_column_type(ctx, "age", "int") # Convert to float, replacing errors with NaN change_column_type(ctx, "price", "float", errors="coerce") # Convert to datetime change_column_type(ctx, "date", "datetime") # Convert to boolean change_column_type(ctx, "is_active", "bool") """ # 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()) # Convert column type # Map string dtype to pandas dtype type_map = { "int": "int64", "float": "float64", "str": "string", "bool": "bool", "datetime": "datetime64[ns]", } target_dtype = type_map.get(dtype) if not target_dtype: msg = "dtype" raise InvalidParameterError(msg, dtype, f"Unsupported type: {dtype}") try: if dtype == "datetime": # Special handling for datetime conversion df[column] = pd.to_datetime(df[column], errors=errors) # General type conversion elif errors == "coerce": if dtype in ["int", "float"]: df[column] = pd.to_numeric(df[column], errors="coerce") else: df[column] = df[column].astype(target_dtype) # type: ignore[call-overload] else: df[column] = df[column].astype(target_dtype) # type: ignore[call-overload] except (ValueError, TypeError) as e: if errors == "raise": msg = "column" raise InvalidParameterError( msg, column, f"Cannot convert to {dtype}: {e}", ) from e # If errors='coerce', the conversion has already handled invalid values # Operation completed # No longer recording operations (simplified MCP architecture) return ColumnOperationResult( operation=f"change_type_to_{dtype}", rows_affected=len(df), columns_affected=[column], ) - src/databeak/servers/column_server.py:657-657 (registration)Registers the change_column_type handler function as an MCP tool named 'change_column_type' on the FastMCP column_server instance.
column_server.tool(name="change_column_type")(change_column_type) - Input schema defined via Annotated types in function signature: ctx (Context), column (str), dtype (Literal['int','float','str','bool','datetime']), errors (Literal['raise','coerce'] default 'coerce'). Output: ColumnOperationResult.
async def change_column_type( ctx: Annotated[Context, Field(description="FastMCP context for session access")], column: Annotated[str, Field(description="Column name to change data type for")], dtype: Annotated[ Literal["int", "float", "str", "bool", "datetime"], Field(description="Target data type (int, float, str, bool, datetime)"), ], errors: Annotated[ Literal["raise", "coerce"], Field( description="Error handling: 'raise' for errors, 'coerce' to replace invalid values with NaN", ), ] = "coerce", ) -> ColumnOperationResult: """Change the data type of a column. Returns: ColumnOperationResult with conversion details Examples: # Convert string numbers to integers change_column_type(ctx, "age", "int") # Convert to float, replacing errors with NaN change_column_type(ctx, "price", "float", errors="coerce") # Convert to datetime change_column_type(ctx, "date", "datetime") # Convert to boolean change_column_type(ctx, "is_active", "bool") """ # 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()) # Convert column type # Map string dtype to pandas dtype type_map = { "int": "int64", "float": "float64", "str": "string", "bool": "bool", "datetime": "datetime64[ns]", } target_dtype = type_map.get(dtype) if not target_dtype: msg = "dtype" raise InvalidParameterError(msg, dtype, f"Unsupported type: {dtype}") try: if dtype == "datetime": # Special handling for datetime conversion df[column] = pd.to_datetime(df[column], errors=errors) # General type conversion elif errors == "coerce": if dtype in ["int", "float"]: df[column] = pd.to_numeric(df[column], errors="coerce") else: df[column] = df[column].astype(target_dtype) # type: ignore[call-overload] else: df[column] = df[column].astype(target_dtype) # type: ignore[call-overload] except (ValueError, TypeError) as e: if errors == "raise": msg = "column" raise InvalidParameterError( msg, column, f"Cannot convert to {dtype}: {e}", ) from e # If errors='coerce', the conversion has already handled invalid values # Operation completed # No longer recording operations (simplified MCP architecture) return ColumnOperationResult( operation=f"change_type_to_{dtype}", rows_affected=len(df), columns_affected=[column], )