rekordbox-mcp

Instructions

Implementation Reference

• rekordbox_mcp/server.py:549-595 (handler)

  Primary MCP tool handler and registration for 'add_tracks_to_playlist'. Handles tool invocation, input validation via type hints/Pydantic, database connection check, delegates to RekordboxDatabase implementation, and returns formatted response with results summary.
  @mcp.tool( annotations={ "readOnlyHint": False, "destructiveHint": False, "idempotentHint": True } ) async def add_tracks_to_playlist( playlist_id: str, track_ids: List[str] ) -> Dict[str, Any]: """ Add multiple tracks to an existing playlist in one operation. ⚠️ CAUTION: This modifies your rekordbox database! Args: playlist_id: ID of the playlist to modify track_ids: List of track IDs to add Returns: Detailed results of the batch operation """ await ensure_database_connected() try: results = await db.add_tracks_to_playlist(playlist_id, track_ids) return { "status": "success", "message": f"Batch add completed: {len(results['added'])} added, {len(results['skipped'])} skipped, {len(results['failed'])} failed", "playlist_id": playlist_id, "summary": { "added_count": len(results['added']), "skipped_count": len(results['skipped']), "failed_count": len(results['failed']) }, "details": results } except Exception as e: logger.error(f"Failed to add tracks to playlist: {e}") return { "status": "error", "message": f"Failed to add tracks to playlist: {str(e)}" }
• rekordbox_mcp/database.py:765-816 (handler)

  Core database implementation of add_tracks_to_playlist in RekordboxDatabase class. Creates backup, iterates over track IDs, calls pyrekordbox's add_to_playlist for each, commits transaction, tracks added/failed, handles rollback on error.
  async def add_tracks_to_playlist(self, playlist_id: str, track_ids: List[str]) -> Dict[str, Any]: """ Add multiple tracks to a playlist. Args: playlist_id: ID of the playlist to modify track_ids: List of track IDs to add Returns: Dictionary with success/failure details """ if not self.db: raise RuntimeError("Database not connected") try: # Create backup before mutation await self._create_backup() results = { "added": [], "failed": [], "skipped": [] } playlist_int_id = int(playlist_id) for track_id in track_ids: try: track_int_id = int(track_id) # Use the same method as the working single-track function self.db.add_to_playlist(playlist_int_id, track_int_id) results["added"].append(track_id) logger.info(f"Added track {track_id} to playlist {playlist_id}") except Exception as e: results["failed"].append({"track_id": track_id, "reason": str(e)}) logger.warning(f"Failed to add track {track_id}: {e}") # Commit all changes self.db.commit() logger.info(f"Batch add to playlist {playlist_id}: {len(results['added'])} added, {len(results['failed'])} failed") return results except Exception as e: logger.error(f"Failed to add tracks to playlist {playlist_id}: {e}") # Rollback on error if hasattr(self.db, 'rollback'): self.db.rollback() raise RuntimeError(f"Failed to add tracks to playlist: {str(e)}")
• rekordbox_mcp/database.py:926-963 (helper)

  Supporting helper method _create_backup called before mutations like add_tracks_to_playlist to create timestamped backup of the master.db file for safety.
  async def _create_backup(self) -> None: """ Create a backup of the database before performing mutations. """ if not self.database_path: return try: import shutil timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # Try different database file patterns possible_files = [ self.database_path / "master.db", self.database_path / "rekordbox" / "master.db", *list(self.database_path.glob("**/master.db")), *list(self.database_path.glob("**/*.db")) ] db_file = None for file_path in possible_files: if file_path.exists() and file_path.is_file(): db_file = file_path break if db_file: backup_path = self.database_path / f"master_backup_{timestamp}.db" shutil.copy2(db_file, backup_path) logger.info(f"Database backup created: {backup_path}") else: # List available files for debugging all_files = list(self.database_path.rglob("*")) db_files = [f for f in all_files if f.suffix == '.db'] logger.warning(f"No database file found for backup. Available .db files: {db_files}") except Exception as e: logger.warning(f"Failed to create database backup: {e}")