rekordbox-mcp

Instructions

Implementation Reference

• rekordbox_mcp/server.py:304-319 (handler)

  MCP tool handler for 'get_playlist_tracks'. Ensures database is initialized, fetches tracks from the database layer using the provided playlist_id, converts Track models to dictionaries via model_dump(), and returns the list. This is the entry point for the MCP tool.

  @mcp.tool()
  async def get_playlist_tracks(playlist_id: str) -> List[Dict[str, Any]]:
      """
      Get all tracks in a specific playlist.
      
      Args:
          playlist_id: The unique playlist identifier
          
      Returns:
          List of tracks in the playlist
      """
      if not db:
          raise RuntimeError("Database not initialized.")
      
      tracks = await db.get_playlist_tracks(playlist_id)
      return [track.model_dump() for track in tracks]

• rekordbox_mcp/models.py:13-55 (schema)

  Pydantic BaseModel defining the Track schema used in the tool's output. The handler returns a list of track.model_dump() which serializes to dicts matching this structure, providing input/output validation and type definitions for track metadata.

  class Track(BaseModel):
      """
      Rekordbox track model with comprehensive metadata.
      """
      
      id: str = Field(..., description="Unique track identifier")
      title: str = Field(..., description="Track title")
      artist: str = Field(..., description="Track artist")
      album: Optional[str] = Field(None, description="Album name")
      genre: Optional[str] = Field(None, description="Musical genre")
      bpm: float = Field(0.0, description="Beats per minute")
      key: Optional[str] = Field(None, description="Musical key (e.g., '5A', '12B')")
      rating: int = Field(0, ge=0, le=5, description="Track rating (0-5)")
      play_count: int = Field(0, ge=0, description="Number of times played")
      length: int = Field(0, ge=0, description="Track length in seconds")
      file_path: Optional[str] = Field(None, description="Path to audio file")
      date_added: Optional[str] = Field(None, description="Date track was added to library")
      date_modified: Optional[str] = Field(None, description="Date track was last modified")
      
      # Additional metadata
      bitrate: Optional[int] = Field(None, description="Audio bitrate in kbps")
      sample_rate: Optional[int] = Field(None, description="Audio sample rate in Hz")
      color: Optional[str] = Field(None, description="Track color tag")
      comments: Optional[str] = Field(None, description="Track comments")
      
      @field_validator('key')
      @classmethod
      def validate_key(cls, v):
          """Validate musical key format."""
          if v and v not in []:  # Add valid key formats
              # Basic validation - could be more sophisticated
              pass
          return v
      
      def duration_formatted(self) -> str:
          """Get track duration in MM:SS format."""
          if self.length <= 0:
              return "0:00"
          
          minutes = self.length // 60
          seconds = self.length % 60
          return f"{minutes}:{seconds:02d}"

• rekordbox_mcp/database.py:267-308 (helper)

  Core helper function in the database layer that implements the logic: queries pyrekordbox for playlist_songs by PlaylistID, filters active ones, matches ContentID to tracks via lookup, sorts by TrackNo to preserve order, converts to Track models using _content_to_track, returns list of Track objects.

  async def get_playlist_tracks(self, playlist_id: str) -> List[Track]:
      """
      Get all tracks in a specific playlist.
      
      Args:
          playlist_id: The playlist's unique identifier
          
      Returns:
          List of tracks in the playlist
      """
      if not self.db:
          raise RuntimeError("Database not connected")
      
      try:
          # Get song-playlist relationships for this playlist
          playlist_songs = list(self.db.get_playlist_songs(PlaylistID=int(playlist_id)))
          # Filter out soft-deleted relationships
          active_songs = [s for s in playlist_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]
          
          # Create a lookup for faster access
          content_lookup = {str(c.ID): c for c in active_content}
          
          # Build tracks list maintaining playlist order
          tracks = []
          # Sort by TrackNo to maintain playlist order
          sorted_songs = sorted(active_songs, key=lambda x: getattr(x, 'TrackNo', 0))
          
          for song_playlist in sorted_songs:
              content_id = str(song_playlist.ContentID)
              if content_id in content_lookup:
                  track = self._content_to_track(content_lookup[content_id])
                  tracks.append(track)
          
          return tracks
          
      except Exception as e:
          logger.error(f"Failed to get playlist tracks for playlist {playlist_id}: {e}")
          return []