rekordbox-mcp

Instructions

Implementation Reference

• rekordbox_mcp/server.py:44-88 (handler)

  MCP tool handler for 'search_tracks', decorated with @mcp.tool(). Constructs SearchOptions from parameters and delegates to RekordboxDatabase.search_tracks method, returning serialized track data.

  @mcp.tool()
  async def search_tracks(
      query: str = "",
      artist: Optional[str] = None,
      title: Optional[str] = None,
      genre: Optional[str] = None,
      key: Optional[str] = None,
      bpm_min: Optional[float] = None,
      bpm_max: Optional[float] = None,
      rating_min: Optional[int] = None,
      limit: int = 50
  ) -> List[Dict[str, Any]]:
      """
      Search tracks in the rekordbox database.
      
      Args:
          query: General search query (searches across multiple fields)
          artist: Filter by artist name
          title: Filter by track title
          genre: Filter by genre
          key: Filter by musical key (e.g., "5A", "12B")
          bpm_min: Minimum BPM
          bpm_max: Maximum BPM
          rating_min: Minimum rating (0-5)
          limit: Maximum number of results to return
      
      Returns:
          List of matching tracks with metadata
      """
      await ensure_database_connected()
      
      search_options = SearchOptions(
          query=query,
          artist=artist,
          title=title,
          genre=genre,
          key=key,
          bpm_min=bpm_min,
          bpm_max=bpm_max,
          rating_min=rating_min,
          limit=limit
      )
      
      tracks = await db.search_tracks(search_options)
      return [track.model_dump() for track in tracks]

• rekordbox_mcp/models.py:151-185 (schema)

  Pydantic BaseModel defining the input schema for track search parameters, including validation for ranges. Used by both the tool handler and database search method.

  class SearchOptions(BaseModel):
      """
      Search criteria for track queries.
      """
      
      query: str = Field("", description="General search query")
      artist: Optional[str] = Field(None, description="Filter by artist name")
      title: Optional[str] = Field(None, description="Filter by track title")
      album: Optional[str] = Field(None, description="Filter by album name")
      genre: Optional[str] = Field(None, description="Filter by genre")
      key: Optional[str] = Field(None, description="Filter by musical key")
      bpm_min: Optional[float] = Field(None, ge=0, description="Minimum BPM")
      bpm_max: Optional[float] = Field(None, ge=0, description="Maximum BPM")
      rating_min: Optional[int] = Field(None, ge=0, le=5, description="Minimum rating")
      rating_max: Optional[int] = Field(None, ge=0, le=5, description="Maximum rating")
      play_count_min: Optional[int] = Field(None, ge=0, description="Minimum play count")
      play_count_max: Optional[int] = Field(None, ge=0, description="Maximum play count")
      limit: int = Field(50, ge=1, le=1000, description="Maximum number of results")
      
      @field_validator('bpm_max')
      @classmethod
      def validate_bpm_range(cls, v, info):
          """Ensure bpm_max is greater than bpm_min."""
          if v and info.data.get('bpm_min') and v < info.data['bpm_min']:
              raise ValueError('bpm_max must be greater than bpm_min')
          return v
      
      @field_validator('rating_max')
      @classmethod
      def validate_rating_range(cls, v, info):
          """Ensure rating_max is greater than rating_min."""
          if v and info.data.get('rating_min') and v < info.data['rating_min']:
              raise ValueError('rating_max must be greater than rating_min')
          return v

• rekordbox_mcp/database.py:115-181 (helper)

  Core implementation of track search logic in RekordboxDatabase class. Filters database content using SearchOptions criteria, converts matches to Track models, and applies limits.

  async def search_tracks(self, options: SearchOptions) -> List[Track]:
      """
      Search for tracks based on the provided options.
      
      Args:
          options: Search criteria and filters
          
      Returns:
          List of matching tracks
      """
      if not self.db:
          raise RuntimeError("Database not connected")
      
      # Get all content from database, filtering out soft-deleted tracks
      all_content = list(self.db.get_content())
      active_content = [c for c in all_content if getattr(c, 'rb_local_deleted', 0) == 0]
      
      # Apply filters
      filtered_tracks = []
      
      for content in active_content:
          # Get extracted field values for filtering
          artist_name = getattr(content, 'ArtistName', '') or ""
          genre_name = getattr(content, 'GenreName', '') or ""
          key_name = getattr(content, 'KeyName', '') or ""
          bpm_value = (getattr(content, 'BPM', 0) or 0) / 100.0
          rating_value = getattr(content, 'Rating', 0) or 0
          
          # Apply text-based filters
          if options.query and not any([
              options.query.lower() in str(content.Title or "").lower(),
              options.query.lower() in artist_name.lower(),
              options.query.lower() in genre_name.lower(),
          ]):
              continue
          
          if options.artist and options.artist.lower() not in artist_name.lower():
              continue
          
          if options.title and options.title.lower() not in str(content.Title or "").lower():
              continue
          
          if options.genre and options.genre.lower() not in genre_name.lower():
              continue
          
          if options.key and options.key != key_name:
              continue
          
          # Apply numeric filters
          if options.bpm_min and bpm_value < options.bpm_min:
              continue
          
          if options.bpm_max and bpm_value > options.bpm_max:
              continue
          
          if options.rating_min and rating_value < options.rating_min:
              continue
          
          # Convert to our Track model
          track = self._content_to_track(content)
          filtered_tracks.append(track)
          
          # Apply limit
          if len(filtered_tracks) >= options.limit:
              break
      
      return filtered_tracks