sort_data
Sort CSV data by one or multiple columns with flexible direction control. Handles mixed data types while maintaining data integrity during the sorting process.
Instructions
Sort data by one or more columns with comprehensive error handling.
Provides flexible sorting capabilities with support for multiple columns and sort directions. Handles mixed data types appropriately and maintains data integrity throughout the sorting process.
Examples: # Simple single column sort sort_data(ctx, ["age"])
# Multi-column sort with different directions
sort_data(ctx, [
{"column": "department", "ascending": True},
{"column": "salary", "ascending": False}
])
# Using SortColumn objects for type safety
sort_data(ctx, [
SortColumn(column="name", ascending=True),
SortColumn(column="age", ascending=False)
])
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| columns | Yes | Column specifications for sorting (strings or SortColumn objects) |
Implementation Reference
- The core handler function that executes the sorting logic. Parses column specifications, validates columns exist, sorts the dataframe using pandas.sort_values(), updates the session dataframe, and returns a SortDataResult.def sort_data( ctx: Annotated[Context, Field(description="FastMCP context for session access")], columns: Annotated[ list[str | SortColumn], Field(description="Column specifications for sorting (strings or SortColumn objects)"), ], ) -> SortDataResult: """Sort data by one or more columns with comprehensive error handling. Provides flexible sorting capabilities with support for multiple columns and sort directions. Handles mixed data types appropriately and maintains data integrity throughout the sorting process. Examples: # Simple single column sort sort_data(ctx, ["age"]) # Multi-column sort with different directions sort_data(ctx, [ {"column": "department", "ascending": True}, {"column": "salary", "ascending": False} ]) # Using SortColumn objects for type safety sort_data(ctx, [ SortColumn(column="name", ascending=True), SortColumn(column="age", ascending=False) ]) """ session_id = ctx.session_id session, df = get_session_data(session_id) # Parse columns into names and ascending flags sort_columns: list[str] = [] ascending: list[bool] = [] for col in columns: if isinstance(col, str): sort_columns.append(col) ascending.append(True) elif isinstance(col, SortColumn): sort_columns.append(col.column) ascending.append(col.ascending) elif isinstance(col, dict) and "column" in col: sort_columns.append(col["column"]) ascending.append(col.get("ascending", True)) else: msg = f"Invalid column specification: {col}" raise ToolError(msg) # Validate all columns exist missing_cols = [col for col in sort_columns if col not in df.columns] if missing_cols: msg = f"Columns not found: {missing_cols}" raise ToolError(msg) # Perform sort session.df = df.sort_values(by=sort_columns, ascending=ascending).reset_index(drop=True) # No longer recording operations (simplified MCP architecture) return SortDataResult( sorted_by=sort_columns, ascending=ascending, rows_processed=len(df), )
- src/databeak/servers/transformation_server.py:414-424 (registration)Registers the sort_data handler as an MCP tool on the transformation_server FastMCP instance using the @tool decorator with explicit name="sort_data".# Create transformation server transformation_server = FastMCP( "DataBeak-Transformation", instructions="Core data transformation server for filtering, sorting, deduplication, and missing value handling", ) # Register the logic functions directly as MCP tools (no wrapper functions needed) transformation_server.tool(name="filter_rows")(filter_rows) transformation_server.tool(name="sort_data")(sort_data) transformation_server.tool(name="remove_duplicates")(remove_duplicates) transformation_server.tool(name="fill_missing_values")(fill_missing_values)
- Pydantic model defining the input schema for sort columns, allowing specification of column name and sort direction.class SortColumn(BaseModel): """Column specification for sorting.""" model_config = ConfigDict(extra="forbid") column: str = Field(description="Column name to sort by") ascending: bool = Field(default=True, description="Sort in ascending order")
- Pydantic model for the output response of the sort_data tool, including sorted columns, directions, and rows processed.class SortDataResult(BaseToolResponse): """Response model for data sorting operations.""" sorted_by: list[str] = Field(description="Column names used for sorting") ascending: list[bool] = Field( description="Sort direction for each column (True=ascending, False=descending)", ) rows_processed: int = Field(description="Number of rows that were sorted")