sort_data
Organize CSV data by selecting specific columns for sorting. Designed for efficient data manipulation within the CSV Editor MCP server, supporting large files and complex operations.
Instructions
Sort data by columns.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| columns | Yes | ||
| session_id | Yes |
Implementation Reference
- Core handler function that executes the sorting logic: parses sort columns and directions, validates columns, sorts the pandas DataFrame, updates the session, records the operation, and returns success status.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:212-219 (registration)Registers the 'sort_data' tool using the @mcp.tool decorator. This wrapper function delegates to the internal implementation (_sort_data). Defines the tool's input schema via type hints.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)