Update app.py

app.py

@@ -21,11 +21,29 @@ SPOTIFY_API_BASE = "https://api.spotify.com/v1"
 SPOTIFY_CLIENT_ID = os.getenv("SPOTIFY_CLIENT_ID", "").strip()
 SPOTIFY_CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET", "").strip()
 SPOTIFY_REDIRECT_URI = os.getenv("SPOTIFY_REDIRECT_URI", "").strip()
-SPOTIFY_SCOPES = os.getenv(
+_SPOTIFY_SCOPES_RAW = os.getenv(
     "SPOTIFY_SCOPES",
     "user-read-private user-read-email playlist-read-private "
-    "playlist-read-collaborative playlist-modify-private playlist-modify-public",
+    "playlist-read-collaborative playlist-modify-private playlist-modify-public "
+    "user-library-read user-read-recently-played",
 ).strip()
+_REQUIRED_SCOPES = [
+    "user-read-private",
+    "user-read-email",
+    "playlist-read-private",
+    "playlist-read-collaborative",
+    "playlist-modify-private",
+    "playlist-modify-public",
+    "user-library-read",
+    "user-library-modify",
+    "user-read-recently-played",
+    "user-top-read",
+]
+_scope_parts = _SPOTIFY_SCOPES_RAW.split()
+for _required_scope in _REQUIRED_SCOPES:
+    if _required_scope not in _scope_parts:
+        _scope_parts.append(_required_scope)
+SPOTIFY_SCOPES = " ".join(_scope_parts)
 SPOTIFY_TOKEN_FILE = Path(os.getenv("SPOTIFY_TOKEN_FILE", "spotify_tokens.json")).resolve()
 MCP_SSE_URL = os.getenv(
     "MCP_PUBLIC_SSE_URL",
@@ -294,10 +312,47 @@ def _short_track(track: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _normalize_track_id(track_ref: str) -> str:
+    value = track_ref.strip()
+    if not value:
+        return ""
+
+    if value.startswith("spotify:track:"):
+        return value.split(":")[-1].strip()
+
+    if "open.spotify.com/track/" in value:
+        return value.split("open.spotify.com/track/")[-1].split("?")[0].strip().strip("/")
+
+    return value
+
+
+def _normalize_track_ids(track_ids: list[str], max_items: int = 500) -> list[str]:
+    normalized: list[str] = []
+    seen: set[str] = set()
+
+    for raw in track_ids:
+        item = _normalize_track_id(raw)
+        if item and item not in seen:
+            normalized.append(item)
+            seen.add(item)
+
+    if not normalized:
+        raise SpotifyAuthError("Provide at least one valid track id or track uri.")
+    if len(normalized) > max_items:
+        raise SpotifyAuthError(f"Too many track ids. Maximum allowed is {max_items}.")
+
+    return normalized
+
+
+def _chunked(items: list[str], size: int) -> list[list[str]]:
+    return [items[i : i + size] for i in range(0, len(items), size)]
+
+
 mcp = FastMCP(
     name="Spotify MCP Server",
     instructions=(
-        "MCP server to query Spotify profile, search tracks, and manage playlists. "
+        "MCP server to query Spotify profile, search tracks, top tracks, library, "
+        "listening history, and manage playlists. "
         "If not authenticated, user must open /auth/login first."
     ),
 )
@@ -361,6 +416,157 @@ def spotify_search_tracks(query: str, limit: int = 5, offset: int = 0) -> dict[s
     }
 
 
+@mcp.tool()
+def spotify_list_saved_tracks(limit: int = 20, offset: int = 0) -> dict[str, Any]:
+    """List user's saved (liked) tracks from library (/me/tracks)."""
+    safe_limit = max(1, min(limit, 50))
+    safe_offset = max(0, offset)
+
+    payload = _spotify_request(
+        "GET",
+        "/me/tracks",
+        params={"limit": safe_limit, "offset": safe_offset},
+    )
+
+    items: list[dict[str, Any]] = []
+    for item in payload.get("items", []):
+        track = _short_track(item.get("track") or {})
+        track["added_at"] = item.get("added_at")
+        items.append(track)
+
+    return {
+        "limit": safe_limit,
+        "offset": safe_offset,
+        "total": payload.get("total", 0),
+        "items": items,
+    }
+
+
+@mcp.tool()
+def spotify_list_recently_played(limit: int = 20) -> dict[str, Any]:
+    """List recently played tracks (/me/player/recently-played)."""
+    safe_limit = max(1, min(limit, 50))
+
+    payload = _spotify_request(
+        "GET",
+        "/me/player/recently-played",
+        params={"limit": safe_limit},
+    )
+
+    items: list[dict[str, Any]] = []
+    for item in payload.get("items", []):
+        track = _short_track(item.get("track") or {})
+        track["played_at"] = item.get("played_at")
+        context = item.get("context") or {}
+        track["context_type"] = context.get("type")
+        track["context_uri"] = context.get("uri")
+        items.append(track)
+
+    return {
+        "limit": safe_limit,
+        "items": items,
+        "cursors": payload.get("cursors"),
+    }
+
+
+@mcp.tool()
+def spotify_get_top_tracks(
+    time_range: str = "medium_term",
+    limit: int = 20,
+    offset: int = 0,
+) -> dict[str, Any]:
+    """Get user top tracks (/me/top/tracks)."""
+    allowed_ranges = {"short_term", "medium_term", "long_term"}
+    safe_range = time_range if time_range in allowed_ranges else "medium_term"
+    safe_limit = max(1, min(limit, 50))
+    safe_offset = max(0, offset)
+
+    payload = _spotify_request(
+        "GET",
+        "/me/top/tracks",
+        params={
+            "time_range": safe_range,
+            "limit": safe_limit,
+            "offset": safe_offset,
+        },
+    )
+
+    return {
+        "time_range": safe_range,
+        "limit": safe_limit,
+        "offset": safe_offset,
+        "total": payload.get("total", 0),
+        "items": [_short_track(t) for t in payload.get("items", [])],
+    }
+
+
+@mcp.tool()
+def spotify_check_saved_tracks(track_ids: list[str]) -> dict[str, Any]:
+    """Check if tracks are saved in user's library (/me/tracks/contains)."""
+    ids = _normalize_track_ids(track_ids)
+    chunks = _chunked(ids, 50)
+    states: list[bool] = []
+
+    for chunk in chunks:
+        payload = _spotify_request(
+            "GET",
+            "/me/tracks/contains",
+            params={"ids": ",".join(chunk)},
+        )
+        if isinstance(payload, list):
+            states.extend(bool(x) for x in payload)
+        else:
+            raise SpotifyAuthError("Unexpected response for /me/tracks/contains.")
+
+    items = [{"id": track_id, "saved": states[idx]} for idx, track_id in enumerate(ids)]
+    return {"total": len(ids), "items": items}
+
+
+@mcp.tool()
+def spotify_save_tracks(track_ids: list[str]) -> dict[str, Any]:
+    """Save tracks to user library (PUT /me/tracks)."""
+    ids = _normalize_track_ids(track_ids)
+    chunks = _chunked(ids, 50)
+
+    for chunk in chunks:
+        _spotify_request(
+            "PUT",
+            "/me/tracks",
+            params={"ids": ",".join(chunk)},
+        )
+
+    return {"saved": len(ids), "track_ids": ids}
+
+
+@mcp.tool()
+def spotify_remove_saved_tracks(track_ids: list[str]) -> dict[str, Any]:
+    """Remove tracks from user library (DELETE /me/tracks)."""
+    ids = _normalize_track_ids(track_ids)
+    chunks = _chunked(ids, 50)
+
+    for chunk in chunks:
+        _spotify_request(
+            "DELETE",
+            "/me/tracks",
+            params={"ids": ",".join(chunk)},
+        )
+
+    return {"removed": len(ids), "track_ids": ids}
+
+
+@mcp.tool()
+def spotify_set_tracks_saved(track_ids: list[str], saved: bool) -> dict[str, Any]:
+    """Set saved state for tracks in user library (alta/baja)."""
+    if saved:
+        result = spotify_save_tracks(track_ids)
+        result["action"] = "saved"
+        return result
+
+    result = spotify_remove_saved_tracks(track_ids)
+    result["action"] = "removed"
+    return result
+
+
 @mcp.tool()
 def spotify_list_my_playlists(limit: int = 10, offset: int = 0) -> dict[str, Any]:
     """List playlists from current user (/me/playlists)."""