ysharma HF Staff commited on
Commit
230e011
·
verified ·
1 Parent(s): 04ca5c4

Update mcp_spaces_finder.py

Browse files
Files changed (1) hide show
  1. mcp_spaces_finder.py +68 -299
mcp_spaces_finder.py CHANGED
@@ -1,339 +1,108 @@
1
  """
2
- HuggingFace MCP Spaces Finder Module
3
-
4
- A reusable module to discover and list Model Context Protocol (MCP) servers
5
- available on HuggingFace Spaces.
6
-
7
  Usage:
8
- from mcp_spaces_finder import MCPSpacesFinder, create_mcp_dropdown
9
 
10
- # In your Gradio app
11
- mcp_dropdown = create_mcp_dropdown()
12
  """
13
 
14
  import gradio as gr
15
- from huggingface_hub import HfApi, list_spaces
16
  import time
17
- from typing import List, Dict, Optional, Tuple
18
- from dataclasses import dataclass
19
  from concurrent.futures import ThreadPoolExecutor, as_completed
20
 
21
- @dataclass
22
- class MCPSpace:
23
- id: str
24
- title: str
25
- author: str
26
- likes: int
27
- status: str
28
- url: str
29
- description: str
30
- sdk: str
31
- last_modified: str
32
- created_at: str
33
-
34
- class MCPSpacesFinder:
35
- """
36
- A class to find and manage MCP-enabled spaces on HuggingFace Hub.
37
- """
38
 
39
  def __init__(self, cache_duration: int = 300):
40
- """
41
- Initialize the MCP Spaces Finder.
42
-
43
- Args:
44
- cache_duration (int): Cache duration in seconds (default: 5 minutes)
45
- """
46
- self.api = HfApi()
47
- self.mcp_spaces_cache = None
48
- self.cache_timestamp = None
49
  self.cache_duration = cache_duration
50
-
51
- def get_space_status(self, space_id: str) -> str:
52
- """Get the current runtime status of a space."""
53
- try:
54
- runtime = self.api.get_space_runtime(space_id)
55
- if hasattr(runtime, 'stage'):
56
- return runtime.stage
57
- return "unknown"
58
- except Exception:
59
- return "error"
60
 
61
- def process_space_batch(self, spaces_batch) -> List[MCPSpace]:
62
- """Process a batch of spaces to get their runtime status."""
63
- processed_spaces = []
64
 
65
- for space in spaces_batch:
66
- try:
67
- space_id = space.id
68
- status = self.get_space_status(space_id)
69
-
70
- processed_space = MCPSpace(
71
- id=space_id,
72
- title=getattr(space, 'title', space_id.split('/')[-1]) or space_id.split('/')[-1],
73
- author=space.author,
74
- likes=getattr(space, 'likes', 0),
75
- status=status,
76
- url=f"https://huggingface.co/spaces/{space_id}",
77
- description=getattr(space, 'description', 'No description available') or 'No description available',
78
- sdk=getattr(space, 'sdk', 'unknown'),
79
- last_modified=str(getattr(space, 'last_modified', 'unknown')),
80
- created_at=str(getattr(space, 'created_at', 'unknown'))
81
- )
82
- processed_spaces.append(processed_space)
83
-
84
- except Exception as e:
85
- print(f"Error processing space {getattr(space, 'id', 'unknown')}: {e}")
86
- continue
87
-
88
- return processed_spaces
89
-
90
- def find_mcp_spaces(self, limit: int = 2000, running_only: bool = True) -> List[MCPSpace]:
91
- """
92
- Find MCP-enabled spaces using the official mcp-server filter.
93
-
94
- Args:
95
- limit (int): Maximum number of spaces to fetch
96
- running_only (bool): If True, only return running/building spaces
97
-
98
- Returns:
99
- List[MCPSpace]: List of MCP spaces sorted by likes
100
- """
101
 
102
- # Check cache first
103
- if (self.mcp_spaces_cache is not None and
104
- self.cache_timestamp is not None and
105
- time.time() - self.cache_timestamp < self.cache_duration):
106
- return self.mcp_spaces_cache
107
 
108
- print("Fetching MCP spaces from HuggingFace Hub...")
109
-
110
- # Get all spaces with the official mcp-server tag, sorted by likes
111
  spaces = list(list_spaces(
112
  filter="mcp-server",
113
- sort="likes",
114
  direction=-1,
115
- limit=limit,
116
  full=True
117
  ))
118
 
119
- print(f"Found {len(spaces)} spaces with mcp-server tag")
120
-
121
- # Process spaces in batches using ThreadPoolExecutor for status checking
122
- batch_size = 50
123
- space_batches = [spaces[i:i + batch_size] for i in range(0, len(spaces), batch_size)]
124
-
125
- all_processed_spaces = []
126
-
127
- with ThreadPoolExecutor(max_workers=5) as executor:
128
- future_to_batch = {
129
- executor.submit(self.process_space_batch, batch): batch
130
- for batch in space_batches
131
- }
132
-
133
- for future in as_completed(future_to_batch):
134
- try:
135
- batch_results = future.result()
136
- all_processed_spaces.extend(batch_results)
137
- except Exception as e:
138
- print(f"Error processing batch: {e}")
139
-
140
- if running_only:
141
- # Filter to only include running or building spaces
142
- filtered_spaces = [
143
- space for space in all_processed_spaces
144
- if space.status in ["RUNNING", "RUNNING_BUILDING", "BUILDING"]
145
- ]
146
- else:
147
- filtered_spaces = all_processed_spaces
148
 
149
- # Sort by likes descending
150
- filtered_spaces.sort(key=lambda x: x.likes, reverse=True)
 
151
 
152
- # Cache the results
153
- self.mcp_spaces_cache = filtered_spaces
154
- self.cache_timestamp = time.time()
155
-
156
- print(f"Found {len(filtered_spaces)} {'running/building' if running_only else 'total'} MCP spaces")
157
- return filtered_spaces
158
-
159
- def get_dropdown_choices(self, running_only: bool = True) -> Tuple[List[Tuple[str, str]], int]:
160
- """
161
- Get dropdown choices for Gradio dropdown component.
162
-
163
- Args:
164
- running_only (bool): If True, only return running/building spaces
165
-
166
- Returns:
167
- Tuple[List[Tuple[str, str]], int]: (choices, total_count)
168
- """
169
- spaces = self.find_mcp_spaces(running_only=running_only)
170
-
171
- choices = [(space.id, space.id) for space in spaces]
172
- return choices, len(spaces)
173
-
174
- def get_space_info(self, space_id: str) -> Optional[MCPSpace]:
175
- """
176
- Get detailed information about a specific space.
177
-
178
- Args:
179
- space_id (str): The space ID (username/spacename)
180
-
181
- Returns:
182
- Optional[MCPSpace]: Space information or None if not found
183
- """
184
- spaces = self.mcp_spaces_cache or self.find_mcp_spaces()
185
- return next((space for space in spaces if space.id == space_id), None)
186
-
187
- def clear_cache(self):
188
- """Clear the cache to force refresh on next request."""
189
- self.mcp_spaces_cache = None
190
- self.cache_timestamp = None
191
 
192
- # Global instance for easy usage
193
- _default_finder = MCPSpacesFinder()
194
 
195
- def create_mcp_dropdown(
196
- label: str = "🤖 Select MCP Server",
197
- info: str = "Choose a running MCP server",
198
- running_only: bool = True,
199
- **kwargs
200
- ) -> gr.Dropdown:
201
  """
202
- Create a Gradio dropdown populated with MCP spaces.
203
 
204
  Args:
205
- label (str): Label for the dropdown
206
- info (str): Info text for the dropdown
207
- running_only (bool): If True, only show running/building spaces
208
- **kwargs: Additional arguments passed to gr.Dropdown
209
 
210
  Returns:
211
- gr.Dropdown: Configured dropdown component
212
  """
213
- choices, count = _default_finder.get_dropdown_choices(running_only=running_only)
214
 
215
- status_text = "running/building" if running_only else "total"
216
- final_label = f"{label} ({count} {status_text})"
217
 
218
- return gr.Dropdown(
219
- choices=choices,
220
- label=final_label,
221
- info=info,
222
- value=None,
223
- **kwargs
224
  )
225
-
226
- def get_mcp_space_info(space_id: str) -> Optional[MCPSpace]:
227
- """
228
- Get information about a specific MCP space.
229
-
230
- Args:
231
- space_id (str): The space ID (username/spacename)
232
-
233
- Returns:
234
- Optional[MCPSpace]: Space information or None if not found
235
- """
236
- return _default_finder.get_space_info(space_id)
237
-
238
- def refresh_mcp_spaces() -> Tuple[List[Tuple[str, str]], int]:
239
- """
240
- Refresh the MCP spaces cache and return updated choices.
241
-
242
- Returns:
243
- Tuple[List[Tuple[str, str]], int]: (choices, total_count)
244
- """
245
- _default_finder.clear_cache()
246
- return _default_finder.get_dropdown_choices()
247
-
248
- def get_mcp_config_text(space_id: str) -> str:
249
- """
250
- Generate MCP configuration text for a given space.
251
 
252
- Args:
253
- space_id (str): The space ID (username/spacename)
254
-
255
- Returns:
256
- str: Formatted MCP configuration instructions
257
- """
258
- space_info = get_mcp_space_info(space_id)
259
 
260
- if not space_info:
261
- return f"Space {space_id} not found or not available."
 
262
 
263
- config_text = f"""
264
- # {space_info.title}
265
-
266
- **Space ID:** {space_info.id}
267
- **Author:** {space_info.author}
268
- **Likes:** ❤️ {space_info.likes}
269
- **Status:** {space_info.status}
270
- **URL:** [{space_info.url}]({space_info.url})
271
-
272
- ## 🔧 MCP Configuration
273
-
274
- ### For VSCode/Cursor/Claude Code (Recommended)
275
- ```json
276
- {{
277
- "servers": {{
278
- "{space_info.id.replace('/', '-')}": {{
279
- "url": "{space_info.url}/gradio_api/mcp/sse"
280
- }}
281
- }}
282
- }}
283
- ```
284
-
285
- ### For Claude Desktop
286
- ```json
287
- {{
288
- "mcpServers": {{
289
- "{space_info.id.replace('/', '-')}": {{
290
- "command": "npx",
291
- "args": [
292
- "mcp-remote",
293
- "{space_info.url}/gradio_api/mcp/sse"
294
- ]
295
- }}
296
- }}
297
- }}
298
- ```
299
-
300
- ### Alternative: Use HF MCP Space Server
301
- ```json
302
- {{
303
- "mcpServers": {{
304
- "hf-spaces": {{
305
- "command": "npx",
306
- "args": [
307
- "-y",
308
- "@llmindset/mcp-hfspace",
309
- "{space_info.id}"
310
- ]
311
- }}
312
- }}
313
- }}
314
- ```
315
-
316
- **Description:** {space_info.description}
317
- """.strip()
318
 
319
- return config_text
320
 
321
- # Utility functions for easy integration
322
- def create_mcp_interface_simple() -> Tuple[gr.Dropdown, gr.Markdown]:
323
- """
324
- Create a simple MCP interface with dropdown and output.
325
-
326
- Returns:
327
- Tuple[gr.Dropdown, gr.Markdown]: (dropdown, output_component)
328
- """
329
- dropdown = create_mcp_dropdown()
330
- output = gr.Markdown("Select an MCP server above to view configuration details.")
331
-
332
- def update_output(space_id):
333
- if space_id:
334
- return get_mcp_config_text(space_id)
335
- return "Select an MCP server above to view configuration details."
336
-
337
- dropdown.change(fn=update_output, inputs=[dropdown], outputs=[output])
338
-
339
- return dropdown, output
 
1
  """
2
+ Simple HuggingFace MCP Spaces Finder Module
3
+ A minimal module to discover MCP servers on HuggingFace Spaces.
4
+ Just a dropdown that displays the selected value in a textbox.
 
 
5
  Usage:
6
+ from mcp_spaces_finder import create_simple_mcp_selector
7
 
8
+ # One-liner in your Gradio app
9
+ dropdown, textbox = create_simple_mcp_selector()
10
  """
11
 
12
  import gradio as gr
13
+ from huggingface_hub import list_spaces
14
  import time
15
+ from typing import List, Tuple
 
16
  from concurrent.futures import ThreadPoolExecutor, as_completed
17
 
18
+ class SimpleMCPFinder:
19
+ """Simple MCP spaces finder with caching."""
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
20
 
21
  def __init__(self, cache_duration: int = 300):
22
+ self.cache = None
23
+ self.cache_time = None
 
 
 
 
 
 
 
24
  self.cache_duration = cache_duration
 
 
 
 
 
 
 
 
 
 
25
 
26
+ def get_mcp_spaces(self) -> List[str]:
27
+ """Get list of running MCP space IDs."""
 
28
 
29
+ # Check cache
30
+ if (self.cache is not None and
31
+ self.cache_time is not None and
32
+ time.time() - self.cache_time < self.cache_duration):
33
+ return self.cache
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
34
 
35
+ print("Fetching MCP spaces...")
 
 
 
 
36
 
37
+ # Get MCP spaces sorted by likes
 
 
38
  spaces = list(list_spaces(
39
  filter="mcp-server",
40
+ sort="likes",
41
  direction=-1,
42
+ limit=1000,
43
  full=True
44
  ))
45
 
46
+ # Extract just the space IDs
47
+ space_ids = [space.id for space in spaces]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
48
 
49
+ # Cache results
50
+ self.cache = space_ids
51
+ self.cache_time = time.time()
52
 
53
+ print(f"Found {len(space_ids)} MCP spaces")
54
+ return space_ids
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
55
 
56
+ # Global instance
57
+ _finder = SimpleMCPFinder()
58
 
59
+ def create_simple_mcp_selector(
60
+ dropdown_label: str = "🤖 Select MCP Server",
61
+ textbox_label: str = "Selected MCP Server",
62
+ placeholder: str = "No server selected"
63
+ ) -> Tuple[gr.Dropdown, gr.Textbox]:
 
64
  """
65
+ Create a simple MCP selector with dropdown and textbox.
66
 
67
  Args:
68
+ dropdown_label (str): Label for the dropdown
69
+ textbox_label (str): Label for the textbox
70
+ placeholder (str): Placeholder text when nothing selected
 
71
 
72
  Returns:
73
+ Tuple[gr.Dropdown, gr.Textbox]: The dropdown and textbox components
74
  """
 
75
 
76
+ # Get MCP spaces
77
+ spaces = _finder.get_mcp_spaces()
78
 
79
+ # Create dropdown with space choices
80
+ dropdown = gr.Dropdown(
81
+ choices=spaces,
82
+ label=f"{dropdown_label} ({len(spaces)} available)",
83
+ value=None
 
84
  )
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
85
 
86
+ # Create textbox to display selection
87
+ textbox = gr.Textbox(
88
+ label=textbox_label,
89
+ value=placeholder,
90
+ interactive=False
91
+ )
 
92
 
93
+ # Connect dropdown to textbox
94
+ def update_textbox(selected_value):
95
+ return selected_value if selected_value else placeholder
96
 
97
+ dropdown.change(
98
+ fn=update_textbox,
99
+ inputs=[dropdown],
100
+ outputs=[textbox]
101
+ )
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
102
 
103
+ return dropdown, textbox
104
 
105
+ def refresh_mcp_spaces():
106
+ """Clear cache to force refresh."""
107
+ _finder.cache = None
108
+ _finder.cache_time = None