Update app.py

app.py

@@ -13,11 +13,20 @@ from fastapi import FastAPI, HTTPException, Query
 from fastapi.responses import HTMLResponse, RedirectResponse
 from mcp.server.fastmcp import FastMCP
 
+# MCP server for Spotify with OAuth2 and read/write tools.
+# This file exposes:
+# 1) Authentication helpers and Spotify HTTP calls.
+# 2) MCP tools registered with FastMCP.
+# 3) FastAPI endpoints for OAuth login and health/status checks.
+
+# Load environment variables from .env for local runs.
 load_dotenv()
 
+# Official Spotify base endpoints.
 SPOTIFY_ACCOUNTS_BASE = "https://accounts.spotify.com"
 SPOTIFY_API_BASE = "https://api.spotify.com/v1"
 
+# Main OAuth client configuration.
 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()
@@ -50,16 +59,19 @@ MCP_SSE_URL = os.getenv(
     "https://pharmaia-demo-mcp-server-spotify.hf.space/gradio_api/mcp/sse",
 ).strip()
 
-# Keep OAuth states in memory for CSRF protection.
+# In-memory OAuth states used to mitigate CSRF on callback.
 OAUTH_STATE_TTL_SECONDS = 600
 _oauth_states: dict[str, int] = {}
 
 
 class SpotifyAuthError(RuntimeError):
+    """Error de autenticacion/autorizacion Spotify con mensaje apto para cliente."""
+
     pass
 
 
 def _require_spotify_config() -> None:
+    """Validate that all required OAuth credentials are configured."""
     missing = []
     if not SPOTIFY_CLIENT_ID:
         missing.append("SPOTIFY_CLIENT_ID")
@@ -75,6 +87,7 @@ def _require_spotify_config() -> None:
 
 
 def _cleanup_expired_states() -> None:
+    """Remove expired OAuth states to avoid unbounded memory growth."""
     now = int(time.time())
     expired = [k for k, ts in _oauth_states.items() if (now - ts) > OAUTH_STATE_TTL_SECONDS]
     for key in expired:
@@ -82,6 +95,7 @@ def _cleanup_expired_states() -> None:
 
 
 def _new_oauth_state() -> str:
+    """Generate a new OAuth state and register it with current timestamp."""
     _cleanup_expired_states()
     state = secrets.token_urlsafe(24)
     _oauth_states[state] = int(time.time())
@@ -89,6 +103,7 @@ def _new_oauth_state() -> str:
 
 
 def _consume_oauth_state(state: str) -> bool:
+    """Consume a one-time OAuth state and validate it is not expired."""
     _cleanup_expired_states()
     ts = _oauth_states.pop(state, None)
     if ts is None:
@@ -97,11 +112,13 @@ def _consume_oauth_state(state: str) -> bool:
 
 
 def _spotify_basic_auth_header() -> str:
+    """Build Basic Auth credential (client_id:client_secret) in base64."""
     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:
+    """Read cached token file from disk; return None if missing or invalid."""
     if not SPOTIFY_TOKEN_FILE.exists():
         return None
     try:
@@ -111,6 +128,7 @@ def _load_tokens() -> dict[str, Any] | None:
 
 
 def _save_tokens(token_payload: dict[str, Any], previous: dict[str, Any] | None = None) -> dict[str, Any]:
+    """Persist tokens on disk, preserve refresh_token, and compute expires_at."""
     now = int(time.time())
     expires_in = int(token_payload.get("expires_in", 3600))
 
@@ -133,11 +151,13 @@ def _save_tokens(token_payload: dict[str, Any], previous: dict[str, Any] | None
 
 
 def _is_expired(tokens: dict[str, Any]) -> bool:
+    """Check access_token expiration with a 30-second safety margin."""
     expires_at = int(tokens.get("expires_at", 0))
     return int(time.time()) >= (expires_at - 30)
 
 
 def _exchange_code_for_tokens(code: str) -> dict[str, Any]:
+    """Exchange OAuth authorization code for access/refresh tokens."""
     _require_spotify_config()
 
     data = {
@@ -168,6 +188,7 @@ def _exchange_code_for_tokens(code: str) -> dict[str, Any]:
 
 
 def _refresh_access_token(tokens: dict[str, Any]) -> dict[str, Any]:
+    """Refresh access_token using refresh_token and persist the result."""
     refresh_token = tokens.get("refresh_token")
     if not refresh_token:
         raise SpotifyAuthError(
@@ -201,6 +222,7 @@ def _refresh_access_token(tokens: dict[str, Any]) -> dict[str, Any]:
 
 
 def _get_valid_access_token() -> str:
+    """Get a usable access_token; refresh automatically if expired."""
     _require_spotify_config()
 
     tokens = _load_tokens()
@@ -226,6 +248,10 @@ def _spotify_request(
     params: dict[str, Any] | None = None,
     json_body: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
+    """Execute authenticated Spotify API request with auto-refresh on 401.
+
+    Converts Spotify HTTP errors into SpotifyAuthError with useful details.
+    """
     token = _get_valid_access_token()
     headers = {"Authorization": f"Bearer {token}"}
 
@@ -281,6 +307,7 @@ def _spotify_request(
 
 
 def _build_auth_url(state: str, show_dialog: bool = True) -> str:
+    """Build Spotify /authorize OAuth URL with scopes and anti-CSRF state."""
     _require_spotify_config()
     params = {
         "response_type": "code",
@@ -294,6 +321,7 @@ def _build_auth_url(state: str, show_dialog: bool = True) -> str:
 
 
 def _token_status() -> dict[str, Any]:
+    """Summarize current authentication/token status for quick diagnostics."""
     tokens = _load_tokens() or {}
     authenticated = bool(tokens.get("access_token"))
     return {
@@ -305,12 +333,14 @@ def _token_status() -> dict[str, Any]:
 
 
 def _token_scope_set() -> set[str]:
+    """Return the set of scopes present in the persisted token."""
     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:
+    """Validate required scopes for a sensitive read/write action."""
     token_scopes = _token_scope_set()
     if not token_scopes:
         return
@@ -323,6 +353,7 @@ def _require_any_scope(required_scopes: list[str], action: str) -> None:
 
 
 def _short_playlist(playlist: dict[str, Any]) -> dict[str, Any]:
+    """Project playlist payload into the compact tool response format."""
     return {
         "id": playlist.get("id"),
         "name": playlist.get("name"),
@@ -335,6 +366,7 @@ def _short_playlist(playlist: dict[str, Any]) -> dict[str, Any]:
 
 
 def _short_track(track: dict[str, Any]) -> dict[str, Any]:
+    """Project track payload into the compact tool response format."""
     return {
         "id": track.get("id"),
         "name": track.get("name"),
@@ -346,6 +378,7 @@ def _short_track(track: dict[str, Any]) -> dict[str, Any]:
 
 
 def _normalize_track_id(track_ref: str) -> str:
+    """Normalize track ID from plain id, Spotify URI, or web URL."""
     value = track_ref.strip()
     if not value:
         return ""
@@ -360,6 +393,7 @@ def _normalize_track_id(track_ref: str) -> str:
 
 
 def _normalize_track_ids(track_ids: list[str], max_items: int = 500) -> list[str]:
+    """Normalize, deduplicate, and validate a list of track IDs."""
     normalized: list[str] = []
     seen: set[str] = set()
 
@@ -378,6 +412,7 @@ def _normalize_track_ids(track_ids: list[str], max_items: int = 500) -> list[str
 
 
 def _normalize_playlist_id(playlist_ref: str) -> str:
+    """Normalize playlist ID from plain id, Spotify URI, or web URL."""
     value = playlist_ref.strip()
     if not value:
         return ""
@@ -392,9 +427,11 @@ def _normalize_playlist_id(playlist_ref: str) -> str:
 
 
 def _chunked(items: list[str], size: int) -> list[list[str]]:
+    """Split a list into fixed-size chunks for Spotify API limits."""
     return [items[i : i + size] for i in range(0, len(items), size)]
 
 
+# MCP registration: defines name, description, and tool usage instructions.
 mcp = FastMCP(
     name="Spotify MCP Server",
     instructions=(
@@ -547,6 +584,7 @@ def spotify_get_top_tracks(
     }
 
 
+# MCP tools focused on playlist and library management.
 @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)."""
@@ -603,7 +641,7 @@ def spotify_remove_saved_tracks(track_ids: list[str]) -> dict[str, Any]:
 
 @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)."""
+    """Set saved state for tracks in user library (save/remove)."""
     if saved:
         result = spotify_save_tracks(track_ids)
         result["action"] = "saved"
@@ -792,12 +830,14 @@ def spotify_delete_playlist(playlist_id: str) -> dict[str, Any]:
     }
 
 
+# Main HTTP app: mounts SSE transport for MCP clients.
 app = FastAPI(title="Spotify MCP Server", version="1.0.0")
 app.mount("/gradio_api/mcp", mcp.sse_app("/gradio_api/mcp"))
 
 
 @app.get("/")
 def root() -> dict[str, Any]:
+    """Root endpoint with useful links and public MCP SSE URL."""
     return {
         "service": "spotify-mcp-server",
         "auth_login_url": "/auth/login",
@@ -810,16 +850,19 @@ def root() -> dict[str, Any]:
 
 @app.get("/health")
 def health() -> dict[str, str]:
+    """Healthcheck simple para monitoreo y despliegue."""
     return {"status": "ok"}
 
 
 @app.get("/auth/status")
 def auth_status() -> dict[str, Any]:
+    """Expose current persisted OAuth token status."""
     return _token_status()
 
 
 @app.get("/auth/reset")
 def auth_reset() -> dict[str, Any]:
+    """Reset authentication by removing the local token file."""
     removed = False
     if SPOTIFY_TOKEN_FILE.exists():
         try:
@@ -835,6 +878,10 @@ def auth_reset() -> dict[str, Any]:
 
 @app.get("/auth/login")
 def auth_login(force: bool = Query(default=True)) -> RedirectResponse:
+    """Start OAuth flow by redirecting to Spotify /authorize.
+
+    `force=true` shows consent dialog to ensure updated scopes.
+    """
     try:
         state = _new_oauth_state()
         url = _build_auth_url(state, show_dialog=force)
@@ -849,6 +896,7 @@ def auth_callback(
     state: str | None = Query(default=None),
     error: str | None = Query(default=None),
 ) -> HTMLResponse:
+    """Handle OAuth callback: validate state, exchange code, and return HTML."""
     if error:
         raise HTTPException(status_code=400, detail=f"Spotify authorization error: {error}")
 
@@ -876,6 +924,7 @@ def auth_callback(
 
 
 if __name__ == "__main__":
+    # Local entrypoint to run the server with Uvicorn.
     import uvicorn
 
     port = int(os.getenv("PORT", "7860"))