Create app.py

app.py

@@ -0,0 +1,357 @@
+import gradio as gr
+import openai
+import requests
+import json
+from typing import Dict, List, Any
+import os
+from datetime import datetime
+
+class MCPClient:
+    def __init__(self, openai_api_key: str):
+        """Initialize the MCP Client with OpenAI integration"""
+        self.openai_client = openai.OpenAI(api_key=openai_api_key)
+        
+        # MCP Server configurations
+        self.mcp_servers = {
+            "terraform": {
+                "name": "MCP Terraform",
+                "space_url": "https://chrissacrumcor-mcp-terraform.hf.space",
+                "description": "Terraform infrastructure management"
+            },
+            "linux": {
+                "name": "MCP Linux", 
+                "space_url": "https://chrissacrumcor-mcp-linux.hf.space",
+                "description": "Linux system operations"
+            },
+            "cisco": {
+                "name": "MCP Cisco",
+                "space_url": "https://chrissacrumcor-mcp-cisco.hf.space", 
+                "description": "Cisco network management"
+            }
+        }
+        
+        self.conversation_history = []
+    
+    def call_mcp_server(self, server_key: str, endpoint: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+        """Make a call to an MCP server"""
+        try:
+            server_config = self.mcp_servers[server_key]
+            url = f"{server_config['space_url']}/{endpoint}"
+            
+            response = requests.post(
+                url,
+                json=payload,
+                headers={"Content-Type": "application/json"},
+                timeout=30
+            )
+            
+            if response.status_code == 200:
+                return {"success": True, "data": response.json()}
+            else:
+                return {
+                    "success": False, 
+                    "error": f"HTTP {response.status_code}: {response.text}"
+                }
+                
+        except requests.exceptions.RequestException as e:
+            return {"success": False, "error": f"Request failed: {str(e)}"}
+        except Exception as e:
+            return {"success": False, "error": f"Unexpected error: {str(e)}"}
+    
+    def get_available_tools(self, server_key: str) -> List[Dict[str, Any]]:
+        """Get available tools from an MCP server"""
+        result = self.call_mcp_server(server_key, "tools/list", {})
+        if result["success"]:
+            return result["data"].get("tools", [])
+        return []
+    
+    def execute_tool(self, server_key: str, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        """Execute a tool on an MCP server"""
+        payload = {
+            "name": tool_name,
+            "arguments": arguments
+        }
+        return self.call_mcp_server(server_key, "tools/call", payload)
+    
+    def analyze_user_request(self, user_input: str) -> Dict[str, Any]:
+        """Use OpenAI to analyze user request and determine which MCP server to use"""
+        system_prompt = f"""
+        You are an intelligent assistant that helps users interact with MCP (Model Context Protocol) servers.
+        
+        Available MCP servers:
+        1. Terraform: {self.mcp_servers['terraform']['description']}
+        2. Linux: {self.mcp_servers['linux']['description']}  
+        3. Cisco: {self.mcp_servers['cisco']['description']}
+        
+        Analyze the user's request and determine:
+        1. Which MCP server(s) would be most appropriate
+        2. What specific tools or operations might be needed
+        3. Any parameters or arguments that should be passed
+        
+        Respond in JSON format with:
+        {{
+            "recommended_server": "server_key",
+            "reasoning": "explanation of why this server was chosen",
+            "suggested_action": "what action to take",
+            "parameters": {{"key": "value"}}
+        }}
+        """
+        
+        try:
+            response = self.openai_client.chat.completions.create(
+                model="gpt-4",
+                messages=[
+                    {"role": "system", "content": system_prompt},
+                    {"role": "user", "content": user_input}
+                ],
+                temperature=0.3
+            )
+            
+            return json.loads(response.choices[0].message.content)
+            
+        except Exception as e:
+            return {
+                "error": f"Failed to analyze request: {str(e)}",
+                "recommended_server": None
+            }
+    
+    def generate_response(self, user_input: str, mcp_results: Dict[str, Any] = None) -> str:
+        """Generate a natural language response using OpenAI"""
+        context = ""
+        if mcp_results:
+            context = f"\nMCP Server Results: {json.dumps(mcp_results, indent=2)}"
+        
+        messages = [
+            {
+                "role": "system", 
+                "content": f"""
+                You are a helpful assistant that works with MCP servers for infrastructure management.
+                Provide clear, helpful responses based on the user's request and any MCP server results.
+                
+                Available MCP Servers:
+                - Terraform: Infrastructure as Code management
+                - Linux: System administration and operations  
+                - Cisco: Network device management
+                
+                {context}
+                """
+            }
+        ]
+        
+        # Add conversation history (last 6 messages)
+        messages.extend(self.conversation_history[-6:])
+        messages.append({"role": "user", "content": user_input})
+        
+        try:
+            response = self.openai_client.chat.completions.create(
+                model="gpt-4",
+                messages=messages,
+                temperature=0.7,
+                max_tokens=1000
+            )
+            
+            return response.choices[0].message.content
+            
+        except Exception as e:
+            return f"Error generating response: {str(e)}"
+    
+    def process_request(self, user_input: str, selected_server: str = None) -> tuple[str, str]:
+        """Process a user request end-to-end"""
+        timestamp = datetime.now().strftime("%H:%M:%S")
+        
+        # Add to conversation history
+        self.conversation_history.append({"role": "user", "content": user_input})
+        
+        try:
+            # Analyze request if no server specified
+            if not selected_server or selected_server == "Auto-detect":
+                analysis = self.analyze_user_request(user_input)
+                if "error" in analysis:
+                    response = f"❌ Analysis Error: {analysis['error']}"
+                    self.conversation_history.append({"role": "assistant", "content": response})
+                    return response, f"[{timestamp}] Analysis failed"
+                
+                selected_server = analysis.get("recommended_server", "terraform")
+                reasoning = analysis.get("reasoning", "No reasoning provided")
+                status = f"[{timestamp}] Selected: {self.mcp_servers[selected_server]['name']} - {reasoning}"
+            else:
+                status = f"[{timestamp}] Using: {self.mcp_servers[selected_server]['name']}"
+            
+            # Get available tools for the selected server
+            tools = self.get_available_tools(selected_server)
+            
+            if not tools:
+                response = f"⚠️ No tools available from {self.mcp_servers[selected_server]['name']} server"
+                self.conversation_history.append({"role": "assistant", "content": response})
+                return response, status + " - No tools available"
+            
+            # For now, just show available tools and generate a helpful response
+            # In a full implementation, you'd use OpenAI to select and execute appropriate tools
+            tools_info = "\n".join([f"- {tool.get('name', 'Unknown')}: {tool.get('description', 'No description')}" 
+                                  for tool in tools[:5]])  # Show first 5 tools
+            
+            server_info = {
+                "server": self.mcp_servers[selected_server]['name'],
+                "available_tools": len(tools),
+                "sample_tools": tools_info
+            }
+            
+            response = self.generate_response(user_input, server_info)
+            self.conversation_history.append({"role": "assistant", "content": response})
+            
+            return response, status + f" - {len(tools)} tools available"
+            
+        except Exception as e:
+            error_response = f"❌ Error processing request: {str(e)}"
+            self.conversation_history.append({"role": "assistant", "content": error_response})
+            return error_response, f"[{timestamp}] Error occurred"
+
+def create_gradio_interface():
+    """Create the Gradio interface"""
+    
+    # Initialize MCP Client (will be set when API key is provided)
+    mcp_client = None
+    
+    def initialize_client(api_key):
+        nonlocal mcp_client
+        if not api_key:
+            return "❌ Please provide your OpenAI API key", ""
+        
+        try:
+            mcp_client = MCPClient(api_key)
+            return "✅ OpenAI client initialized successfully!", ""
+        except Exception as e:
+            return f"❌ Failed to initialize: {str(e)}", ""
+    
+    def process_message(message, server_choice, history):
+        if not mcp_client:
+            return history + [["Please initialize the OpenAI client first.", ""]], ""
+        
+        if not message.strip():
+            return history, ""
+        
+        try:
+            response, status = mcp_client.process_request(message, server_choice)
+            history.append([message, response])
+            return history, ""
+        except Exception as e:
+            error_msg = f"❌ Error: {str(e)}"
+            history.append([message, error_msg])
+            return history, ""
+    
+    def get_server_status():
+        status_info = "## MCP Servers Status\n\n"
+        for key, config in mcp_client.mcp_servers.items() if mcp_client else []:
+            try:
+                # Simple health check
+                response = requests.get(f"{config['space_url']}/health", timeout=5)
+                status = "🟢 Online" if response.status_code == 200 else "🟡 Issues"
+            except:
+                status = "🔴 Offline"
+            
+            status_info += f"**{config['name']}**: {status}\n"
+            status_info += f"- URL: {config['space_url']}\n"
+            status_info += f"- Description: {config['description']}\n\n"
+        
+        return status_info
+    
+    # Create Gradio interface
+    with gr.Blocks(title="MCP Client with OpenAI", theme=gr.themes.Soft()) as app:
+        gr.Markdown("# 🤖 MCP Client with OpenAI Integration")
+        gr.Markdown("Connect to your Terraform, Linux, and Cisco MCP servers with AI-powered assistance")
+        
+        with gr.Row():
+            with gr.Column(scale=3):
+                # API Key input
+                api_key = gr.Textbox(
+                    label="OpenAI API Key",
+                    type="password",
+                    placeholder="Enter your OpenAI API key..."
+                )
+                init_btn = gr.Button("Initialize Client", variant="primary")
+                init_status = gr.Textbox(label="Status", interactive=False)
+                
+                # Server selection
+                server_choice = gr.Dropdown(
+                    choices=["Auto-detect", "terraform", "linux", "cisco"],
+                    value="Auto-detect",
+                    label="Select MCP Server"
+                )
+                
+                # Chat interface
+                chatbot = gr.Chatbot(
+                    label="Conversation",
+                    height=400,
+                    show_label=True
+                )
+                
+                msg = gr.Textbox(
+                    label="Message",
+                    placeholder="Ask about infrastructure, Linux systems, or Cisco networks...",
+                    lines=2
+                )
+                
+                with gr.Row():
+                    send_btn = gr.Button("Send", variant="primary")
+                    clear_btn = gr.Button("Clear Chat")
+            
+            with gr.Column(scale=1):
+                gr.Markdown("## 📊 Server Status")
+                server_status = gr.Markdown("Initialize client to see server status")
+                refresh_btn = gr.Button("Refresh Status")
+                
+                gr.Markdown("## 💡 Example Queries")
+                gr.Markdown("""
+                **Terraform:**
+                - "Show me the current infrastructure state"
+                - "Plan a new deployment"
+                
+                **Linux:**
+                - "Check system resources"
+                - "List running processes"
+                
+                **Cisco:**
+                - "Show interface status"
+                - "Check network configuration"
+                """)
+        
+        # Event handlers
+        init_btn.click(
+            initialize_client,
+            inputs=[api_key],
+            outputs=[init_status, server_status]
+        )
+        
+        send_btn.click(
+            process_message,
+            inputs=[msg, server_choice, chatbot],
+            outputs=[chatbot, msg]
+        )
+        
+        msg.submit(
+            process_message,
+            inputs=[msg, server_choice, chatbot],
+            outputs=[chatbot, msg]
+        )
+        
+        clear_btn.click(
+            lambda: [],
+            outputs=[chatbot]
+        )
+        
+        refresh_btn.click(
+            get_server_status,
+            outputs=[server_status]
+        )
+    
+    return app
+
+if __name__ == "__main__":
+    # Create and launch the Gradio app
+    app = create_gradio_interface()
+    app.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=True,
+        debug=True
+    )