Create mcp_spaces_finder.py

mcp_spaces_finder.py

@@ -0,0 +1,187 @@
+"""
+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)
+