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