rekordbox-mcp

Instructions

Implementation Reference

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

  MCP tool handler that fetches history sessions from the database, sorts them by creation date (most recent first), applies the limit, and serializes to dictionaries using model_dump().
  @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/models.py:94-114 (schema)

  Pydantic BaseModel defining the output structure for HistorySession objects used 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
• rekordbox_mcp/database.py:540-604 (helper)

  Database class method that implements the core logic for retrieving DJ history sessions and folders using pyrekordbox, filtering deleted entries, calculating track counts and total durations.
  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 []