Delete fixed_mcp_gradio_app.py

fixed_mcp_gradio_app.py

@@ -1,703 +0,0 @@
-#!/usr/bin/env python3
-"""
-🏆 Fixed TopCoder MCP Agent - Enhanced Gradio Interface
-Implements proper MCP integration with comprehensive error handling and debugging
-"""
-
-import asyncio
-import gradio as gr
-import json
-import logging
-from datetime import datetime
-import threading
-import time
-
-# Import our fixed MCP client
-from fixed_mcp_client import FixedTopCoderMCPClient
-
-# Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-class EnhancedMCPAgent:
-    """
-    Enhanced MCP Agent with comprehensive debugging and error handling
-    """
-    
-    def __init__(self):
-        self.mcp_client = None
-        self.is_connected = False
-        self.connection_status = "Not Connected"
-        self.transport_method = "HTTP"  # Start with HTTP
-        self.connection_log = []
-        self.stats = {
-            "connection_attempts": 0,
-            "successful_calls": 0,
-            "failed_calls": 0,
-            "total_queries": 0,
-            "endpoint_discovered": None,
-            "last_error": None
-        }
-    
-    def log_message(self, message):
-        """Add timestamped message to connection log"""
-        timestamp = datetime.now().strftime('%H:%M:%S')
-        log_entry = f"[{timestamp}] {message}"
-        self.connection_log.append(log_entry)
-        logger.info(message)
-        
-        # Keep only last 50 entries
-        if len(self.connection_log) > 50:
-            self.connection_log = self.connection_log[-50:]
-        
-        return "\n".join(self.connection_log)
-    
-    async def initialize_mcp_connection(self):
-        """Initialize MCP connection with enhanced error handling"""
-        try:
-            self.log_message(f"🔄 Starting MCP connection via {self.transport_method}")
-            self.stats["connection_attempts"] += 1
-            
-            # Create new client
-            self.mcp_client = FixedTopCoderMCPClient()
-            self.mcp_client.transport_method = self.transport_method
-            
-            # Initialize with endpoint discovery
-            success = await self.mcp_client.initialize()
-            
-            if success:
-                self.is_connected = True
-                self.connection_status = f"Connected via {self.transport_method}"
-                self.stats["endpoint_discovered"] = self.mcp_client.working_endpoint
-                self.log_message(f"✅ MCP connected! Using endpoint: {self.mcp_client.working_endpoint}")
-                return True
-            else:
-                self.is_connected = False
-                self.connection_status = "Connection Failed"
-                self.stats["last_error"] = "Initialize failed"
-                self.log_message("❌ MCP connection failed")
-                return False
-                
-        except Exception as e:
-            self.is_connected = False
-            self.connection_status = f"Error: {str(e)}"
-            self.stats["last_error"] = str(e)
-            self.log_message(f"❌ Connection error: {e}")
-            return False
-    
-    def sync_initialize_mcp(self):
-        """Synchronous wrapper for MCP initialization"""
-        try:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            result = loop.run_until_complete(self.initialize_mcp_connection())
-            return result
-        except Exception as e:
-            self.log_message(f"❌ Sync initialize error: {e}")
-            return False
-        finally:
-            try:
-                loop.close()
-            except:
-                pass
-    
-    async def test_connection_async(self):
-        """Test MCP connection"""
-        try:
-            if not self.mcp_client:
-                return {"success": False, "error": "No MCP client"}
-            
-            result = await self.mcp_client.test_connection()
-            
-            if result.get("success"):
-                self.log_message(f"✅ Connection test passed")
-            else:
-                self.log_message(f"❌ Connection test failed: {result.get('error')}")
-            
-            return result
-            
-        except Exception as e:
-            self.log_message(f"❌ Test error: {e}")
-            return {"success": False, "error": str(e)}
-    
-    def sync_test_connection(self):
-        """Synchronous wrapper for connection test"""
-        try:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            result = loop.run_until_complete(self.test_connection_async())
-            return result
-        except Exception as e:
-            return {"success": False, "error": str(e)}
-        finally:
-            try:
-                loop.close()
-            except:
-                pass
-    
-    async def query_challenges_async(self, **params):
-        """Query challenges via MCP"""
-        try:
-            if not self.is_connected or not self.mcp_client:
-                return {
-                    "success": False,
-                    "error": "Not connected to MCP server",
-                    "data": None
-                }
-            
-            self.stats["total_queries"] += 1
-            self.log_message(f"🔍 Querying challenges with params: {params}")
-            
-            result = await self.mcp_client.query_challenges(**params)
-            
-            if result.get("success"):
-                self.stats["successful_calls"] += 1
-                self.log_message("✅ Query successful")
-            else:
-                self.stats["failed_calls"] += 1
-                self.log_message(f"❌ Query failed: {result.get('error')}")
-            
-            return result
-            
-        except Exception as e:
-            self.stats["failed_calls"] += 1
-            self.log_message(f"❌ Query error: {e}")
-            return {"success": False, "error": str(e)}
-    
-    def sync_query_challenges(self, **params):
-        """Synchronous wrapper for challenge queries"""
-        try:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            result = loop.run_until_complete(self.query_challenges_async(**params))
-            return result
-        except Exception as e:
-            return {"success": False, "error": str(e)}
-        finally:
-            try:
-                loop.close()
-            except:
-                pass
-    
-    def switch_transport_method(self, method: str):
-        """Switch transport method"""
-        try:
-            if method.upper() in ["SSE", "HTTP"]:
-                self.transport_method = method.upper()
-                
-                if self.mcp_client:
-                    self.mcp_client.switch_transport(method)
-                
-                self.log_message(f"🔄 Switched to {self.transport_method} transport")
-                return True
-            else:
-                self.log_message(f"⚠️ Invalid transport: {method}")
-                return False
-        except Exception as e:
-            self.log_message(f"❌ Transport switch error: {e}")
-            return False
-    
-    def get_connection_status(self):
-        """Get comprehensive connection status"""
-        return {
-            "status": self.connection_status,
-            "is_connected": self.is_connected,
-            "transport": self.transport_method,
-            "working_endpoint": getattr(self.mcp_client, 'working_endpoint', None) if self.mcp_client else None,
-            "stats": self.stats.copy(),
-            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-            "tools_available": len(getattr(self.mcp_client, 'available_tools', []))
-        }
-
-def create_fixed_mcp_app():
-    """
-    Create enhanced Gradio app with comprehensive debugging
-    """
-    
-    # Create agent instance
-    agent = EnhancedMCPAgent()
-    
-    # Enhanced CSS
-    css = """
-    .status-connected { 
-        background: linear-gradient(45deg, #28a745, #20c997) !important;
-        color: white !important; padding: 15px !important; border-radius: 8px !important; 
-        margin: 10px 0 !important; font-weight: bold !important;
-    }
-    .status-error {
-        background: linear-gradient(45deg, #dc3545, #c82333) !important;
-        color: white !important; padding: 15px !important; border-radius: 8px !important;
-        margin: 10px 0 !important; font-weight: bold !important;
-    }
-    .status-testing {
-        background: linear-gradient(45deg, #ffc107, #ff8f00) !important;
-        color: white !important; padding: 15px !important; border-radius: 8px !important;
-        margin: 10px 0 !important; font-weight: bold !important;
-    }
-    .app-header {
-        background: linear-gradient(135deg, #667eea 0%, #764ba2 100%) !important;
-        color: white !important; padding: 30px !important; border-radius: 15px !important;
-        text-align: center !important; margin-bottom: 25px !important;
-    }
-    .transport-info {
-        background: linear-gradient(45deg, #17a2b8, #138496) !important;
-        color: white !important; padding: 12px !important; border-radius: 6px !important;
-        margin: 8px 0 !important; font-size: 14px !important;
-    }
-    .debug-section {
-        background: #f8f9fa !important; border: 1px solid #dee2e6 !important;
-        border-radius: 8px !important; padding: 15px !important; margin: 10px 0 !important;
-    }
-    """
-    
-    with gr.Blocks(css=css, title="🔧 Fixed TopCoder MCP Agent") as interface:
-        
-        # Enhanced Header
-        gr.HTML("""
-        <div class="app-header">
-            <h1>🔧 Fixed TopCoder MCP Agent</h1>
-            <h2>Enhanced Debug Version with Endpoint Discovery</h2>
-            <p><strong>✅ Tests multiple endpoints automatically</strong></p>
-            <p><strong>✅ Enhanced error handling and diagnostics</strong></p>
-            <p><strong>✅ Real-time connection status monitoring</strong></p>
-            <p style="margin-top: 15px; font-size: 14px;">
-                <strong>Tested Endpoints:</strong><br>
-                https://api.topcoder-dev.com/v6/mcp<br>
-                https://api.topcoder-dev.com/v6/mcp/sse<br>
-                https://api.topcoder-dev.com/v6/mcp/mcp<br>
-                + Production endpoints
-            </p>
-        </div>
-        """)
-        
-        # Connection Control Section
-        with gr.Row():
-            with gr.Column(scale=1):
-                gr.Markdown("### 🚀 Connection Control")
-                
-                transport_choice = gr.Radio(
-                    choices=["HTTP", "SSE"],
-                    value="HTTP",
-                    label="Transport Method",
-                    info="HTTP recommended for initial testing"
-                )
-                
-                with gr.Row():
-                    connect_btn = gr.Button("🔌 Connect & Discover", variant="primary", size="lg")
-                    test_btn = gr.Button("🧪 Test Connection", variant="secondary")
-                    refresh_btn = gr.Button("🔄 Refresh Status", variant="secondary")
-                
-            with gr.Column(scale=1):
-                connection_display = gr.HTML(
-                    value='<div class="status-error">🔴 Not Connected - Click Connect & Discover</div>',
-                    label="Connection Status"
-                )
-                
-                transport_info = gr.HTML(
-                    value='<div class="transport-info">📡 Transport: HTTP<br>🔍 Endpoint Discovery: Ready</div>',
-                    label="Transport & Endpoint Info"
-                )
-        
-        # Enhanced Statistics and Diagnostics
-        with gr.Row():
-            with gr.Column():
-                connection_stats = gr.JSON(
-                    label="📊 Connection Statistics",
-                    value={
-                        "status": "Ready",
-                        "transport": "HTTP",
-                        "endpoint_discovered": None,
-                        "stats": {"connection_attempts": 0, "successful_calls": 0, "failed_calls": 0}
-                    }
-                )
-            
-            with gr.Column():
-                test_results = gr.JSON(
-                    label="🧪 Last Test Results",
-                    value={"ready": "Click Test Connection to see results"}
-                )
-        
-        # Main Interface Tabs
-        with gr.Tabs():
-            # Challenge Query Tab
-            with gr.TabItem("📋 MCP Challenge Query"):
-                with gr.Row():
-                    with gr.Column(scale=1):
-                        gr.Markdown("### 🔍 Query Parameters")
-                        gr.Markdown("*Enhanced with multiple fallback methods*")
-                        
-                        status_filter = gr.Dropdown(
-                            choices=["Any", "Active", "Completed", "Draft"],
-                            value="Active",
-                            label="Challenge Status",
-                            info="Filter by challenge status"
-                        )
-                        
-                        track_filter = gr.Dropdown(
-                            choices=["Any", "DEVELOP", "DESIGN", "DATA_SCIENCE"],
-                            value="DEVELOP",
-                            label="Track",
-                            info="Filter by challenge track"
-                        )
-                        
-                        subtrack_filter = gr.Dropdown(
-                            choices=["Any", "Code", "Algorithm", "Marathon", "UI Prototype", "Bug Hunt"],
-                            value="Any",
-                            label="Sub Track",
-                            info="Optional sub-track filter"
-                        )
-                        
-                        with gr.Row():
-                            per_page = gr.Number(
-                                value=5,
-                                minimum=1,
-                                maximum=20,
-                                label="Results per Page",
-                                info="Number of results to return"
-                            )
-                            page = gr.Number(
-                                value=1,
-                                minimum=1,
-                                label="Page Number",
-                                info="Page number for pagination"
-                            )
-                        
-                        sort_order = gr.Dropdown(
-                            choices=["Any", "asc", "desc"],
-                            value="desc",
-                            label="Sort Order",
-                            info="Sort results ascending or descending"
-                        )
-                        
-                        query_btn = gr.Button("🔍 Query MCP Server", variant="primary", size="lg")
-                        
-                    with gr.Column(scale=2):
-                        gr.Markdown("### 📊 Query Results")
-                        
-                        query_status = gr.HTML(
-                            value='<div class="debug-section">Ready to query MCP server - Connect first</div>',
-                            label="Query Status"
-                        )
-                        
-                        raw_results = gr.JSON(
-                            label="Raw MCP Response",
-                            value={},
-                            info="Complete response from MCP server"
-                        )
-                        
-                        formatted_results = gr.Textbox(
-                            label="Formatted Results",
-                            lines=15,
-                            interactive=False,
-                            placeholder="Query results will appear here...",
-                            info="Human-readable format of the results"
-                        )
-            
-            # Challenge Details Tab
-            with gr.TabItem("📄 Challenge Details"):
-                with gr.Row():
-                    with gr.Column(scale=1):
-                        gr.Markdown("### 📝 Get Challenge Details")
-                        gr.Markdown("*Fetch detailed information for specific challenges*")
-                        
-                        challenge_id_input = gr.Textbox(
-                            label="Challenge ID",
-                            placeholder="Enter challenge ID (e.g., 30377453)",
-                            value="",
-                            info="Get this from the challenge query results"
-                        )
-                        
-                        details_btn = gr.Button("📄 Get Details", variant="primary", size="lg")
-                        
-                    with gr.Column(scale=2):
-                        gr.Markdown("### 📋 Challenge Details")
-                        
-                        details_status = gr.HTML(
-                            value='<div class="debug-section">Enter challenge ID and click Get Details</div>'
-                        )
-                        
-                        details_raw = gr.JSON(
-                            label="Raw Details Response",
-                            value={}
-                        )
-                        
-                        details_formatted = gr.Textbox(
-                            label="Formatted Details",
-                            lines=15,
-                            interactive=False,
-                            placeholder="Challenge details will appear here..."
-                        )
-            
-            # Debug & Diagnostics Tab
-            with gr.TabItem("🔧 Debug & Diagnostics"):
-                with gr.Row():
-                    with gr.Column():
-                        gr.Markdown("### 🏥 Connection Health")
-                        
-                        health_display = gr.JSON(
-                            label="Connection Health Status",
-                            value={"status": "Not tested"}
-                        )
-                        
-                        endpoint_discovery = gr.JSON(
-                            label="Endpoint Discovery Results",
-                            value={
-                                "tested_endpoints": [],
-                                "working_endpoints": [],
-                                "failed_endpoints": []
-                            }
-                        )
-                        
-                    with gr.Column():
-                        gr.Markdown("### 📝 Live Connection Log")
-                        
-                        connection_log = gr.Textbox(
-                            label="Connection Log (Live Updates)",
-                            lines=20,
-                            interactive=False,
-                            value="Enhanced MCP Agent ready\n✅ Multiple endpoint testing\n✅ Comprehensive error handling\n✅ Real-time diagnostics\n\nSelect transport method and click Connect & Discover to start",
-                            info="Real-time log of all MCP operations"
-                        )
-        
-        # Event Handlers
-        def handle_transport_change(transport):
-            """Handle transport method change with logging"""
-            try:
-                agent.switch_transport_method(transport)
-                
-                transport_html = f'<div class="transport-info">📡 Transport: {transport}<br>🔍 Ready for endpoint discovery</div>'
-                log = agent.log_message(f"🔄 Transport changed to {transport}")
-                
-                return transport_html, log
-                
-            except Exception as e:
-                error_html = f'<div class="status-error">❌ Transport change failed: {str(e)}</div>'
-                log = agent.log_message(f"❌ Transport change error: {e}")
-                return error_html, log
-        
-        def handle_connect():
-            """Handle MCP server connection with endpoint discovery"""
-            try:
-                # Show connecting status
-                connecting_html = '<div class="status-testing">🔄 Connecting & Discovering Endpoints...</div>'
-                
-                # Perform connection
-                success = agent.sync_initialize_mcp()
-                status = agent.get_connection_status()
-                
-                if success:
-                    endpoint = status.get("working_endpoint", "Unknown")
-                    status_html = f'<div class="status-connected">✅ MCP Connected Successfully<br>🌐 Transport: {status["transport"]}<br>🔗 Endpoint: {endpoint}<br><small>Connected at {status["timestamp"]}</small></div>'
-                    
-                    transport_html = f'<div class="transport-info">📡 Transport: {status["transport"]}<br>🔗 Working Endpoint: {endpoint}<br>🛠️ Tools Available: {status["tools_available"]}</div>'
-                else:
-                    status_html = f'<div class="status-error">❌ Connection Failed<br><small>{status["status"]}</small><br><small>Last Error: {status["stats"].get("last_error", "Unknown")}</small></div>'
-                    
-                    transport_html = f'<div class="status-error">❌ No working endpoint found<br>📡 Tried: {status["transport"]} transport</div>'
-                
-                log = "\n".join(agent.connection_log[-30:])
-                
-                return (
-                    status_html,
-                    transport_html,
-                    status,
-                    log
-                )
-                
-            except Exception as e:
-                error_html = f'<div class="status-error">❌ Connection Error<br><small>{str(e)}</small></div>'
-                error_status = {"status": f"Error: {str(e)}", "is_connected": False}
-                log = agent.log_message(f"❌ Connect error: {e}")
-                return (error_html, error_html, error_status, log)
-        
-        def handle_test_connection():
-            """Handle connection test with detailed results"""
-            try:
-                result = agent.sync_test_connection()
-                log = "\n".join(agent.connection_log[-20:])
-                
-                # Update health display based on test results
-                if result.get("success"):
-                    health_status = {
-                        "connection_healthy": True,
-                        "transport": result.get("transport"),
-                        "endpoint": result.get("endpoint"),
-                        "tools_available": result.get("tools_available", 0),
-                        "last_test": datetime.now().strftime('%H:%M:%S')
-                    }
-                else:
-                    health_status = {
-                        "connection_healthy": False,
-                        "error": result.get("error"),
-                        "last_test": datetime.now().strftime('%H:%M:%S')
-                    }
-                
-                return result, health_status, log
-                
-            except Exception as e:
-                error_result = {"success": False, "error": str(e)}
-                health_status = {"connection_healthy": False, "error": str(e)}
-                log = agent.log_message(f"❌ Test error: {e}")
-                return error_result, health_status, log
-        
-        def handle_refresh_status():
-            """Refresh connection status display"""
-            try:
-                status = agent.get_connection_status()
-                log = "\n".join(agent.connection_log[-20:])
-                
-                return status, log
-                
-            except Exception as e:
-                error_status = {"error": f"Failed to refresh: {e}"}
-                log = agent.log_message(f"❌ Refresh error: {e}")
-                return error_status, log
-        
-        def handle_query_challenges(status, track, subtrack, per_page, page, sort_order):
-            """Handle MCP challenge queries with enhanced error handling"""
-            try:
-                if not agent.is_connected:
-                    error_html = '<div class="status-error">❌ Not Connected - Please connect first</div>'
-                    return (
-                        error_html,
-                        {"error": "Not connected to MCP server"},
-                        "Please connect to the MCP server first"
-                    )
-                
-                # Build query parameters
-                query_params = {}
-                if status and status != "Any":
-                    query_params["status"] = status
-                if track and track != "Any":
-                    query_params["track"] = track
-                if subtrack and subtrack != "Any":
-                    query_params["subTrack"] = subtrack
-                if per_page:
-                    query_params["perPage"] = int(per_page)
-                if page:
-                    query_params["page"] = int(page)
-                if sort_order and sort_order != "Any":
-                    query_params["sortOrder"] = sort_order
-                
-                # Execute query
-                result = agent.sync_query_challenges(**query_params)
-                timestamp = datetime.now().strftime('%H:%M:%S')
-                
-                if result.get("success"):
-                    status_html = f'<div class="status-connected">✅ Query Successful<br>🌐 Transport: {result.get("transport", "Unknown")}<br><small>Query completed at {timestamp}</small></div>'
-                    
-                    # Format results for display
-                    data = result.get("data", {})
-                    formatted_text = f"MCP Query Results:\n"
-                    formatted_text += f"Transport: {result.get('transport', 'Unknown')}\n"
-                    formatted_text += f"Format: {result.get('format', 'Unknown')}\n"
-                    formatted_text += f"Time: {timestamp}\n\n"
-                    
-                    # Try to extract and format challenge data
-                    if isinstance(data, dict):
-                        if "challenges" in data:
-                            challenges = data["challenges"]
-                            formatted_text += f"Found {len(challenges)} challenges:\n\n"
-                            
-                            for i, challenge in enumerate(challenges[:10], 1):
-                                name = challenge.get("name", "Unknown")
-                                challenge_id = challenge.get("id", "Unknown")
-                                track_name = challenge.get("track", "Unknown")
-                                status_name = challenge.get("status", "Unknown")
-                                prize = challenge.get("totalPrize", 0)
-                                
-                                formatted_text += f"{i}. {name}\n"
-                                formatted_text += f"   ID: {challenge_id}\n"
-                                formatted_text += f"   Track: {track_name}\n"
-                                formatted_text += f"   Status: {status_name}\n"
-                                formatted_text += f"   Prize: ${prize}\n\n"
-                        else:
-                            # Format other types of responses
-                            formatted_text += f"Response type: {type(data).__name__}\n"
-                            formatted_text += f"Keys: {list(data.keys()) if isinstance(data, dict) else 'N/A'}\n\n"
-                            formatted_text += f"Raw data preview:\n{str(data)[:500]}..."
-                    else:
-                        formatted_text += f"Non-dict response: {str(data)[:500]}..."
-                    
-                    return (
-                        status_html,
-                        result,
-                        formatted_text
-                    )
-                else:
-                    error_html = f'<div class="status-error">❌ Query Failed<br><small>{result.get("error", "Unknown error")}</small></div>'
-                    return (
-                        error_html,
-                        result,
-                        f"Query failed: {result.get('error', 'Unknown error')}"
-                    )
-                    
-            except Exception as e:
-                error_html = f'<div class="status-error">❌ Query Error<br><small>{str(e)}</small></div>'
-                return (
-                    error_html,
-                    {"error": str(e)},
-                    f"Query error: {str(e)}"
-                )
-        
-        # Connect all event handlers
-        transport_choice.change(
-            fn=handle_transport_change,
-            inputs=[transport_choice],
-            outputs=[transport_info, connection_log]
-        )
-        
-        connect_btn.click(
-            fn=handle_connect,
-            outputs=[connection_display, transport_info, connection_stats, connection_log]
-        )
-        
-        test_btn.click(
-            fn=handle_test_connection,
-            outputs=[test_results, health_display, connection_log]
-        )
-        
-        refresh_btn.click(
-            fn=handle_refresh_status,
-            outputs=[connection_stats, connection_log]
-        )
-        
-        query_btn.click(
-            fn=handle_query_challenges,
-            inputs=[status_filter, track_filter, subtrack_filter, per_page, page, sort_order],
-            outputs=[query_status, raw_results, formatted_results]
-        )
-        
-        # Initialize interface
-        interface.load(
-            fn=lambda: (
-                '<div class="status-error">🔴 Not Connected - Select transport and connect</div>',
-                {"status": "Ready to connect", "transport": "HTTP", "endpoint_discovered": None},
-                "Enhanced MCP Agent loaded - Ready for connection\nSelect transport method and click 'Connect & Discover'\n"
-            ),
-            outputs=[connection_display, connection_stats, connection_log]
-        )
-    
-    return interface
-
-# Main execution
-def main():
-    """Main function for the enhanced MCP agent"""
-    print("🚀 Starting Enhanced TopCoder MCP Agent")
-    print("🔧 Multiple endpoint testing")
-    print("📊 Comprehensive diagnostics")
-    print("🌐 HTTP & SSE transport support")
-    print("=" * 60)
-    
-    app = create_fixed_mcp_app()
-    
-    # Launch with settings compatible with Hugging Face Spaces
-    app.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        share=False,
-        show_error=True
-    )
-
-if __name__ == "__main__":
-    main()