mcp_spaces_finder.py

"""
Simple HuggingFace MCP Spaces Finder Module - Corrected Version
A minimal module to discover MCP servers on HuggingFace Spaces.
Fixed to fetch ALL available MCP servers using proper pagination.
Usage:
    from mcp_spaces_finder import create_simple_mcp_selector
    
    # One-liner in your Gradio app
    dropdown, textbox = create_simple_mcp_selector()
"""
import gradio as gr
from huggingface_hub import list_spaces
import time
from typing import List, Tuple

class SimpleMCPFinder:
    """Simple MCP spaces finder with caching and proper pagination."""
    
    def __init__(self, cache_duration: int = 300):
        self.cache = None
        self.cache_time = None
        self.cache_duration = cache_duration
    
    def get_mcp_spaces(self) -> List[str]:
        """Get list of ALL running MCP space IDs using proper pagination."""
        
        # Check cache
        if (self.cache is not None and 
            self.cache_time is not None and 
            time.time() - self.cache_time < self.cache_duration):
            return self.cache
        
        print("Fetching ALL MCP spaces...")
        
        # Get ALL MCP spaces by setting a much higher limit
        # The HF API supports pagination, so we set limit high enough to get all
        spaces = list(list_spaces(
            filter="mcp-server",
            sort="likes",
            direction=-1,
            limit=5000,  # Increased from 1000 to capture all ~2500 spaces
            full=True
        ))
        
        # Extract just the space IDs 
        space_ids = [space.id for space in spaces]
        
        # Cache results
        self.cache = space_ids
        self.cache_time = time.time()
        
        print(f"Found {len(space_ids)} MCP spaces")
        return space_ids
    
    def get_mcp_spaces_paginated(self) -> List[str]:
        """Alternative method: Get ALL MCP spaces using explicit pagination if needed."""
        
        # Check cache
        if (self.cache is not None and 
            self.cache_time is not None and 
            time.time() - self.cache_time < self.cache_duration):
            return self.cache
        
        print("Fetching ALL MCP spaces with pagination...")
        
        all_space_ids = []
        limit_per_page = 1000
        
        # Keep fetching until we get all spaces
        # Note: HuggingFace API handles pagination internally with the iterator
        try:
            spaces = list(list_spaces(
                filter="mcp-server",
                sort="likes",
                direction=-1,
                limit=None,  # No limit to get all
                full=True
            ))
            all_space_ids = [space.id for space in spaces]
            
        except Exception as e:
            print(f"Error with unlimited fetch, trying with high limit: {e}")
            # Fallback to high limit
            spaces = list(list_spaces(
                filter="mcp-server", 
                sort="likes",
                direction=-1,
                limit=5000,  # High limit as fallback
                full=True
            ))
            all_space_ids = [space.id for space in spaces]
        
        # Cache results
        self.cache = all_space_ids
        self.cache_time = time.time()
        
        print(f"Found {len(all_space_ids)} MCP spaces total")
        return all_space_ids

# Global instance
_finder = SimpleMCPFinder()

def create_simple_mcp_selector(
    dropdown_label: str = "🤖 Select MCP Server",
    textbox_label: str = "Selected MCP Server",
    placeholder: str = "No server selected"
) -> Tuple[gr.Dropdown, gr.Textbox]:
    """
    Create a simple MCP selector with dropdown and textbox.
    
    Args:
        dropdown_label (str): Label for the dropdown
        textbox_label (str): Label for the textbox  
        placeholder (str): Placeholder text when nothing selected
        
    Returns:
        Tuple[gr.Dropdown, gr.Textbox]: The dropdown and textbox components
    """
    
    # Get MCP spaces
    spaces = _finder.get_mcp_spaces()
    
    # Create dropdown with space choices
    dropdown = gr.Dropdown(
        choices=spaces,
        label=f"{dropdown_label} ({len(spaces)} available)",
        value=None,
        allow_custom_value=True,  # Allow users to type custom space IDs
        info="Choose from discovered MCP spaces or type a custom space ID"
    )
    
    # Create textbox to display selection
    textbox = gr.Textbox(
        label=textbox_label,
        value=placeholder,
        interactive=False
    )
    
    # Connect dropdown to textbox
    def update_textbox(selected_value):
        return selected_value if selected_value else placeholder
    
    dropdown.change(
        fn=update_textbox,
        inputs=[dropdown], 
        outputs=[textbox]
    )
    
    return dropdown, textbox

def refresh_mcp_spaces():
    """Clear cache to force refresh."""
    _finder.cache = None
    _finder.cache_time = None 

def test_space_exists(space_id: str) -> bool:
    """Test if a specific space exists in our discovered list."""
    spaces = _finder.get_mcp_spaces()
    return space_id in spaces

def debug_search_for_spaces(space_ids: List[str]):
    """Debug function to check if specific spaces are found."""
    spaces = _finder.get_mcp_spaces()
    print(f"Total MCP spaces found: {len(spaces)}")
    
    for space_id in space_ids:
        if space_id in spaces:
            print(f"✅ Found: {space_id}")
        else:
            print(f"❌ Missing: {space_id}")
    
    # Show first 10 spaces for reference
    print(f"\nFirst 10 spaces found:")
    for i, space in enumerate(spaces[:10]):
        print(f"  {i+1}. {space}")

if __name__ == "__main__":
    # Test the specific spaces you mentioned
    test_spaces = [
        "ysharma/Kokoro-TTS-mcp-test",
        "ysharma/ltx-video-distilled", 
        "ysharma/dalle-3-xl-lora-v2"
    ]
    
    print("Testing MCP space discovery...")
    debug_search_for_spaces(test_spaces)