""" Server management utilities for Universal MCP Client """ import asyncio import re import logging import traceback from typing import Tuple from config import MCPServerConfig from mcp_client import UniversalMCPClient logger = logging.getLogger(__name__) class ServerManager: """Manages MCP server connections and status""" def __init__(self, mcp_client: UniversalMCPClient): self.mcp_client = mcp_client def convert_hf_space_to_url(self, space_name: str) -> str: """ Convert HuggingFace space name to proper URL format. HuggingFace URL rules: - Replace "/" with "-" - Convert to lowercase - Replace dots and other special chars with "-" - Remove consecutive hyphens """ if "/" not in space_name: raise ValueError("Space name should be in format: username/space-name") # Replace "/" with "-" url_name = space_name.replace("/", "-") # Convert to lowercase url_name = url_name.lower() # Replace dots and other special characters with hyphens url_name = re.sub(r'[^a-z0-9\-]', '-', url_name) # Remove consecutive hyphens url_name = re.sub(r'-+', '-', url_name) # Remove leading/trailing hyphens url_name = url_name.strip('-') return f"https://{url_name}.hf.space" def add_custom_server(self, name: str, space_name: str) -> Tuple[str, str]: """Add a custom MCP server from HuggingFace space name""" logger.info(f"➕ Adding MCP server: {name} from space: {space_name}") if not name or not space_name: return "❌ Please provide both server name and space name", "" space_name = space_name.strip() try: # Use the improved URL conversion mcp_url = self.convert_hf_space_to_url(space_name) logger.info(f"🔗 Converted {space_name} → {mcp_url}") except ValueError as e: return f"❌ {str(e)}", "" config = MCPServerConfig( name=name.strip(), url=mcp_url, description=f"MCP server from HuggingFace space: {space_name}", space_id=space_name ) try: # Run async function def run_async(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: return loop.run_until_complete(self.mcp_client.add_server_async(config)) finally: loop.close() success, message = run_async() logger.info(f"Server addition result: {success} - {message}") if success: # Format success message for accordion display tools_info = "" if 'Found' in message and 'tools:' in message: tools_section = message.split('Found')[1] tools_info = f"**Available Tools:**\n{tools_section}" details_html = f"""
✅ {name} - Connection Details

Space: {space_name}

Base URL: {mcp_url}

Status: Connected successfully!

{tools_info.replace('**', '').replace('**', '').replace(chr(10), '
')}
""" return "✅ Server added successfully!", details_html else: error_html = f"""
❌ {name} - Connection Failed

{message}

""" return f"❌ Failed to add server: {name}", error_html except Exception as e: error_msg = f"❌ Failed to add server: {str(e)}" logger.error(error_msg) logger.error(traceback.format_exc()) return error_msg, "" def get_server_status(self) -> Tuple[str, str]: """Get status of all servers in accordion format""" try: status = self.mcp_client.get_server_status() server_count = f"**Total MCP Servers**: {len(status)}" if not status: return server_count, "

No MCP servers configured yet.

" accordion_html = "" for name, state in status.items(): server_config = self.mcp_client.servers[name] base_url = server_config.url.replace("/gradio_api/mcp/sse", "") # Determine health status health = "🟢 Healthy" if "✅ Connected" in state else "🔴 Unhealthy" accordion_html += f"""
🔧 {name}

Title: {name}

Status: Connected (MCP Protocol)

Health: {health}

Base URL: {base_url}

""" return server_count, accordion_html except Exception as e: return "**Total MCP Servers**: 0", f"

❌ Error getting status: {str(e)}

"