CSV Editor

Instructions

Implementation Reference

• src/csv_editor/tools/transformations.py:110-169 (handler)

  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)