feat: Last.fm integration for music discovery 🎵

Major rewrite — Last.fm replaces deprecated Spotify endpoints:
- suggest: Last.fm artist.getSimilar seeded from Spotify top artists
- discover_artist: Last.fm similar artists → Spotify playback IDs
- fresh_finds: Last.fm tag.getTopTracks → Spotify search
- taste_profile: unchanged (Spotify user data)

Requires LASTFM_API_KEY env var (free from last.fm/api)
This commit is contained in:
Alex Kazaiev
2026-04-06 20:33:51 -05:00
parent 065dd35ceb
commit 1a405b5334
3 changed files with 204 additions and 252 deletions

View File

@@ -3,11 +3,12 @@
MusicTail — Vixy's Music Discovery MCP 🎵🦊
Part of the Tail Family.
Provides intelligent music discovery and taste analysis
using Spotify's recommendation engine seeded with actual listening history.
Uses Last.fm for music intelligence (similar artists, tracks, tags)
and Spotify for taste analysis and playback integration.
Author: Vivienne Rousseau
Created: Day 156 (April 6, 2026)
Updated: Day 156 — Last.fm integration (goodbye deprecated Spotify endpoints!)
"""
from mcp.server.fastmcp import FastMCP
@@ -16,6 +17,7 @@ from typing import Optional, List
from enum import Enum
import os
import json
import httpx
import spotipy
from spotipy.oauth2 import SpotifyOAuth
@@ -23,8 +25,27 @@ from spotipy.oauth2 import SpotifyOAuth
# Initialize MCP server
mcp = FastMCP("musictail")
# Spotify scopes needed
SCOPES = "user-top-read user-read-recently-played user-library-read"
# ========== API Clients ==========
LASTFM_BASE = "http://ws.audioscrobbler.com/2.0/"
SPOTIFY_SCOPES = "user-top-read user-read-recently-played user-library-read"
async def lastfm_call(method: str, **params) -> dict:
"""Call Last.fm API."""
api_key = os.environ.get("LASTFM_API_KEY")
if not api_key:
raise ValueError("LASTFM_API_KEY not set")
params.update({
"method": method,
"api_key": api_key,
"format": "json",
})
async with httpx.AsyncClient() as client:
r = await client.get(LASTFM_BASE, params=params, timeout=15)
r.raise_for_status()
return r.json()
def get_spotify_client() -> spotipy.Spotify:
"""Create authenticated Spotify client."""
@@ -33,50 +54,24 @@ def get_spotify_client() -> spotipy.Spotify:
client_id=os.environ.get("SPOTIFY_CLIENT_ID"),
client_secret=os.environ.get("SPOTIFY_CLIENT_SECRET"),
redirect_uri="http://127.0.0.1:8888/callback",
scope=SCOPES,
scope=SPOTIFY_SCOPES,
cache_path=cache_path,
open_browser=False
open_browser=False,
)
return spotipy.Spotify(auth_manager=auth_manager)
# ========== Models ==========
class MoodEnum(str, Enum):
"""Mood categories mapped to audio features."""
CHILL = "chill"
ENERGETIC = "energetic"
MELANCHOLY = "melancholy"
FOCUSED = "focused"
DREAMY = "dreamy"
DARK = "dark"
UPLIFTING = "uplifting"
INTENSE = "intense"
# Mood to Spotify audio feature mapping
MOOD_FEATURES = {
"chill": {"max_energy": 0.5, "max_tempo": 110, "min_valence": 0.3},
"energetic": {"min_energy": 0.7, "min_tempo": 120, "min_valence": 0.5},
"melancholy": {"max_energy": 0.4, "max_valence": 0.3, "max_tempo": 100},
"focused": {"min_energy": 0.3, "max_energy": 0.7, "max_speechiness": 0.1},
"dreamy": {"max_energy": 0.5, "max_tempo": 110, "min_instrumentalness": 0.5},
"dark": {"max_valence": 0.3, "min_energy": 0.4},
"uplifting": {"min_valence": 0.6, "min_energy": 0.5},
"intense": {"min_energy": 0.8, "min_tempo": 130},
}
class SuggestInput(BaseModel):
"""Input for music suggestions."""
model_config = ConfigDict(extra='forbid')
mood: Optional[MoodEnum] = Field(
mood: Optional[str] = Field(
default=None,
description="Mood filter: chill, energetic, melancholy, focused, dreamy, dark, uplifting, intense"
)
count: int = Field(
default=10,
description="Number of tracks to suggest (1-50)",
ge=1, le=50
description="Mood/tag to filter by (e.g. 'chill', 'dark', 'ambient', 'synthwave', 'energetic')"
)
count: int = Field(default=10, description="Number of tracks (1-30)", ge=1, le=30)
class DiscoverInput(BaseModel):
"""Input for artist discovery."""
model_config = ConfigDict(extra='forbid')
@@ -92,183 +87,143 @@ class TasteInput(BaseModel):
)
class FreshFindsInput(BaseModel):
"""Input for fresh discoveries based on recent listening."""
"""Input for fresh discoveries."""
model_config = ConfigDict(extra='forbid')
count: int = Field(default=10, description="Number of tracks (1-50)", ge=1, le=50)
adventurous: bool = Field(
default=False,
description="If true, push further from comfort zone"
)
count: int = Field(default=10, description="Number of tracks (1-30)", ge=1, le=30)
adventurous: bool = Field(default=False, description="If true, push further from comfort zone")
# ========== Tools ==========
@mcp.tool(
name="musictail_suggest",
annotations={
"title": "Suggest Music By Mood",
"readOnlyHint": True,
"destructiveHint": False,
}
)
@mcp.tool(name="musictail_suggest")
async def suggest_music(params: SuggestInput) -> str:
"""
Suggest music based on your recent listening history and optional mood filter.
Uses genre-based search seeded from your top artists' genres.
"""
"""Suggest music using Last.fm similar artists seeded from your Spotify top artists."""
sp = get_spotify_client()
# Get top artists to extract genres
top_artists = sp.current_user_top_artists(limit=10, time_range="short_term")
# Get top artists from Spotify
top = sp.current_user_top_artists(limit=5, time_range="short_term")
if not top["items"]:
return "No listening history found. Listen to more music first!"
# Collect genres from top artists
genres = []
for a in top_artists["items"]:
genres.extend(a.get("genres", []))
# For each top artist, get Last.fm similar artists
discovered = []
seen_names = {a["name"].lower() for a in top["items"]}
# Deduplicate and pick top genres
genre_count = {}
for g in genres:
genre_count[g] = genre_count.get(g, 0) + 1
sorted_genres = sorted(genre_count.items(), key=lambda x: -x[1])
top_genres = [g for g, _ in sorted_genres[:3]]
for artist in top["items"][:3]:
try:
data = await lastfm_call("artist.getSimilar",
artist=artist["name"], limit=10)
similar = data.get("similarartists", {}).get("artist", [])
for s in similar:
name = s.get("name", "")
if name.lower() not in seen_names:
seen_names.add(name.lower())
discovered.append(name)
except Exception:
continue
if not top_genres:
return "Could not determine your genres. Listen to more music first!"
if not discovered:
return "Last.fm couldn't find similar artists. Try again later."
# Add mood keywords to search
mood_keywords = {
"chill": "chill ambient",
"energetic": "energetic upbeat",
"melancholy": "melancholy sad",
"focused": "instrumental focus",
"dreamy": "dreamy ethereal",
"dark": "dark atmospheric",
"uplifting": "uplifting bright",
"intense": "intense powerful",
}
# Search Spotify for tracks by these artists
lines = [f"🎵 **MusicTail Suggestions** (via Last.fm)"]
if params.mood:
lines[0] += f" — mood: {params.mood}"
lines.append(f"Based on artists similar to: {', '.join(a['name'] for a in top['items'][:3])}\n")
# Build search query from genres + mood
query_parts = top_genres[:2]
if params.mood and params.mood.value in mood_keywords:
query_parts.append(mood_keywords[params.mood.value])
query = " ".join(query_parts)
found = 0
for artist_name in discovered:
if found >= params.count:
break
query = f"artist:{artist_name}"
if params.mood:
query += f" {params.mood}"
search = sp.search(q=query, type="track", limit=2)
for track in search["tracks"]["items"]:
if found >= params.count:
break
artists = ", ".join(a["name"] for a in track["artists"])
lines.append(f"{found+1}. **{track['name']}** — {artists}")
lines.append(f" Album: {track['album']['name']} | ID: `{track['id']}`")
found += 1
# Search for tracks
results = sp.search(q=query, type="track", limit=params.count)
# Filter out tracks by artists already in top list
top_artist_ids = {a["id"] for a in top_artists["items"]}
tracks = [t for t in results["tracks"]["items"]
if not any(a["id"] in top_artist_ids for a in t["artists"])]
# If too few new tracks, include some known artists
if len(tracks) < params.count // 2:
tracks = results["tracks"]["items"][:params.count]
lines = [f"🎵 **MusicTail Suggestions**" + (f" — mood: {params.mood.value}" if params.mood else "")]
lines.append(f"Based on your genres: {', '.join(top_genres)}\n")
for i, track in enumerate(tracks[:params.count], 1):
artists = ", ".join(a["name"] for a in track["artists"])
album = track["album"]["name"]
tid = track["id"]
lines.append(f"{i}. **{track['name']}** — {artists}")
lines.append(f" Album: {album} | ID: `{tid}`")
if found == 0:
lines.append("No matching tracks found on Spotify.")
return "\n".join(lines)
@mcp.tool(
name="musictail_discover_artist",
annotations={
"title": "Discover Similar Artists",
"readOnlyHint": True,
"destructiveHint": False,
}
)
@mcp.tool(name="musictail_discover_artist")
async def discover_artist(params: DiscoverInput) -> str:
"""
Find artists similar to one you like, with their top tracks.
Uses genre-based search to find artists in the same musical space.
"""
"""Find similar artists via Last.fm, with Spotify playback links."""
# Get similar artists from Last.fm
try:
data = await lastfm_call("artist.getSimilar",
artist=params.artist_name, limit=params.count)
except Exception as e:
return f"Last.fm error: {e}"
similar = data.get("similarartists", {}).get("artist", [])
if not similar:
return f"No similar artists found for '{params.artist_name}' on Last.fm."
# Also get tags for the source artist
try:
tag_data = await lastfm_call("artist.getTopTags", artist=params.artist_name)
tags = [t["name"] for t in tag_data.get("toptags", {}).get("tag", [])[:5]]
except Exception:
tags = []
tag_str = ", ".join(tags) if tags else "unknown"
lines = [f"🔍 **Artists similar to {params.artist_name}** (via Last.fm)"]
lines.append(f"Tags: {tag_str}\n")
sp = get_spotify_client()
# Search for the artist
search = sp.search(q=params.artist_name, type="artist", limit=1)
if not search["artists"]["items"]:
return f"Could not find artist: {params.artist_name}"
artist = search["artists"]["items"][0]
artist_genres = artist.get("genres", [])
genres_str = ", ".join(artist_genres[:5]) or "no genres listed"
if not artist_genres:
return f"**{artist['name']}** has no genres listed on Spotify — can't find similar artists without genre data."
# Search for artists in the same genres
seen_ids = {artist["id"]}
similar = []
for genre in artist_genres[:3]:
if len(similar) >= params.count:
break
genre_search = sp.search(q=f"genre:\"{genre}\"", type="artist", limit=20)
for a in genre_search["artists"]["items"]:
if a["id"] not in seen_ids and len(similar) < params.count:
seen_ids.add(a["id"])
similar.append(a)
lines = [f"🔍 **Artists similar to {artist['name']}**"]
lines.append(f"Genres: {genres_str}\n")
for i, sim in enumerate(similar, 1):
sim_genres = ", ".join(sim.get("genres", [])[:3]) or ""
popularity = sim.get("popularity", 0)
for i, sim in enumerate(similar[:params.count], 1):
name = sim.get("name", "Unknown")
match_score = sim.get("match", "?")
# Get top tracks for this artist
top = sp.artist_top_tracks(sim["id"])
top_tracks = top["tracks"][:3]
track_list = "; ".join(t["name"] for t in top_tracks)
# Search Spotify for this artist
search = sp.search(q=f"artist:{name}", type="artist", limit=1)
spotify_artists = search["artists"]["items"]
lines.append(f"{i}. **{sim['name']}** (popularity: {popularity})")
lines.append(f" Genres: {sim_genres}")
lines.append(f" Top tracks: {track_list}")
if top_tracks:
lines.append(f" Play ID: `{top_tracks[0]['id']}`")
if spotify_artists:
sa = spotify_artists[0]
genres = ", ".join(sa.get("genres", [])[:3]) or ""
# Get top tracks
top = sp.artist_top_tracks(sa["id"])
top_tracks = top["tracks"][:3]
track_list = "; ".join(t["name"] for t in top_tracks)
play_id = top_tracks[0]["id"] if top_tracks else None
lines.append(f"{i}. **{name}** (match: {float(match_score):.0%})")
lines.append(f" Genres: {genres}")
lines.append(f" Top tracks: {track_list}")
if play_id:
lines.append(f" Play ID: `{play_id}`")
else:
lines.append(f"{i}. **{name}** (match: {float(match_score):.0%})")
lines.append(f" Not found on Spotify")
lines.append("")
if not similar:
lines.append("No similar artists found in these genres.")
return "\n".join(lines)
@mcp.tool(
name="musictail_taste_profile",
annotations={
"title": "Analyze Taste Profile",
"readOnlyHint": True,
"destructiveHint": False,
}
)
@mcp.tool(name="musictail_taste_profile")
async def taste_profile(params: TasteInput) -> str:
"""
Analyze your listening patterns across time. Shows top artists, top genres, and musical characteristics for the selected time range.
"""
"""Analyze your Spotify listening patterns — top artists, genres, tracks."""
sp = get_spotify_client()
range_label = {
"short_term": "Last ~4 weeks",
"medium_term": "Last ~6 months",
"medium_term": "Last ~6 months",
"long_term": "All time"
}.get(params.time_range, params.time_range)
# Top artists
top_artists = sp.current_user_top_artists(limit=15, time_range=params.time_range)
# Top tracks
top_tracks = sp.current_user_top_tracks(limit=20, time_range=params.time_range)
# Collect genres
genre_count = {}
for a in top_artists["items"]:
for g in a.get("genres", []):
@@ -276,7 +231,6 @@ async def taste_profile(params: TasteInput) -> str:
top_genres = sorted(genre_count.items(), key=lambda x: -x[1])[:10]
lines = [f"📊 **MusicTail Taste Profile** — {range_label}\n"]
lines.append("**Top Artists:**")
for i, a in enumerate(top_artists["items"][:10], 1):
lines.append(f" {i}. {a['name']}")
@@ -292,69 +246,70 @@ async def taste_profile(params: TasteInput) -> str:
return "\n".join(lines)
@mcp.tool(
name="musictail_fresh_finds",
annotations={
"title": "Fresh Finds",
"readOnlyHint": True,
"destructiveHint": False,
}
)
@mcp.tool(name="musictail_fresh_finds")
async def fresh_finds(params: FreshFindsInput) -> str:
"""
Discover new tracks based on your listening, with an adventure dial.
Normal mode stays close to your taste. Adventurous mode pushes boundaries.
"""
"""Discover new music via Last.fm tag exploration, playable on Spotify."""
sp = get_spotify_client()
# Get genres from top artists
# Get user's genres from Spotify top artists
time_range = "long_term" if params.adventurous else "short_term"
top_artists = sp.current_user_top_artists(limit=15, time_range=time_range)
# Collect and rank genres
genre_count = {}
for a in top_artists["items"]:
for g in a.get("genres", []):
genre_count[g] = genre_count.get(g, 0) + 1
sorted_genres = sorted(genre_count.items(), key=lambda x: -x[1])
# Adventurous = less common genres; Normal = top genres
# Adventurous = less dominant genres; Normal = top genres
if params.adventurous and len(sorted_genres) > 3:
pick_genres = [g for g, _ in sorted_genres[3:6]] # Less dominant genres
query_extra = "new"
pick_tags = [g for g, _ in sorted_genres[3:7]]
else:
pick_genres = [g for g, _ in sorted_genres[:3]]
query_extra = ""
pick_tags = [g for g, _ in sorted_genres[:3]]
if not pick_genres:
return "Could not determine your genres. Listen to more music first!"
if not pick_tags:
return "Could not determine your genres."
# Search for tracks in those genres
query = " ".join(pick_genres[:2])
if query_extra:
query += f" {query_extra}"
# Use Last.fm tag.getTopTracks for each genre
known_artists = {a["name"].lower() for a in top_artists["items"]}
discovered = []
results = sp.search(q=query, type="track", limit=min(params.count * 2, 50))
# Filter out tracks by known top artists for freshness
top_artist_ids = {a["id"] for a in top_artists["items"]}
fresh = [t for t in results["tracks"]["items"]
if not any(a["id"] in top_artist_ids for a in t["artists"])]
# Fall back to all results if too few fresh ones
tracks = fresh[:params.count] if len(fresh) >= params.count // 2 else results["tracks"]["items"][:params.count]
for tag in pick_tags:
if len(discovered) >= params.count:
break
try:
data = await lastfm_call("tag.getTopTracks", tag=tag, limit=20)
tracks = data.get("tracks", {}).get("track", [])
for t in tracks:
artist_name = t.get("artist", {}).get("name", "")
track_name = t.get("name", "")
if artist_name.lower() not in known_artists:
discovered.append((track_name, artist_name))
if len(discovered) >= params.count:
break
except Exception:
continue
mode = "🗺️ Adventurous" if params.adventurous else "🏠 Comfort Zone"
lines = [f"🎵 **MusicTail Fresh Finds** — {mode}"]
lines.append(f"Searching: {', '.join(pick_genres)}\n")
lines = [f"🎵 **MusicTail Fresh Finds** — {mode} (via Last.fm)"]
lines.append(f"Exploring tags: {', '.join(pick_tags)}\n")
for i, track in enumerate(tracks[:params.count], 1):
artists = ", ".join(a["name"] for a in track["artists"])
album = track["album"]["name"]
pop = track.get("popularity", 0)
tid = track["id"]
lines.append(f"{i}. **{track['name']}** — {artists}")
lines.append(f" Album: {album} | Popularity: {pop} | ID: `{tid}`")
# Find on Spotify
found = 0
for track_name, artist_name in discovered:
if found >= params.count:
break
search = sp.search(q=f"track:{track_name} artist:{artist_name}", type="track", limit=1)
items = search["tracks"]["items"]
if items:
t = items[0]
artists = ", ".join(a["name"] for a in t["artists"])
lines.append(f"{found+1}. **{t['name']}** — {artists}")
lines.append(f" Album: {t['album']['name']} | ID: `{t['id']}`")
found += 1
if found == 0:
lines.append("No matching tracks found on Spotify for these tags.")
return "\n".join(lines)
@@ -362,36 +317,32 @@ async def fresh_finds(params: FreshFindsInput) -> str:
# ========== Entry Point ==========
if __name__ == "__main__":
# When run directly, do initial OAuth dance
import sys
if "--auth" in sys.argv:
print("🎵 MusicTail — First-time authorization")
print("🎵 MusicTail — First-time Spotify authorization")
print()
cache_path = os.path.expanduser("~/.musictail_cache")
auth_manager = SpotifyOAuth(
client_id=os.environ.get("SPOTIFY_CLIENT_ID"),
client_secret=os.environ.get("SPOTIFY_CLIENT_SECRET"),
redirect_uri="http://127.0.0.1:8888/callback",
scope=SCOPES,
scope=SPOTIFY_SCOPES,
cache_path=cache_path,
open_browser=False
open_browser=False,
)
# Get the auth URL
auth_url = auth_manager.get_authorize_url()
print(f"Open this URL in your browser:\n{auth_url}\n")
print("After authorizing, you'll be redirected to a URL.")
print("Paste the FULL redirect URL here (it will start with http://127.0.0.1:8888/callback?code=...):")
print("Paste the FULL redirect URL here:")
response_url = input("> ").strip()
code = auth_manager.parse_response_code(response_url)
token_info = auth_manager.get_access_token(code)
if token_info:
sp = spotipy.Spotify(auth_manager=auth_manager)
user = sp.current_user()
print(f"\n✅ Authorized as: {user['display_name']} ({user['id']})")
print(f"\n✅ Authorized as: {user['display_name']}")
print(f"Token cached at: {cache_path}")
print("MusicTail is ready! 🦊")
else:
print("❌ Authorization failed. Check your credentials.")
print("❌ Authorization failed.")
else:
mcp.run()
mcp.run()