sort_data
Sort CSV data by specified columns to organize information for analysis or presentation.
Instructions
Sort data by columns.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| session_id | Yes | ||
| columns | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- Core handler function that implements the sorting logic using pandas DataFrame.sort_values, parses column specifications, validates columns, updates the session dataframe, and records the operation.
async def sort_data( session_id: str, columns: List[Union[str, Dict[str, str]]], ctx: Context = None ) -> Dict[str, Any]: """ Sort data by one or more columns. Args: session_id: Session identifier columns: List of column names or dicts with 'column' and 'ascending' keys 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 # Parse columns into names and ascending flags sort_columns = [] ascending = [] for col in columns: if isinstance(col, str): sort_columns.append(col) ascending.append(True) elif isinstance(col, dict): sort_columns.append(col["column"]) ascending.append(col.get("ascending", True)) else: return {"success": False, "error": f"Invalid column specification: {col}"} # Validate columns exist for col in sort_columns: if col not in df.columns: return {"success": False, "error": f"Column '{col}' not found"} session.df = df.sort_values(by=sort_columns, ascending=ascending).reset_index(drop=True) session.record_operation(OperationType.SORT, { "columns": sort_columns, "ascending": ascending }) return { "success": True, "sorted_by": sort_columns, "ascending": ascending } except Exception as e: logger.error(f"Error sorting data: {str(e)}") return {"success": False, "error": str(e)} - src/csv_editor/server.py:211-218 (registration)MCP tool registration decorator (@mcp.tool) and wrapper function that delegates to the core _sort_data implementation from transformations module.
@mcp.tool async def sort_data( session_id: str, columns: List[Any], ctx: Context = None ) -> Dict[str, Any]: """Sort data by columns.""" return await _sort_data(session_id, columns, ctx)