server_manager.py

"""
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"""
                <details style="margin-top: 10px;">
                    <summary style="cursor: pointer; padding: 8px; background: #f0f0f0; border-radius: 4px;"><strong>✅ {name} - Connection Details</strong></summary>
                    <div style="padding: 10px; border-left: 3px solid #28a745; margin-left: 10px; margin-top: 5px;">
                        <p><strong>Space:</strong> {space_name}</p>
                        <p><strong>Base URL:</strong> {mcp_url}</p>
                        <p><strong>Status:</strong> Connected successfully!</p>
                        <div style="margin-top: 10px;">
                            {tools_info.replace('**', '<strong>').replace('**', '</strong>').replace(chr(10), '<br>')}
                        </div>
                    </div>
                </details>
                """
                return "✅ Server added successfully!", details_html
            else:
                error_html = f"""
                <details style="margin-top: 10px;">
                    <summary style="cursor: pointer; padding: 8px; background: #f8d7da; border-radius: 4px;"><strong>❌ {name} - Connection Failed</strong></summary>
                    <div style="padding: 10px; border-left: 3px solid #dc3545; margin-left: 10px; margin-top: 5px;">
                        <p>{message}</p>
                    </div>
                </details>
                """
                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, "<p><em>No MCP servers configured yet.</em></p>"
            
            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"""
                <details style="margin-bottom: 10px;">
                    <summary style="cursor: pointer; padding: 8px; background: #e9ecef; border-radius: 4px;"><strong>🔧 {name}</strong></summary>
                    <div style="padding: 10px; border-left: 3px solid #007bff; margin-left: 10px; margin-top: 5px;">
                        <p><strong>Title:</strong> {name}</p>
                        <p><strong>Status:</strong> Connected (MCP Protocol)</p>
                        <p><strong>Health:</strong> {health}</p>
                        <p><strong>Base URL:</strong> {base_url}</p>
                    </div>
                </details>
                """
            
            return server_count, accordion_html
            
        except Exception as e:
            return "**Total MCP Servers**: 0", f"<p style='color: red;'>❌ Error getting status: {str(e)}</p>"