Update app.py

app.py

@@ -3,377 +3,511 @@ import json
 import os
 import secrets
 import time
-import urllib.parse
-from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlencode
 
-import gradio as gr
-import requests
+import httpx
 from dotenv import load_dotenv
-
+from fastapi import FastAPI, HTTPException, Query
+from fastapi.responses import HTMLResponse, RedirectResponse
+from mcp.server.fastmcp import FastMCP
 
 load_dotenv()
 
 SPOTIFY_ACCOUNTS_BASE = "https://accounts.spotify.com"
 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",
+    "user-read-private user-read-email playlist-read-private "
+    "playlist-read-collaborative playlist-modify-private playlist-modify-public",
+).strip()
+SPOTIFY_TOKEN_FILE = Path(os.getenv("SPOTIFY_TOKEN_FILE", "spotify_tokens.json")).resolve()
+MCP_SSE_URL = os.getenv(
+    "MCP_PUBLIC_SSE_URL",
+    "https://pharmaia-demo-mcp-server-spotify.hf.space/gradio_api/mcp/sse",
+).strip()
+
+# Keep OAuth states in memory for CSRF protection.
+OAUTH_STATE_TTL_SECONDS = 600
+_oauth_states: dict[str, int] = {}
+
+
+class SpotifyAuthError(RuntimeError):
+    pass
+
+
+def _require_spotify_config() -> None:
+    missing = []
+    if not SPOTIFY_CLIENT_ID:
+        missing.append("SPOTIFY_CLIENT_ID")
+    if not SPOTIFY_CLIENT_SECRET:
+        missing.append("SPOTIFY_CLIENT_SECRET")
+    if not SPOTIFY_REDIRECT_URI:
+        missing.append("SPOTIFY_REDIRECT_URI")
+
+    if missing:
+        raise SpotifyAuthError(
+            "Missing required environment variables: " + ", ".join(missing)
+        )
 
-@dataclass
-class AuthConfig:
-    client_id: str
-    client_secret: str
-    redirect_uri: str
-    scopes: str
-    token_file: Path
-    state_file: Path
-    env_refresh_token: str | None
 
+def _cleanup_expired_states() -> None:
+    now = int(time.time())
+    expired = [k for k, ts in _oauth_states.items() if (now - ts) > OAUTH_STATE_TTL_SECONDS]
+    for key in expired:
+        _oauth_states.pop(key, None)
 
-class SpotifyClient:
-    def __init__(self, config: AuthConfig) -> None:
-        self.config = config
 
-    def _read_json(self, path: Path) -> dict[str, Any] | None:
-        if not path.exists():
-            return None
-        try:
-            return json.loads(path.read_text(encoding="utf-8"))
-        except (json.JSONDecodeError, OSError):
-            return None
+def _new_oauth_state() -> str:
+    _cleanup_expired_states()
+    state = secrets.token_urlsafe(24)
+    _oauth_states[state] = int(time.time())
+    return state
+
+
+def _consume_oauth_state(state: str) -> bool:
+    _cleanup_expired_states()
+    ts = _oauth_states.pop(state, None)
+    if ts is None:
+        return False
+    return (int(time.time()) - ts) <= OAUTH_STATE_TTL_SECONDS
+
+
+def _spotify_basic_auth_header() -> str:
+    token = f"{SPOTIFY_CLIENT_ID}:{SPOTIFY_CLIENT_SECRET}".encode("utf-8")
+    return base64.b64encode(token).decode("utf-8")
+
+
+def _load_tokens() -> dict[str, Any] | None:
+    if not SPOTIFY_TOKEN_FILE.exists():
+        return None
+    try:
+        return json.loads(SPOTIFY_TOKEN_FILE.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError):
+        return None
+
+
+def _save_tokens(token_payload: dict[str, Any], previous: dict[str, Any] | None = None) -> dict[str, Any]:
+    now = int(time.time())
+    expires_in = int(token_payload.get("expires_in", 3600))
+
+    merged: dict[str, Any] = {}
+    if previous:
+        merged.update(previous)
+    merged.update(token_payload)
+
+    if not merged.get("refresh_token") and previous:
+        merged["refresh_token"] = previous.get("refresh_token")
+
+    merged["saved_at"] = now
+    merged["expires_at"] = now + expires_in
+
+    SPOTIFY_TOKEN_FILE.write_text(
+        json.dumps(merged, indent=2, ensure_ascii=True),
+        encoding="utf-8",
+    )
+    return merged
 
-    def _write_json(self, path: Path, payload: dict[str, Any]) -> None:
-        path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
 
-    def _read_token_data(self) -> dict[str, Any] | None:
-        return self._read_json(self.config.token_file)
+def _is_expired(tokens: dict[str, Any]) -> bool:
+    expires_at = int(tokens.get("expires_at", 0))
+    return int(time.time()) >= (expires_at - 30)
 
-    def _write_token_data(self, token_data: dict[str, Any]) -> None:
-        token_data["saved_at"] = int(time.time())
-        self._write_json(self.config.token_file, token_data)
 
-    def _basic_auth_header(self) -> str:
-        raw = f"{self.config.client_id}:{self.config.client_secret}".encode("utf-8")
-        return base64.b64encode(raw).decode("ascii")
+def _exchange_code_for_tokens(code: str) -> dict[str, Any]:
+    _require_spotify_config()
 
-    def _token_is_expired(self, token_data: dict[str, Any]) -> bool:
-        expires_in = int(token_data.get("expires_in", 0))
-        saved_at = int(token_data.get("saved_at", 0))
-        return time.time() >= saved_at + expires_in - 60
+    data = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": SPOTIFY_REDIRECT_URI,
+    }
+    headers = {
+        "Authorization": f"Basic {_spotify_basic_auth_header()}",
+        "Content-Type": "application/x-www-form-urlencoded",
+    }
 
-    def _request_token(self, data: dict[str, Any]) -> dict[str, Any]:
-        response = requests.post(
+    with httpx.Client(timeout=20.0) as client:
+        response = client.post(
             f"{SPOTIFY_ACCOUNTS_BASE}/api/token",
-            headers={
-                "Authorization": f"Basic {self._basic_auth_header()}",
-                "Content-Type": "application/x-www-form-urlencoded",
-            },
             data=data,
-            timeout=30,
+            headers=headers,
         )
-        response.raise_for_status()
-        return response.json()
 
-    def _refresh_token(self, refresh_token: str) -> dict[str, Any]:
-        new_token = self._request_token(
-            {
-                "grant_type": "refresh_token",
-                "refresh_token": refresh_token,
-            }
-        )
-        new_token["refresh_token"] = new_token.get("refresh_token", refresh_token)
-        self._write_token_data(new_token)
-        return new_token
-
-    def _get_access_token(self) -> str:
-        token_data = self._read_token_data()
-        if token_data:
-            if self._token_is_expired(token_data):
-                refresh = token_data.get("refresh_token") or self.config.env_refresh_token
-                if not refresh:
-                    raise RuntimeError("Token expired and no refresh token is available.")
-                token_data = self._refresh_token(refresh)
-            token = token_data.get("access_token")
-            if token:
-                return token
-
-        if self.config.env_refresh_token:
-            token_data = self._refresh_token(self.config.env_refresh_token)
-            token = token_data.get("access_token")
-            if token:
-                return token
-
-        raise RuntimeError(
-            "No auth session. Set SPOTIFY_REFRESH_TOKEN in env, or run spotify_auth_url and spotify_exchange_code."
-        )
+    if response.status_code != 200:
+        try:
+            payload = response.json()
+        except ValueError:
+            payload = {"error": response.text}
+        raise SpotifyAuthError(f"Failed token exchange: {payload}")
 
-    def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
-        headers = kwargs.pop("headers", {})
-        headers["Authorization"] = f"Bearer {self._get_access_token()}"
-        headers.setdefault("Content-Type", "application/json")
+    return _save_tokens(response.json())
 
-        response = requests.request(
-            method,
-            f"{SPOTIFY_API_BASE}{path}",
-            headers=headers,
-            timeout=30,
-            **kwargs,
-        )
-        if response.status_code >= 400:
-            raise RuntimeError(f"Spotify API error {response.status_code}: {response.text}")
-        if response.status_code == 204:
-            return {"ok": True}
-        return response.json()
 
-    def auth_url(self, state: str | None = None) -> dict[str, Any]:
-        state_value = state or secrets.token_urlsafe(24)
-        self._write_json(self.config.state_file, {"state": state_value, "saved_at": int(time.time())})
-        url = (
-            f"{SPOTIFY_ACCOUNTS_BASE}/authorize?"
-            + urllib.parse.urlencode(
-                {
-                    "client_id": self.config.client_id,
-                    "response_type": "code",
-                    "redirect_uri": self.config.redirect_uri,
-                    "scope": self.config.scopes,
-                    "state": state_value,
-                    "show_dialog": "true",
-                }
-            )
+def _refresh_access_token(tokens: dict[str, Any]) -> dict[str, Any]:
+    refresh_token = tokens.get("refresh_token")
+    if not refresh_token:
+        raise SpotifyAuthError(
+            "No refresh token found. Re-authenticate at /auth/login."
         )
-        return {
-            "auth_url": url,
-            "state": state_value,
-            "redirect_uri": self.config.redirect_uri,
-            "next_step": "Open auth_url, approve app, then call spotify_exchange_code with returned code and state.",
-        }
-
-    def exchange_code(self, code: str, state: str | None = None) -> dict[str, Any]:
-        state_data = self._read_json(self.config.state_file) or {}
-        expected = state_data.get("state")
-        if expected and state and state != expected:
-            raise RuntimeError("OAuth state mismatch.")
-
-        token_data = self._request_token(
-            {
-                "grant_type": "authorization_code",
-                "code": code,
-                "redirect_uri": self.config.redirect_uri,
-            }
+
+    data = {
+        "grant_type": "refresh_token",
+        "refresh_token": refresh_token,
+    }
+    headers = {
+        "Authorization": f"Basic {_spotify_basic_auth_header()}",
+        "Content-Type": "application/x-www-form-urlencoded",
+    }
+
+    with httpx.Client(timeout=20.0) as client:
+        response = client.post(
+            f"{SPOTIFY_ACCOUNTS_BASE}/api/token",
+            data=data,
+            headers=headers,
         )
-        self._write_token_data(token_data)
-        me = self.me()
-        return {
-            "status": "ok",
-            "user_id": me.get("id"),
-            "display_name": me.get("display_name"),
-            "has_refresh_token": bool(token_data.get("refresh_token")),
-        }
-
-    def me(self) -> dict[str, Any]:
-        me = self._request("GET", "/me")
-        return {
-            "id": me.get("id"),
-            "display_name": me.get("display_name"),
-            "email": me.get("email"),
-            "product": me.get("product"),
-        }
-
-    def search_tracks(self, query: str, limit: int = 5) -> dict[str, Any]:
-        payload = self._request(
-            "GET",
-            "/search",
-            params={"q": query, "type": "track", "limit": max(1, min(limit, 50))},
+
+    if response.status_code != 200:
+        try:
+            payload = response.json()
+        except ValueError:
+            payload = {"error": response.text}
+        raise SpotifyAuthError(f"Failed token refresh: {payload}")
+
+    return _save_tokens(response.json(), previous=tokens)
+
+
+def _get_valid_access_token() -> str:
+    _require_spotify_config()
+
+    tokens = _load_tokens()
+    if not tokens:
+        raise SpotifyAuthError(
+            "Spotify is not authenticated yet. Open /auth/login first."
         )
-        items = payload.get("tracks", {}).get("items", [])
-        return {
-            "results": [
-                {
-                    "id": t["id"],
-                    "name": t["name"],
-                    "artists": ", ".join(a["name"] for a in t.get("artists", [])),
-                    "uri": t["uri"],
-                }
-                for t in items
-            ]
-        }
-
-    def create_playlist(self, name: str, description: str = "", public: bool = False) -> dict[str, Any]:
-        playlist = self._request(
-            "POST",
-            "/me/playlists",
-            json={"name": name, "description": description, "public": public},
+
+    if _is_expired(tokens):
+        tokens = _refresh_access_token(tokens)
+
+    access_token = tokens.get("access_token")
+    if not access_token:
+        raise SpotifyAuthError(
+            "Access token missing. Re-authenticate at /auth/login."
         )
-        return {
-            "id": playlist.get("id"),
-            "name": playlist.get("name"),
-            "url": playlist.get("external_urls", {}).get("spotify"),
-        }
-
-    def add_tracks(self, playlist_id: str, track_ids: list[str]) -> dict[str, Any]:
-        uris = [tid if tid.startswith("spotify:track:") else f"spotify:track:{tid}" for tid in track_ids]
-        payload = self._request("POST", f"/playlists/{playlist_id}/items", json={"uris": uris})
-        return {"snapshot_id": payload.get("snapshot_id"), "added": len(uris)}
-
-
-def load_config() -> AuthConfig:
-    client_id = os.getenv("SPOTIFY_CLIENT_ID", "").strip()
-    client_secret = os.getenv("SPOTIFY_CLIENT_SECRET", "").strip()
-    redirect_uri = os.getenv("SPOTIFY_REDIRECT_URI", "").strip()
-    scopes = os.getenv(
-        "SPOTIFY_SCOPES",
-        "playlist-modify-public playlist-modify-private user-read-private user-read-email",
-    ).strip()
-    token_file = Path(os.getenv("SPOTIFY_TOKEN_FILE", "spotify_tokens.json"))
-    state_file = Path(os.getenv("SPOTIFY_STATE_FILE", "spotify_oauth_state.json"))
-    env_refresh_token = os.getenv("SPOTIFY_REFRESH_TOKEN", "").strip() or None
-
-    if not client_id or not client_secret or not redirect_uri:
-        raise RuntimeError("SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET and SPOTIFY_REDIRECT_URI are required.")
-
-    return AuthConfig(
-        client_id=client_id,
-        client_secret=client_secret,
-        redirect_uri=redirect_uri,
-        scopes=scopes,
-        token_file=token_file,
-        state_file=state_file,
-        env_refresh_token=env_refresh_token,
-    )
+    return str(access_token)
 
 
-_spotify: SpotifyClient | None = None
-_config_error: str | None = None
+def _spotify_request(
+    method: str,
+    path: str,
+    params: dict[str, Any] | None = None,
+    json_body: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    token = _get_valid_access_token()
+    headers = {"Authorization": f"Bearer {token}"}
 
+    with httpx.Client(timeout=20.0) as client:
+        response = client.request(
+            method,
+            f"{SPOTIFY_API_BASE}{path}",
+            params=params,
+            json=json_body,
+            headers=headers,
+        )
 
-def get_spotify() -> SpotifyClient:
-    global _spotify, _config_error
-    if _spotify is not None:
-        return _spotify
-    try:
-        _spotify = SpotifyClient(load_config())
-        _config_error = None
-        return _spotify
-    except Exception as exc:
-        _config_error = str(exc)
-        raise RuntimeError(
-            "Spotify config missing/invalid. Set HF Space Secrets: "
-            "SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET, SPOTIFY_REDIRECT_URI. "
-            f"Details: {_config_error}"
-        ) from exc
+        if response.status_code == 401:
+            refreshed = _refresh_access_token(_load_tokens() or {})
+            headers["Authorization"] = f"Bearer {refreshed.get('access_token', '')}"
+            response = client.request(
+                method,
+                f"{SPOTIFY_API_BASE}{path}",
+                params=params,
+                json=json_body,
+                headers=headers,
+            )
 
+    if response.status_code >= 400:
+        try:
+            payload = response.json()
+        except ValueError:
+            payload = {"error": response.text}
+        raise SpotifyAuthError(f"Spotify API error {response.status_code}: {payload}")
 
-def spotify_auth_url(state: str = "") -> dict[str, Any]:
-    return get_spotify().auth_url(state=state or None)
+    if not response.content:
+        return {}
 
+    try:
+        return response.json()
+    except ValueError:
+        return {"raw": response.text}
+
+
+def _build_auth_url(state: str) -> str:
+    _require_spotify_config()
+    params = {
+        "response_type": "code",
+        "client_id": SPOTIFY_CLIENT_ID,
+        "scope": SPOTIFY_SCOPES,
+        "redirect_uri": SPOTIFY_REDIRECT_URI,
+        "state": state,
+        "show_dialog": "false",
+    }
+    return f"{SPOTIFY_ACCOUNTS_BASE}/authorize?{urlencode(params)}"
+
+
+def _token_status() -> dict[str, Any]:
+    tokens = _load_tokens() or {}
+    authenticated = bool(tokens.get("access_token"))
+    return {
+        "authenticated": authenticated,
+        "token_file": str(SPOTIFY_TOKEN_FILE),
+        "expires_at": tokens.get("expires_at"),
+        "scopes": tokens.get("scope"),
+    }
+
+
+def _short_playlist(playlist: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "id": playlist.get("id"),
+        "name": playlist.get("name"),
+        "uri": playlist.get("uri"),
+        "public": playlist.get("public"),
+        "owner": (playlist.get("owner") or {}).get("id"),
+        "total_items": (playlist.get("tracks") or {}).get("total"),
+        "external_url": (playlist.get("external_urls") or {}).get("spotify"),
+    }
+
+
+def _short_track(track: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "id": track.get("id"),
+        "name": track.get("name"),
+        "uri": track.get("uri"),
+        "artists": [a.get("name") for a in track.get("artists", [])],
+        "album": (track.get("album") or {}).get("name"),
+        "external_url": (track.get("external_urls") or {}).get("spotify"),
+    }
+
+
+mcp = FastMCP(
+    name="Spotify MCP Server",
+    instructions=(
+        "MCP server to query Spotify profile, search tracks, and manage playlists. "
+        "If not authenticated, user must open /auth/login first."
+    ),
+)
 
-def spotify_exchange_code(code: str, state: str = "") -> dict[str, Any]:
-    return get_spotify().exchange_code(code=code, state=state or None)
 
+@mcp.tool()
+def spotify_auth_status() -> dict[str, Any]:
+    """Return whether the server already has a valid Spotify token."""
+    return _token_status()
+
+
+@mcp.tool()
+def spotify_get_auth_url() -> dict[str, str]:
+    """Generate the Spotify OAuth URL. Open it in a browser to connect the account."""
+    state = _new_oauth_state()
+    return {
+        "auth_url": _build_auth_url(state),
+        "callback": SPOTIFY_REDIRECT_URI,
+        "note": "Open auth_url in browser and finish login/consent.",
+    }
+
+
+@mcp.tool()
+def spotify_get_my_profile() -> dict[str, Any]:
+    """Get current Spotify user profile (/me)."""
+    me = _spotify_request("GET", "/me")
+    return {
+        "id": me.get("id"),
+        "display_name": me.get("display_name"),
+        "uri": me.get("uri"),
+        "country": me.get("country"),
+        "product": me.get("product"),
+    }
+
+
+@mcp.tool()
+def spotify_search_tracks(query: str, limit: int = 5, offset: int = 0) -> dict[str, Any]:
+    """Search tracks using Spotify /search endpoint."""
+    safe_limit = max(1, min(limit, 10))
+    safe_offset = max(0, offset)
+
+    payload = _spotify_request(
+        "GET",
+        "/search",
+        params={
+            "q": query,
+            "type": "track",
+            "limit": safe_limit,
+            "offset": safe_offset,
+        },
+    )
+    tracks = payload.get("tracks") or {}
+    items = [_short_track(t) for t in tracks.get("items", [])]
+
+    return {
+        "query": query,
+        "limit": safe_limit,
+        "offset": safe_offset,
+        "total": tracks.get("total", 0),
+        "items": items,
+    }
+
+
+@mcp.tool()
+def spotify_list_my_playlists(limit: int = 10, offset: int = 0) -> dict[str, Any]:
+    """List playlists from current user (/me/playlists)."""
+    safe_limit = max(1, min(limit, 50))
+    safe_offset = max(0, offset)
+
+    payload = _spotify_request(
+        "GET",
+        "/me/playlists",
+        params={"limit": safe_limit, "offset": safe_offset},
+    )
 
-def spotify_me() -> dict[str, Any]:
-    return get_spotify().me()
+    return {
+        "limit": safe_limit,
+        "offset": safe_offset,
+        "total": payload.get("total", 0),
+        "items": [_short_playlist(p) for p in payload.get("items", [])],
+    }
+
+
+@mcp.tool()
+def spotify_create_playlist(
+    name: str,
+    description: str = "",
+    public: bool = False,
+) -> dict[str, Any]:
+    """Create a playlist in the current user's account."""
+    if not name.strip():
+        raise SpotifyAuthError("Playlist name cannot be empty.")
+
+    me = _spotify_request("GET", "/me")
+    user_id = str(me.get("id", "")).strip()
+    if not user_id:
+        raise SpotifyAuthError("Could not resolve current Spotify user id.")
+
+    payload = _spotify_request(
+        "POST",
+        f"/users/{user_id}/playlists",
+        json_body={
+            "name": name.strip(),
+            "description": description,
+            "public": public,
+        },
+    )
 
+    return {
+        "id": payload.get("id"),
+        "name": payload.get("name"),
+        "uri": payload.get("uri"),
+        "public": payload.get("public"),
+        "external_url": (payload.get("external_urls") or {}).get("spotify"),
+    }
+
+
+@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)."""
+    clean_uris = [u.strip() for u in uris if u and u.strip()]
+
+    if not playlist_id.strip():
+        raise SpotifyAuthError("playlist_id is required.")
+    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(
+        "POST",
+        f"/playlists/{playlist_id.strip()}/items",
+        json_body={"uris": clean_uris},
+    )
+    return {
+        "playlist_id": playlist_id.strip(),
+        "added": len(clean_uris),
+        "snapshot_id": payload.get("snapshot_id"),
+    }
 
-def search_tracks(query: str, limit: int = 5) -> dict[str, Any]:
-    return get_spotify().search_tracks(query=query, limit=limit)
 
+app = FastAPI(title="Spotify MCP Server", version="1.0.0")
+app.mount("/gradio_api/mcp", mcp.sse_app("/gradio_api/mcp"))
 
-def create_playlist(name: str, description: str = "", public: bool = False) -> dict[str, Any]:
-    return get_spotify().create_playlist(name=name, description=description, public=public)
 
+@app.get("/")
+def root() -> dict[str, Any]:
+    return {
+        "service": "spotify-mcp-server",
+        "auth_login_url": "/auth/login",
+        "auth_status_url": "/auth/status",
+        "mcp_sse_path": "/gradio_api/mcp/sse",
+        "public_mcp_sse_url": MCP_SSE_URL,
+    }
 
-def add_tracks_to_playlist(playlist_id: str, track_ids_csv: str) -> dict[str, Any]:
-    track_ids = [x.strip() for x in track_ids_csv.split(",") if x.strip()]
-    return get_spotify().add_tracks(playlist_id=playlist_id, track_ids=track_ids)
 
+@app.get("/health")
+def health() -> dict[str, str]:
+    return {"status": "ok"}
 
-def create_playlist_from_search(name: str, query: str, limit: int = 10, public: bool = False) -> dict[str, Any]:
-    spotify = get_spotify()
-    tracks = spotify.search_tracks(query=query, limit=limit).get("results", [])
-    if not tracks:
-        raise RuntimeError("No tracks found for query.")
-    playlist = spotify.create_playlist(name=name, description=f"Auto playlist for query: {query}", public=public)
-    added = spotify.add_tracks(playlist_id=playlist["id"], track_ids=[t["id"] for t in tracks])
-    return {"playlist": playlist, "added": added, "tracks": tracks}
 
+@app.get("/auth/status")
+def auth_status() -> dict[str, Any]:
+    return _token_status()
 
-auth_tab = gr.Interface(
-    fn=spotify_auth_url,
-    inputs=gr.Textbox(label="State (optional)", placeholder="optional_state"),
-    outputs=gr.JSON(label="Spotify Auth URL"),
-    title="Spotify Auth URL",
-    api_name="_spotify_auth_url",
-)
 
-exchange_tab = gr.Interface(
-    fn=spotify_exchange_code,
-    inputs=[
-        gr.Textbox(label="Authorization Code", placeholder="code from Spotify callback"),
-        gr.Textbox(label="State (optional)", placeholder="state from spotify_auth_url"),
-    ],
-    outputs=gr.JSON(label="Exchange Result"),
-    title="Spotify Exchange Code",
-    api_name="_spotify_exchange_code",
-)
+@app.get("/auth/login")
+def auth_login() -> RedirectResponse:
+    try:
+        state = _new_oauth_state()
+        url = _build_auth_url(state)
+    except SpotifyAuthError as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+    return RedirectResponse(url=url, status_code=307)
 
-me_tab = gr.Interface(
-    fn=spotify_me,
-    inputs=[],
-    outputs=gr.JSON(label="Profile"),
-    title="Spotify Me",
-    api_name="_spotify_me",
-)
 
-search_tab = gr.Interface(
-    fn=search_tracks,
-    inputs=[gr.Textbox(label="Query"), gr.Slider(1, 50, value=5, step=1, label="Limit")],
-    outputs=gr.JSON(label="Tracks"),
-    title="Search Tracks",
-    api_name="_search_tracks",
-)
+@app.get("/auth/callback", response_class=HTMLResponse)
+def auth_callback(
+    code: str | None = Query(default=None),
+    state: str | None = Query(default=None),
+    error: str | None = Query(default=None),
+) -> HTMLResponse:
+    if error:
+        raise HTTPException(status_code=400, detail=f"Spotify authorization error: {error}")
 
-create_playlist_tab = gr.Interface(
-    fn=create_playlist,
-    inputs=[
-        gr.Textbox(label="Playlist Name"),
-        gr.Textbox(label="Description", placeholder="optional", lines=2),
-        gr.Checkbox(label="Public", value=False),
-    ],
-    outputs=gr.JSON(label="Playlist"),
-    title="Create Playlist",
-    api_name="_create_playlist",
-)
+    if not code or not state:
+        raise HTTPException(status_code=400, detail="Missing code/state in callback.")
 
-add_tracks_tab = gr.Interface(
-    fn=add_tracks_to_playlist,
-    inputs=[
-        gr.Textbox(label="Playlist ID"),
-        gr.Textbox(label="Track IDs CSV", placeholder="id1,id2,id3 or spotify:track:..."),
-    ],
-    outputs=gr.JSON(label="Add Tracks Result"),
-    title="Add Tracks",
-    api_name="_add_tracks_to_playlist",
-)
+    if not _consume_oauth_state(state):
+        raise HTTPException(status_code=400, detail="Invalid or expired OAuth state.")
 
-auto_tab = gr.Interface(
-    fn=create_playlist_from_search,
-    inputs=[
-        gr.Textbox(label="Playlist Name"),
-        gr.Textbox(label="Search Query"),
-        gr.Slider(1, 50, value=10, step=1, label="Limit"),
-        gr.Checkbox(label="Public", value=False),
-    ],
-    outputs=gr.JSON(label="Result"),
-    title="Create Playlist From Search",
-    api_name="_create_playlist_from_search",
-)
+    try:
+        _exchange_code_for_tokens(code)
+    except SpotifyAuthError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    html = """
+    <html>
+      <head><title>Spotify conectado</title></head>
+      <body style='font-family: Arial, sans-serif; padding: 24px;'>
+        <h2>Spotify conectado correctamente</h2>
+        <p>Ya puedes volver a tu cliente MCP y usar las tools.</p>
+      </body>
+    </html>
+    """
+    return HTMLResponse(content=html, status_code=200)
 
-demo = gr.TabbedInterface(
-    [auth_tab, exchange_tab, me_tab, search_tab, create_playlist_tab, add_tracks_tab, auto_tab],
-    ["Auth URL", "Exchange Code", "Me", "Search", "Create Playlist", "Add Tracks", "Auto Playlist"],
-    theme=gr.themes.Base(),
-)
 
 if __name__ == "__main__":
-    demo.launch(mcp_server=True)
+    import uvicorn
+
+    port = int(os.getenv("PORT", "7860"))
+    uvicorn.run("app:app", host="0.0.0.0", port=port)