rekordbox-mcp

Instructions

Implementation Reference

• rekordbox_mcp/server.py:365-385 (handler)

  MCP tool handler for get_history_sessions. The @mcp.tool() decorator registers this function as the tool implementation. It ensures database connection, fetches sessions from database helper, sorts them, applies limit, and returns serialized data.

  @mcp.tool()
  async def get_history_sessions(
      include_folders: bool = False,
      limit: int = 100
  ) -> List[Dict[str, Any]]:
      """
      Get DJ history sessions from rekordbox.
      
      Args:
          include_folders: Whether to include folder entries (years/months)
          limit: Maximum number of sessions to return
          
      Returns:
          List of history sessions with metadata
      """
      await ensure_database_connected()
      
      sessions = await db.get_history_sessions(include_folders=include_folders)
      # Sort by date created, most recent first
      sessions.sort(key=lambda x: x.date_created or "", reverse=True)
      return [session.model_dump() for session in sessions[:limit]]

• rekordbox_mcp/database.py:540-604 (helper)

  Core database method that queries pyrekordbox database for history sessions, filters active ones, calculates track counts and durations, and constructs HistorySession models.

  async def get_history_sessions(self, include_folders: bool = False) -> List[HistorySession]:
      """
      Get all DJ history sessions from the database.
      
      Args:
          include_folders: Whether to include folder entries
          
      Returns:
          List of history sessions
      """
      if not self.db:
          raise RuntimeError("Database not connected")
      
      try:
          # Get all histories, filtering out soft-deleted ones
          all_histories = list(self.db.get_history())
          active_histories = [h for h in all_histories if getattr(h, 'rb_local_deleted', 0) == 0]
          
          sessions = []
          for history in active_histories:
              # Filter by type: Attribute 1 = folder, Attribute 0 = session
              is_folder = history.Attribute == 1
              
              if not include_folders and is_folder:
                  continue
              
              # Get track count for sessions
              track_count = 0
              duration_minutes = None
              if not is_folder:
                  try:
                      history_songs = list(self.db.get_history_songs(HistoryID=history.ID))
                      active_songs = [s for s in history_songs if getattr(s, 'rb_local_deleted', 0) == 0]
                      track_count = len(active_songs)
                      
                      # Calculate duration if we have tracks
                      if active_songs:
                          all_content = list(self.db.get_content())
                          content_lookup = {str(c.ID): c for c in all_content if getattr(c, 'rb_local_deleted', 0) == 0}
                          
                          total_seconds = 0
                          for song in active_songs:
                              content_id = str(song.ContentID)
                              if content_id in content_lookup:
                                  track_length = getattr(content_lookup[content_id], 'Length', 0) or 0
                                  total_seconds += track_length
                          duration_minutes = round(total_seconds / 60) if total_seconds > 0 else None
                  except Exception:
                      track_count = 0
              
              sessions.append(HistorySession(
                  id=str(history.ID),
                  name=history.Name or "",
                  parent_id=str(history.ParentID) if history.ParentID and history.ParentID != "root" else None,
                  is_folder=is_folder,
                  date_created=history.DateCreated,
                  track_count=track_count,
                  duration_minutes=duration_minutes
              ))
          
          return sessions
          
      except Exception as e:
          logger.error(f"Failed to get history sessions: {e}")
          return []

• rekordbox_mcp/models.py:94-113 (schema)

  Pydantic BaseModel defining the output structure for HistorySession objects returned by the tool.

  class HistorySession(BaseModel):
      """
      Rekordbox DJ history session model.
      """
      
      id: str = Field(..., description="Unique history session identifier")
      name: str = Field(..., description="Session name (usually date)")
      parent_id: Optional[str] = Field(None, description="Parent folder ID")
      is_folder: bool = Field(False, description="Whether this is a folder")
      date_created: Optional[str] = Field(None, description="Date session was created")
      track_count: int = Field(0, ge=0, description="Number of tracks in session")
      duration_minutes: Optional[int] = Field(None, description="Total session duration in minutes")
      
      @field_validator('date_created', mode='before')
      @classmethod
      def validate_date(cls, v):
          """Convert datetime objects to strings."""
          if hasattr(v, 'strftime'):  # datetime object
              return v.strftime("%Y-%m-%d %H:%M:%S")
          return str(v) if v is not None else None