Update app.py

app.py

@@ -1,550 +1,360 @@
-import gradio as gr
-from llama_index.core import VectorStoreIndex
-from llama_index.core import (
-    StorageContext,
-    load_index_from_storage,
-)
-from llama_index.tools.arxiv import ArxivToolSpec
-from llama_index.core import Settings
-from llama_index.llms.azure_openai import AzureOpenAI
-from llama_index.embeddings.azure_openai import AzureOpenAIEmbedding
-from llama_index.llms.openai import OpenAI
-from llama_index.embeddings.openai import OpenAIEmbedding
-from typing import Optional, List, Dict, Any
-from pathlib import Path
-import aiohttp
+import base64
 import json
 import os
-import asyncio
+import secrets
+import time
+import urllib.parse
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
 
+import gradio as gr
+import requests
+from dotenv import load_dotenv
 
-from gradio_client import Client, handle_file
-HF_TOKEN = os.environ.get('HF_TOKEN')
 
+load_dotenv()
 
+SPOTIFY_ACCOUNTS_BASE = "https://accounts.spotify.com"
+SPOTIFY_API_BASE = "https://api.spotify.com/v1"
 
-##### LLM #####
-openai_api_key = os.environ.get('OPENAI_API_KEY')
 
+@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
 
-llm = OpenAI(
-    model="gpt-4.1",
-    api_key=openai_api_key,
-)
-embed_model = OpenAIEmbedding(
-    model="text-embedding-ada-002",
-    api_key=openai_api_key,
-)
 
-Settings.llm = llm
-Settings.embed_model = embed_model
-##### END LLM #####
-
-
-
-##### LOAD RETRIEVERS #####
-DOCUMENTS_BASE_PATH = "./"
-RETRIEVERS_JSON_PATH = Path("./retrievers.json")
-
-# Load metadata
-def load_retrievers_metadata():
-    try:
-        with open(RETRIEVERS_JSON_PATH, 'r', encoding='utf-8') as f:
-            return json.load(f)
-    except Exception as e:
-        print(f"Error loading retrievers.json: {str(e)}")
-        print(f"Error details: {traceback.format_exc()}") # You would need to import traceback
-        return {}
-
-retrievers_metadata = load_retrievers_metadata()
-SOURCES = {source: f"{source.lower()}/" for source in retrievers_metadata.keys()}
-
-# Load indexes
-indices: Dict[str, VectorStoreIndex] = {}
-
-for source, rel_path in SOURCES.items():
-    full_path = os.path.join(DOCUMENTS_BASE_PATH, rel_path)
-    if not os.path.exists(full_path):
-        print(f"Warning: Path not found for {source}")
-        continue
-
-    for root, dirs, files in os.walk(full_path):
-        if "storage_nodes" in dirs:
-            try:
-                storage_path = os.path.join(root, "storage_nodes")
-                storage_context = StorageContext.from_defaults(persist_dir=storage_path)
-                index_name = os.path.basename(root)
-                indices[index_name] = load_index_from_storage(storage_context) #, index_id="vector_index"
-                print(f"Index loaded successfully: {index_name}")
-            except Exception as e:
-                print(f"Error loading index {index_name}: {str(e)}")
-                print(f"Error details: {traceback.format_exc()}")
-                
-
-
-
-
-
-##### ARXIV INSTANCE #####
-arxiv_tool = ArxivToolSpec(max_results=5).to_tool_list()[0]
-arxiv_tool.return_direct = True
-
-
-
-##### MCP TOOLS #####
-
-async def search_arxiv(
-    query: str,
-    max_results: int = 5
-) -> Dict[str, Any]:
-    """
-    Searches for academic papers on ArXiv.
-    
-    Args:
-        query: Search terms (e.g. "deep learning")
-        max_results: Maximum number of results (1-10, default 5)
-    
-    Returns:
-        Dict: Search results with paper metadata
-    """
-    try:
-        # Configure maximum results
-        max_results = min(max(1, max_results), 10)
-        arxiv_tool.metadata.max_results = max_results
-        
-        # Execute search and get results
-        tool_output = arxiv_tool(query=query)
-        
-        # Process documents
-        papers = []
-        for doc in tool_output.raw_output:  # Correctly access documents
-            content = doc.text_resource.text.split('\n')
-            papers.append({
-                'title': content[0].split(': ')[1] if ': ' in content[0] else content[0],
-                'abstract': '\n'.join(content[1:]).strip(),
-                'pdf_url': content[0].split(': ')[0].replace('http://', 'https://'),
-                'arxiv_id': content[0].split(': ')[0].split('/')[-1].replace('v1', '')
-            })
-        
+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 _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 _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 _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
+
+    def _request_token(self, data: dict[str, Any]) -> dict[str, Any]:
+        response = requests.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,
+        )
+        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."
+        )
+
+    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")
+
+        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",
+                }
+            )
+        )
         return {
-            'papers': papers,
-            'count': len(papers),
-            'query': query,
-            'status': 'success'
+            "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.",
         }
-        
-    except Exception as e:
+
+    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,
+            }
+        )
+        self._write_token_data(token_data)
+        me = self.me()
         return {
-            'papers': [],
-            'count': 0,
-            'query': query,
-            'status': 'error',
-            'error': str(e)
+            "status": "ok",
+            "user_id": me.get("id"),
+            "display_name": me.get("display_name"),
+            "has_refresh_token": bool(token_data.get("refresh_token")),
         }
 
-async def list_retrievers(source: str = None) -> dict:
-    """
-    Returns the list of available retrievers.
-    If a source is specified and exists, filters by it; if it doesn't exist, returns all.
-    
-    Args:
-        source (str, optional): Source to filter by. If it doesn't exist, it will be ignored. Defaults to None.
-    
-    Returns:
-        dict: {
-            "retrievers": List of retrievers (filtered or complete),
-            "count": Total count,
-            "status": "success"|"error",
-            "source_requested": source,  # Shows what was requested
-            "source_used": "all"|source  # Shows what was actually used
-        }
-    """
-    try:
-        available = []
-        source_exists = source in retrievers_metadata if source else False
-        
-        for current_source, indexes in retrievers_metadata.items():
-            # Only filter if source exists, otherwise show all
-            if source_exists and current_source != source:
-                continue
-                
-            for index_name, metadata in indexes.items():
-                available.append({
-                    "name": index_name,
-                    "source": current_source,
-                    "title": metadata.get("title", ""),
-                    "description": metadata.get("description", "")
-                })
-        
+    def me(self) -> dict[str, Any]:
+        me = self._request("GET", "/me")
         return {
-            "retrievers": available,
-            "count": len(available),
-            "status": "success",
-            "source_requested": source,
-            "source_used": source if source_exists else "all"
+            "id": me.get("id"),
+            "display_name": me.get("display_name"),
+            "email": me.get("email"),
+            "product": me.get("product"),
         }
-    except Exception as e:
+
+    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))},
+        )
+        items = payload.get("tracks", {}).get("items", [])
         return {
-            "retrievers": [],
-            "count": 0,
-            "status": "error",
-            "error": str(e),
-            "source_requested": source,
-            "source_used": "none"
+            "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 retrieve_docs(
-    query: str,
-    retrievers: List[str],
-    top_k: int = 3
-) -> dict:
-    """
-    Performs semantic search on indexed documents.
-    
-    Parameters:
-        query (str): Search text (required)
-        retrievers (List[str]): Names of retrievers to query (required)
-        top_k (int): Number of results per retriever (optional, default=3)
-    """
-    print(f"Starting search for query: '{query}'")
-    print(f"Parameters - retrievers: {retrievers}, top_k: {top_k}")
-    
-    results = {}
-    invalid = []
-    
-    for name in retrievers:
-        if name not in indices:
-            print(f"Retriever not found: {name}")
-            invalid.append(name)
-            continue
-            
-        try:
-            print(f"Processing retriever: {name}")
-            retriever = indices[name].as_retriever(similarity_top_k=top_k)
-            nodes = retriever.retrieve(query)
-            print(f"Retrieved {len(nodes)} documents from {name}")
-            
-            # 2. Search for COMPLETE metadata
-            metadata = {}
-            source = "unknown"
-            for src, indexes in retrievers_metadata.items():
-                if name in indexes:
-                    metadata = indexes[name]
-                    source = src
-                    break
-            print(f"Metadata found for {name}: {metadata.keys()}")
-            
-            # 3. Build response
-            results[name] = {
-                "title": metadata.get("title", name),
-                "documents": [
-                    {
-                        "content": node.get_content(),
-                        "metadata": node.metadata,
-                        "score": node.score
-                    }
-                    for node in nodes
-                ],
-                "description": metadata.get("description", ""),
-                "source": source,
-                "last_updated": metadata.get("last_updated", "")
-            }
-            print(f"Retriever {name} processed successfully")
-            
-        except Exception as e:
-            print(f"Error processing retriever {name}: {str(e)}", exc_info=True)
-            results[name] = {
-                "error": str(e),
-                "retriever": name
-            }
-    
-    # Build final response
-    response = {
-        "query": query,
-        "results": results,
-        "top_k": top_k,
-    }
-    
-    if invalid:
-        print(f"Invalid retrievers: {invalid}. Valid options: {list(indices.keys())}")
-        response["warnings"] = {
-            "invalid_retrievers": invalid,
-            "valid_options": list(indices.keys())
-        }
-    
-    print(f"Search completed. Total results: {len(results)}")
-    return response
-
-
-async def search_tavily(
-    query: str,
-    days: int = 7,
-    max_results: int = 1,
-    include_answer: bool = False
-) -> dict:
-    """Perform a web search using the Tavily API.
-    
-    Args:
-        query: Search query string (required)
-        days: Restrict search to last N days (default: 7)
-        max_results: Maximum results to return (default: 1)
-        include_answer: Include a direct answer only when requested by the user (default: False)
-        
-    Returns:
-        dict: Search results from Tavily
-    """
-    # Get API key from environment variables
-    tavily_api_key = os.environ.get('TAVILY_API_KEY')
-    if not tavily_api_key:
-        raise ValueError("TAVILY_API_KEY environment variable not set")
-    
-    headers = {
-        "Authorization": f"Bearer {tavily_api_key}",
-        "Content-Type": "application/json"
-    }
-    
-    payload = {
-        "query": query,
-        "search_depth": "basic",
-        "max_results": max_results,
-        "days": days if days else None,
-        "include_answer": include_answer
-    }
-    
-    try:
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
-                "https://api.tavily.com/search",
-                headers=headers,
-                json=payload
-            ) as response:
-                response.raise_for_status()
-                result = await response.json()
-                return result
-                
-    except Exception as e:
+    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},
+        )
         return {
-            "error": str(e),
-            "status": "failed",
-            "query": query
+            "id": playlist.get("id"),
+            "name": playlist.get("name"),
+            "url": playlist.get("external_urls", {}).get("spotify"),
         }
 
-##### EVALS #####
-async def evaluate_answer_relevancy(
-    query: str,
-    response: str,
-) -> float:
-    """Evaluate how relevant the answer is to the query using AnswerRelevancyEvaluator.
-    
-    Args:
-        query: Original user query (required)
-        response: Generated response to evaluate (required)
-        
-    Returns:
-        float: Relevancy score between 0 and 1 (higher is better)
-    """
-    try:
-        from llama_index.core.evaluation import AnswerRelevancyEvaluator
-        
-        # Initialize the evaluator
-        evaluator = AnswerRelevancyEvaluator(llm=llm)
-        
-        # Perform the evaluation
-        eval_result = evaluator.evaluate(query=query, response=response)
-        
-        # Return the score as a float
-        return float(eval_result.score)
-        
-    except Exception as e:
-        # In case of error, return 0.0 (minimum score) and log the error
-        print(f"Error in relevancy evaluation: {str(e)}")
-        return 0.0
-
-async def evaluate_context_relevancy(
-    context: str,
-    query: str,
-    response: str
-) -> float:
-    """Evaluates the relevance of the response considering both the query and the context.
-    
-    Args:
-        context: Contextual information / knowledge base (required)
-        query: Original user query (required)
-        response: Generated response to evaluate (required)
-        
-    Returns:
-        float: Relevance score between 0 and 1 (higher is better)
-    """
-    try:
-        from llama_index.core.evaluation import ContextRelevancyEvaluator
-
-        # Initialize the relevancy evaluator with context
-        evaluator = ContextRelevancyEvaluator(llm=llm)
-        
-        # Perform the evaluation (adapted to handle context)
-        eval_result = evaluator.evaluate(
-            query=query,
-            response=response,
-            contexts=[context]
-        )
-        
-        return float(eval_result.score)
-        
-    except Exception as e:
-        print(f"Error during context relevancy evaluation: {str(e)}")
-        return 0.0
-
-async def evaluate_faithfulness(
-    query: str,
-    response: str,
-    context: str
-) -> float:
-    """Evaluate how faithful (factually consistent) the response is to the provided context.
-    
-    Args:
-        query: Original user query (required)
-        response: Generated response to evaluate (required)
-        context: Source context/knowledge base used for the response (required)
-        
-    Returns:
-        float: Faithfulness score between 0 and 1 (higher is better)
-    """
-    try:
-        from llama_index.core.evaluation import FaithfulnessEvaluator
-        
-        # Initialize evaluator
-        evaluator = FaithfulnessEvaluator(llm=llm)
-        
-        # Perform evaluation
-        eval_result = evaluator.evaluate(
-            query=query,
-            response=response,
-            contexts=[context]
-        )
-        
-        # Return score as float
-        return float(eval_result.score)
-        
-    except Exception as e:
-        # On error, return 0.0 (minimum score) and log the error
-        print(f"Error in faithfulness evaluation: {str(e)}")
-        return 0.0
+    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,
+    )
 
 
+spotify = SpotifyClient(load_config())
 
 
+def spotify_auth_url(state: str = "") -> dict[str, Any]:
+    return spotify.auth_url(state=state or None)
 
 
+def spotify_exchange_code(code: str, state: str = "") -> dict[str, Any]:
+    return spotify.exchange_code(code=code, state=state or None)
 
 
+def spotify_me() -> dict[str, Any]:
+    return spotify.me()
 
-# Gradio interface
-with gr.Blocks(title="MCP Tools", theme=gr.themes.Base()) as arxiv_tab:
-    arxiv_interface = gr.Interface(
-        fn=search_arxiv,
-        inputs=[
-            gr.Textbox(label="Search terms", placeholder="E.g.: deep learning"),
-            gr.Slider(1, 10, value=5, step=1, label="Maximum number of results")
-        ],
-        outputs=gr.JSON(label="Search results"),
-        title="ArXiv Search",
-        description="Search for academic papers on ArXiv using keywords.",
-        api_name="_search_arxiv"
-    )
 
-with gr.Blocks(title="MCP Tools", theme=gr.themes.Base()) as list_retrievers_tab:
-    retrievers_interface = gr.Interface(
-        fn=list_retrievers,
-        inputs=gr.Textbox(label="Source (optional)", placeholder="Leave empty to list all"),
-        outputs=gr.JSON(label="List of retrievers"),
-        title="List of Retrievers",
-        description="Shows available retrievers, optionally filtered by source.",
-        api_name="_list_retrievers"
-    )
+def search_tracks(query: str, limit: int = 5) -> dict[str, Any]:
+    return spotify.search_tracks(query=query, limit=limit)
 
-with gr.Blocks(title="MCP Tools", theme=gr.themes.Base()) as tavily_tab:
-    tavily_interface = gr.Interface(
-        fn=search_tavily,
-        inputs=[
-            gr.Textbox(label="Search query", placeholder="E.g.: latest news about AI"),
-            gr.Slider(1, 30, value=7, step=1, label="Last N days (0 for no limit)"),
-            gr.Slider(1, 10, value=1, step=1, label="Maximum results"),
-            gr.Checkbox(label="Include direct answer", value=False)
-        ],
-        outputs=gr.JSON(label="Tavily results"),
-        title="Web Search (Tavily)",
-        description="Perform web searches using the Tavily API.",
-        api_name="_search_tavily"
-    )
-    
-with gr.Blocks(title="MCP Tools", theme=gr.themes.Base()) as retrieve_tab:
-    # Interface for retrieve_docs
-    retrieve_interface = gr.Interface(
-        fn=retrieve_docs,
-        inputs=[
-            gr.Textbox(label="Query", placeholder="Enter your question or search terms..."),
-            gr.Dropdown(
-                choices=list(indices.keys()),
-                label="Retrievers",
-                multiselect=True,
-                info="Select one or more retrievers"
-            ),
-            gr.Slider(1, 10, value=3, step=1, label="Number of results per retriever (top_k)")
-        ],
-        outputs=gr.JSON(label="Semantic search results"),
-        title="Semantic Document Search",
-        description="""Perform semantic search on indexed documents using retrievers.
-                    Select available retrievers and adjust the number of results.""",
-        api_name="_retrieve"
-    )
 
-with gr.Blocks(title="MCP Tools", theme=gr.themes.Base()) as asw_relevance_tab:
-    relevancy_interface = gr.Interface(
-        fn=evaluate_answer_relevancy,
-        inputs=[
-            gr.Textbox(label="Original Query", placeholder="E.g.: How does photosynthesis work?"),
-            gr.Textbox(label="Answer to Evaluate", placeholder="Paste the generated answer here", lines=5),
-        ],
-        outputs=gr.Number(label="Relevancy Score (0-1)", precision=3),
-        title="Relevancy Evaluator (Query-Answer)",
-        description="Evaluates how relevant an answer is to the original query (1 = perfectly relevant).",
-        api_name="_evaluate_relevancy"
-    )
+def create_playlist(name: str, description: str = "", public: bool = False) -> dict[str, Any]:
+    return spotify.create_playlist(name=name, description=description, public=public)
 
-with gr.Blocks(title="MCP Tools", theme=gr.themes.Base()) as ctx_relevance_tab:
-    context_relevancy_interface = gr.Interface(
-        fn=evaluate_context_relevancy,
-        inputs=[
-            gr.Textbox(label="Context", placeholder="Relevant text / knowledge base", lines=3),
-            gr.Textbox(label="Original Query", placeholder="What question is being answered?"),
-            gr.Textbox(label="Generated Answer", placeholder="The answer to evaluate", lines=5),
-        ],
-        outputs=gr.Number(label="Relevancy Score (0-1)", precision=3),
-        title="Relevancy Evaluator (Context-Query-Answer)",
-        description="Evaluates how relevant the answer is considering both the query and the reference context.",
-        api_name="_evaluate_context_relevancy"
-    )
 
-with gr.Blocks(title="MCP Tools", theme=gr.themes.Base()) as faithfulness_tab:
-    faithfulness_interface = gr.Interface(
-        fn=evaluate_faithfulness,
-        inputs=[
-            gr.Textbox(label="Original Query", placeholder="E.g.: What are the causes of climate change?"),
-            gr.Textbox(label="Answer to Evaluate", placeholder="Paste the generated answer here", lines=5),
-            gr.Textbox(label="Context", placeholder="Reference text / knowledge base", lines=3),
-        ],
-        outputs=gr.Number(label="Faithfulness Score (0-1)", precision=3),
-        title="Faithfulness Evaluator",
-        description="Evaluates how faithful/factually consistent the answer is with respect to the provided context (1 = perfectly faithful).",
-        api_name="_evaluate_faithfulness"
-    )
+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 spotify.add_tracks(playlist_id=playlist_id, track_ids=track_ids)
+
+
+def create_playlist_from_search(name: str, query: str, limit: int = 10, public: bool = False) -> dict[str, Any]:
+    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}
+
+
+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",
+)
+
+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",
+)
+
+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",
+)
+
+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",
+)
+
+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",
+)
 
-# Create the interface with separate tabs
 demo = gr.TabbedInterface(
-    [arxiv_tab, tavily_tab, list_retrievers_tab, retrieve_tab, asw_relevance_tab, ctx_relevance_tab, faithfulness_tab],
-    ["ArXiv", "Tavily", "List Retrievers", "Retrieve", "Answer Relevance", "Context Relevance", "Faithfulness"],
+    [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(),
 )
 
-demo.launch(mcp_server=True)
+if __name__ == "__main__":
+    demo.launch(mcp_server=True)