Update app.py

app.py

@@ -254,6 +254,21 @@ def _spotify_request(
             payload = response.json()
         except ValueError:
             payload = {"error": response.text}
+        if response.status_code == 403:
+            error_message = ""
+            if isinstance(payload, dict):
+                err = payload.get("error")
+                if isinstance(err, dict):
+                    error_message = str(err.get("message", "")).strip()
+                elif isinstance(err, str):
+                    error_message = err.strip()
+            if "Insufficient client scope" in error_message:
+                raise SpotifyAuthError(
+                    "Spotify token without required scopes. Re-authenticate at "
+                    "/auth/reset then /auth/login (force consent), and confirm "
+                    "SPOTIFY_SCOPES includes playlist-modify-private and "
+                    "playlist-modify-public."
+                )
         raise SpotifyAuthError(f"Spotify API error {response.status_code}: {payload}")
 
     if not response.content:
@@ -265,7 +280,7 @@ def _spotify_request(
         return {"raw": response.text}
 
 
-def _build_auth_url(state: str) -> str:
+def _build_auth_url(state: str, show_dialog: bool = True) -> str:
     _require_spotify_config()
     params = {
         "response_type": "code",
@@ -273,7 +288,7 @@ def _build_auth_url(state: str) -> str:
         "scope": SPOTIFY_SCOPES,
         "redirect_uri": SPOTIFY_REDIRECT_URI,
         "state": state,
-        "show_dialog": "false",
+        "show_dialog": "true" if show_dialog else "false",
     }
     return f"{SPOTIFY_ACCOUNTS_BASE}/authorize?{urlencode(params)}"
 
@@ -289,6 +304,24 @@ def _token_status() -> dict[str, Any]:
     }
 
 
+def _token_scope_set() -> set[str]:
+    tokens = _load_tokens() or {}
+    scope_str = str(tokens.get("scope", "")).strip()
+    return {scope for scope in scope_str.split() if scope}
+
+
+def _require_any_scope(required_scopes: list[str], action: str) -> None:
+    token_scopes = _token_scope_set()
+    if not token_scopes:
+        return
+    if any(scope in token_scopes for scope in required_scopes):
+        return
+    raise SpotifyAuthError(
+        f"Missing scope for {action}. Need one of: {', '.join(required_scopes)}. "
+        "Run /auth/reset and /auth/login to re-authorize with updated scopes."
+    )
+
+
 def _short_playlist(playlist: dict[str, Any]) -> dict[str, Any]:
     return {
         "id": playlist.get("id"),
@@ -344,6 +377,20 @@ def _normalize_track_ids(track_ids: list[str], max_items: int = 500) -> list[str
     return normalized
 
 
+def _normalize_playlist_id(playlist_ref: str) -> str:
+    value = playlist_ref.strip()
+    if not value:
+        return ""
+
+    if value.startswith("spotify:playlist:"):
+        return value.split(":")[-1].strip()
+
+    if "open.spotify.com/playlist/" in value:
+        return value.split("open.spotify.com/playlist/")[-1].split("?")[0].strip().strip("/")
+
+    return value
+
+
 def _chunked(items: list[str], size: int) -> list[list[str]]:
     return [items[i : i + size] for i in range(0, len(items), size)]
 
@@ -596,6 +643,8 @@ def spotify_create_playlist(
     """Create a playlist in the current user's account."""
     if not name.strip():
         raise SpotifyAuthError("Playlist name cannot be empty.")
+    required_scope = "playlist-modify-public" if public else "playlist-modify-private"
+    _require_any_scope([required_scope], "create playlist")
 
     me = _spotify_request("GET", "/me")
     user_id = str(me.get("id", "")).strip()
@@ -624,6 +673,10 @@ def spotify_create_playlist(
 @mcp.tool()
 def spotify_add_items_to_playlist(playlist_id: str, uris: list[str]) -> dict[str, Any]:
     """Add Spotify item URIs to a playlist (/playlists/{id}/items)."""
+    _require_any_scope(
+        ["playlist-modify-private", "playlist-modify-public"],
+        "add items to playlist",
+    )
     clean_uris = [u.strip() for u in uris if u and u.strip()]
 
     if not playlist_id.strip():
@@ -635,16 +688,115 @@ def spotify_add_items_to_playlist(playlist_id: str, uris: list[str]) -> dict[str
 
     payload = _spotify_request(
         "POST",
-        f"/playlists/{playlist_id.strip()}/items",
+        f"/playlists/{_normalize_playlist_id(playlist_id)}/items",
         json_body={"uris": clean_uris},
     )
     return {
-        "playlist_id": playlist_id.strip(),
+        "playlist_id": _normalize_playlist_id(playlist_id),
         "added": len(clean_uris),
         "snapshot_id": payload.get("snapshot_id"),
     }
 
 
+@mcp.tool()
+def spotify_update_playlist_details(
+    playlist_id: str,
+    name: str | None = None,
+    description: str | None = None,
+    public: bool | None = None,
+    collaborative: bool | None = None,
+) -> dict[str, Any]:
+    """Update playlist metadata (PUT /playlists/{id})."""
+    _require_any_scope(
+        ["playlist-modify-private", "playlist-modify-public"],
+        "update playlist details",
+    )
+    pid = _normalize_playlist_id(playlist_id)
+    if not pid:
+        raise SpotifyAuthError("playlist_id is required.")
+
+    body: dict[str, Any] = {}
+    if name is not None:
+        body["name"] = name.strip()
+    if description is not None:
+        body["description"] = description
+    if public is not None:
+        body["public"] = public
+    if collaborative is not None:
+        body["collaborative"] = collaborative
+
+    if not body:
+        raise SpotifyAuthError("Provide at least one field to update.")
+
+    # Spotify collaborative playlists must be private.
+    if body.get("collaborative") is True and body.get("public") is True:
+        raise SpotifyAuthError("Collaborative playlists cannot be public.")
+    if body.get("collaborative") is True and "public" not in body:
+        body["public"] = False
+
+    _spotify_request(
+        "PUT",
+        f"/playlists/{pid}",
+        json_body=body,
+    )
+
+    return {
+        "playlist_id": pid,
+        "updated_fields": list(body.keys()),
+    }
+
+
+@mcp.tool()
+def spotify_replace_playlist_items(playlist_id: str, uris: list[str]) -> dict[str, Any]:
+    """Replace all items in playlist (PUT /playlists/{id}/tracks)."""
+    _require_any_scope(
+        ["playlist-modify-private", "playlist-modify-public"],
+        "replace playlist items",
+    )
+    pid = _normalize_playlist_id(playlist_id)
+    if not pid:
+        raise SpotifyAuthError("playlist_id is required.")
+
+    clean_uris = [u.strip() for u in uris if u and u.strip()]
+    if not clean_uris:
+        raise SpotifyAuthError("Provide at least one URI.")
+    if len(clean_uris) > 100:
+        raise SpotifyAuthError("Spotify allows up to 100 URIs per request.")
+
+    payload = _spotify_request(
+        "PUT",
+        f"/playlists/{pid}/tracks",
+        json_body={"uris": clean_uris},
+    )
+    return {
+        "playlist_id": pid,
+        "replaced_with": len(clean_uris),
+        "snapshot_id": payload.get("snapshot_id"),
+    }
+
+
+@mcp.tool()
+def spotify_delete_playlist(playlist_id: str) -> dict[str, Any]:
+    """Delete playlist from library (DELETE /playlists/{id}/followers)."""
+    _require_any_scope(
+        ["playlist-modify-private", "playlist-modify-public"],
+        "delete playlist",
+    )
+    pid = _normalize_playlist_id(playlist_id)
+    if not pid:
+        raise SpotifyAuthError("playlist_id is required.")
+
+    _spotify_request(
+        "DELETE",
+        f"/playlists/{pid}/followers",
+    )
+    return {
+        "playlist_id": pid,
+        "deleted": True,
+        "note": "Spotify performs unfollow; own playlists are removed from your profile.",
+    }
+
+
 app = FastAPI(title="Spotify MCP Server", version="1.0.0")
 app.mount("/gradio_api/mcp", mcp.sse_app("/gradio_api/mcp"))
 
@@ -654,6 +806,7 @@ def root() -> dict[str, Any]:
     return {
         "service": "spotify-mcp-server",
         "auth_login_url": "/auth/login",
+        "auth_reset_url": "/auth/reset",
         "auth_status_url": "/auth/status",
         "mcp_sse_path": "/gradio_api/mcp/sse",
         "public_mcp_sse_url": MCP_SSE_URL,
@@ -670,11 +823,26 @@ def auth_status() -> dict[str, Any]:
     return _token_status()
 
 
+@app.get("/auth/reset")
+def auth_reset() -> dict[str, Any]:
+    removed = False
+    if SPOTIFY_TOKEN_FILE.exists():
+        try:
+            SPOTIFY_TOKEN_FILE.unlink(missing_ok=True)
+            removed = True
+        except OSError as exc:
+            raise HTTPException(
+                status_code=500,
+                detail=f"Could not remove token file: {exc}",
+            ) from exc
+    return {"reset": True, "removed_token_file": removed}
+
+
 @app.get("/auth/login")
-def auth_login() -> RedirectResponse:
+def auth_login(force: bool = Query(default=True)) -> RedirectResponse:
     try:
         state = _new_oauth_state()
-        url = _build_auth_url(state)
+        url = _build_auth_url(state, show_dialog=force)
     except SpotifyAuthError as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc
     return RedirectResponse(url=url, status_code=307)