CSV Editor

Instructions

Implementation Reference

• src/csv_editor/server.py:517-523 (registration)

  MCP tool registration for 'get_history'. This thin wrapper registers the tool with FastMCP and delegates to the main handler in history_operations.py.

  async def get_history(
      session_id: str,
      limit: Optional[int] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """Get operation history for a session."""
      return await _get_operation_history(session_id, limit, ctx)

• src/csv_editor/tools/history_operations.py:129-183 (handler)

  Primary handler logic for the get_history tool. Retrieves the session, calls session.get_history(limit), handles errors, logs via ctx, and returns OperationResult.

  async def get_operation_history(
      session_id: str,
      limit: Optional[int] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """
      Get operation history for a session.
      
      Args:
          session_id: Session identifier
          limit: Maximum number of operations to return
          ctx: FastMCP context
          
      Returns:
          Dict with history and statistics
      """
      try:
          manager = get_session_manager()
          session = manager.get_session(session_id)
          
          if not session:
              return OperationResult(
                  success=False,
                  message="Session not found",
                  error=f"No session with ID: {session_id}"
              ).model_dump()
          
          if ctx:
              await ctx.info(f"Getting operation history for session {session_id}")
          
          result = session.get_history(limit)
          
          if result["success"]:
              return OperationResult(
                  success=True,
                  message="History retrieved successfully",
                  session_id=session_id,
                  data=result
              ).model_dump()
          else:
              return OperationResult(
                  success=False,
                  message="Failed to get history",
                  error=result.get("error")
              ).model_dump()
              
      except Exception as e:
          logger.error(f"Error getting history: {str(e)}")
          if ctx:
              await ctx.error(f"Failed to get history: {str(e)}")
          return OperationResult(
              success=False,
              message="Failed to get history",
              error=str(e)
          ).model_dump()

• src/csv_editor/models/csv_session.py:284-306 (helper)

  CSVSession.get_history method called by the handler. Delegates to history_manager.get_history and adds statistics, or falls back to legacy history.

  def get_history(self, limit: Optional[int] = None) -> Dict[str, Any]:
      """Get operation history."""
      if not self.history_manager:
          # Return legacy history if new history is not enabled
          return {
              "success": True,
              "history": self.operations_history[-limit:] if limit else self.operations_history,
              "total": len(self.operations_history)
          }
      
      try:
          history = self.history_manager.get_history(limit)
          stats = self.history_manager.get_statistics()
          
          return {
              "success": True,
              "history": history,
              "statistics": stats
          }
      except Exception as e:
          logger.error(f"Error getting history: {str(e)}")
          return {"success": False, "error": str(e)}

• src/csv_editor/models/history_manager.py:301-315 (helper)

  Core HistoryManager.get_history implementation that slices recent history entries and enriches with index, current status, and restore capability.

  def get_history(self, limit: Optional[int] = None) -> List[Dict[str, Any]]:
      """Get operation history."""
      history_list = []
      
      start = 0 if limit is None else max(0, len(self.history) - limit)
      
      for i, entry in enumerate(self.history[start:], start=start):
          history_dict = entry.to_dict()
          history_dict["index"] = i
          history_dict["is_current"] = (i == self.current_index)
          history_dict["can_restore"] = entry.data_snapshot is not None
          history_list.append(history_dict)
      
      return history_list