File size: 5,983 Bytes
5899d68
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
"""
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)