rekordbox-mcp

Instructions

Implementation Reference

• rekordbox_mcp/server.py:388-402 (handler)

  MCP tool handler function decorated with @mcp.tool(). Ensures database connection, calls the database helper to fetch session tracks, and serializes them to dictionaries using model_dump().
  @mcp.tool() async def get_session_tracks(session_id: str) -> List[Dict[str, Any]]: """ Get all tracks from a specific DJ history session. Args: session_id: The session's unique identifier Returns: List of tracks in the session with performance context """ await ensure_database_connected() tracks = await db.get_session_tracks(session_id) return [track.model_dump() for track in tracks]
• rekordbox_mcp/models.py:116-135 (schema)

  Pydantic BaseModel defining the structure of each HistoryTrack object returned by the database helper and serialized in the tool handler.
  class HistoryTrack(BaseModel): """ Track within a DJ history session with performance context. """ # Track basic info (from Track model) id: str = Field(..., description="Track identifier") title: str = Field("", description="Track title") artist: str = Field("", description="Artist name") album: Optional[str] = Field(None, description="Album name") genre: Optional[str] = Field(None, description="Genre") bpm: float = Field(0.0, ge=0, description="Beats per minute") key: Optional[str] = Field(None, description="Musical key") length: int = Field(0, ge=0, description="Track length in seconds") # History-specific context track_number: int = Field(..., ge=1, description="Position in DJ set") history_id: str = Field(..., description="History session ID") play_order: Optional[int] = Field(None, description="Order played in session")
• rekordbox_mcp/database.py:606-665 (helper)

  Database class method that implements the core logic: queries history_songs and content tables from pyrekordbox database, matches tracks, constructs HistoryTrack objects preserving play order.
  async def get_session_tracks(self, session_id: str) -> List[HistoryTrack]: """ Get all tracks from a specific DJ history session. Args: session_id: The session's unique identifier Returns: List of tracks in the session with performance context """ if not self.db: raise RuntimeError("Database not connected") try: # Get songs for this session history_songs = list(self.db.get_history_songs(HistoryID=int(session_id))) active_songs = [s for s in history_songs if getattr(s, 'rb_local_deleted', 0) == 0] # Get all content to match against all_content = list(self.db.get_content()) active_content = [c for c in all_content if getattr(c, 'rb_local_deleted', 0) == 0] content_lookup = {str(c.ID): c for c in active_content} # Build tracks list maintaining session order tracks = [] sorted_songs = sorted(active_songs, key=lambda x: x.TrackNo) for song in sorted_songs: content_id = str(song.ContentID) if content_id in content_lookup: content = content_lookup[content_id] # Extract track info using same logic as _content_to_track bmp_value = getattr(content, 'BPM', 0) or 0 bpm_float = float(bmp_value) / 100.0 if bmp_value else 0.0 artist_name = getattr(content, 'ArtistName', '') or "" album_name = getattr(content, 'AlbumName', '') or "" genre_name = getattr(content, 'GenreName', '') or "" key_name = getattr(content, 'KeyName', '') or "" tracks.append(HistoryTrack( id=str(content.ID), title=content.Title or "", artist=artist_name, album=album_name, genre=genre_name, bpm=bpm_float, key=key_name, length=int(getattr(content, 'Length', 0) or 0), track_number=song.TrackNo, history_id=session_id, play_order=song.TrackNo )) return tracks except Exception as e: logger.error(f"Failed to get session tracks for session {session_id}: {e}") return []