Update app.py

app.py

@@ -1,219 +1,86 @@
 import gradio as gr
 import requests
-from bs4 import BeautifulSoup
 import json
-from urllib.parse import quote_plus
-import time
+from smolagents import CodeAgent
 import logging
-import re
-from smolagents.tools import tool
-from fastapi import FastAPI
-import uvicorn
+from tenacity import retry, stop_after_attempt, wait_exponential
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-app = FastAPI()
+HF_SPACE_URL = "https://manavraj-troubleshoot-mcp.hf.space"
 
-@tool
-def search_knowledge_base(issue: str) -> str:
-    """
-    Search the knowledge base for solutions related to the provided issue.
-    Args:
-        issue (str): The technical issue to search for
-    Returns:
-        str: Solution steps or message if no solution found
-    """
+@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
+def call_mcp_server(message: str, tool_type: str = "knowledge_base") -> str:
+    """Call MCP server endpoint"""
     try:
-        issue = issue.lower()
-        if "wifi" in issue:
-            return "1. Check if your router is powered on\n2. Restart your router\n3. Try connecting again\n4. If problem persists, contact your ISP"
-        elif "screen" in issue:
-            return "1. Adjust display cable connections\n2. Update graphics drivers\n3. Restart the system\n4. Try a different monitor if available"
-        elif "sound" in issue or "audio" in issue:
-            return "1. Check volume settings\n2. Verify audio output device selection\n3. Update audio drivers\n4. Test with headphones"
-        return "No predefined solution found in the knowledge base. Please try our web search tool for more information."
-    except Exception as e:
-        logger.error(f"Error in knowledge base search: {str(e)}")
-        return f"Error searching knowledge base: {str(e)}"
-
-def search_web_duckduckgo(query: str) -> str:
-    """
-    Perform actual web search using DuckDuckGo API.
-    Args:
-        query (str): The search query
-    Returns:
-        str: Formatted search results
-    """
-    try:
-        url = f"https://api.duckduckgo.com/?q={quote_plus(query)}&format=json&no_html=1&skip_disambig=1"
-        
-        response = requests.get(url, timeout=10)
+        endpoint = f"{HF_SPACE_URL}/gradio_api/mcp/sse"
+        response = requests.post(
+            endpoint,
+            json={
+                "tool": tool_type,
+                "input": message
+            },
+            stream=True,
+            timeout=30
+        )
         response.raise_for_status()
         
-        data = response.json()
-        results = []
-        
-        if data.get('AbstractText'):
-            results.append(f"Summary: {data['AbstractText']}")
-        
-        if data.get('RelatedTopics'):
-            results.append("\nRelated Topics:")
-            for i, topic in enumerate(data['RelatedTopics'][:3], 1):
-                if isinstance(topic, dict) and topic.get('Text'):
-                    results.append(f"{i}. {topic['Text']}")
-                elif hasattr(topic, 'Topics'):
-                    for subtopic in topic['Topics'][:3]:
-                        if subtopic.get('Text'):
-                            results.append(f"{i}. {subtopic['Text']}")
-        
-        if data.get('Definition'):
-            results.append(f"\nDefinition: {data['Definition']}")
-        
-        if not results or len(''.join(results).strip()) < 20:
-            return search_web_scraper(query)
-        return "\n".join(results)
+        for line in response.iter_lines():
+            if line.startswith(b'data:'):
+                return json.loads(line[5:])["output"]
+                
+        return "No response received from MCP server"
     except Exception as e:
-        logger.error(f"Web search API error: {str(e)}")
-        return search_web_scraper(query)
+        logger.error(f"API call failed: {str(e)}")
+        return f"Error: {str(e)}"
 
-def search_web_scraper(query: str) -> str:
-    """
-    Alternative web search using web scraping (backup method).
-    Args:
-        query (str): The search query
-    Returns:
-        str: Formatted search results
-    """
-    try:
-        search_url = f"https://duckduckgo.com/html/?q={quote_plus(query)}"
-        
-        headers = {
-            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
-        }
-        
-        response = requests.get(search_url, headers=headers, timeout=10)
-        response.raise_for_status()
-        
-        soup = BeautifulSoup(response.content, 'html.parser')
-        results = []
-        result_links = soup.find_all('a', class_='result__a')[:5]
-        
-        for i, link in enumerate(result_links, 1):
-            title = link.get_text(strip=True)
-            if title:
-                results.append(f"{i}. {title} (https://{link['href']})" if link.get('href') else f"{i}. {title}")
-        
-        if results:
-            return f"Search Results for '{query}':\n" + "\n".join(results)
-        return f"No results found for '{query}'. Try different keywords."
-    except Exception as e:
-        logger.error(f"Web scraping error: {str(e)}")
-        return f"Error in backup search: {str(e)}"
-
-@tool
-def search_web(query: str) -> str:
-    """
-    Main web search function that tries multiple methods.
-    Args:
-        query (str): The search query
-    Returns:
-        str: Formatted search results
-    """
-    try:
-        if not query.strip():
-            return "Please enter a search query."
-        return search_web_duckduckgo(query)
-    except Exception as e:
-        logger.error(f"Error in main web search: {str(e)}")
-        return f"Error processing your search: {str(e)}"
+agent = CodeAgent(
+    tools=[],
+    model="microsoft/DialoGPT-medium",
+    system_prompt="""{{authorized_imports}}
+- requests for API calls
+- standard Python libraries
+{{managed_agents_descriptions}}
+You are a Technical Support Assistant with these capabilities:
+1. Troubleshooting technical issues
+2. Finding information via web search
+3. Formatting instructions
+Access tools through MCP server:
+- knowledge_base: For technical issues
+- web_search: For information lookup
+- formatter: To organize steps
+Response workflow:
+1. Analyze user request
+2. Choose appropriate tool
+3. Return clear response
+Example:
+USER: My wifi disconnected
+THOUGHT: Should use knowledge_base
+ACTION: knowledge_base("wifi disconnection")
+RESPONSE: Try these steps: [solution steps]
+Never show internal workflow to user."""
+)
 
-@tool
-def format_response(raw_steps: str) -> str:
-    """
-    Format the raw steps into a numbered list.
-    Args:
-        raw_steps (str): Raw text steps separated by periods
-    Returns:
-        str: Formatted numbered list of steps
-    """
+def chat_interface(message: str, history: list) -> str:
+    """Handle chat interaction"""
     try:
-        if not raw_steps.strip():
-            return "Please enter some text to format."
-        
-        steps = re.split(r'[.,;]\s*', raw_steps)
-        steps = [step.strip() for step in steps if step.strip()]
-        steps = [f"{i+1}. {step}" for i, step in enumerate(steps)]
-        return "\n".join(steps)
+        response = agent.run(message)
+        if "ACTION:" in response:
+            tool_call = response.split("ACTION:")[1].split("(")[0].strip()
+            input_text = response.split("(")[1].split(")")[0].strip('"')
+            return call_mcp_server(input_text, tool_call)
+        return response
     except Exception as e:
-        logger.error(f"Error formatting steps: {str(e)}")
-        return f"Error formatting your steps: {str(e)}"
-
-# Create interfaces
-demo1 = gr.Interface(
-    fn=search_knowledge_base,
-    inputs=[gr.Textbox(label="Technical Issue", placeholder="Enter your technical problem")],
-    outputs=[gr.Textbox(label="Solution")],
-    title="Knowledge Base Search"
-)
+        logger.error(f"Chat error: {str(e)}")
+        return f"Error: {str(e)}"
 
-demo2 = gr.Interface(
-    fn=search_web,
-    inputs=[gr.Textbox(label="Search Query", placeholder="Enter your search query")],
-    outputs=[gr.Textbox(label="Search Results")],
-    title="Web Search"
+demo = gr.ChatInterface(
+    fn=chat_interface,
+    title="🔧 Technical Support",
+    examples=["Wifi not working", "Find Windows 11 specs", "Format: Turn off. Wait. Restart"]
 )
 
-demo3 = gr.Interface(
-    fn=format_response,
-    inputs=[gr.Textbox(label="Raw Steps", placeholder="Enter steps separated by periods")],
-    outputs=[gr.Textbox(label="Formatted Steps")],
-    title="Response Formatter"
-)
-
-demo = gr.TabbedInterface([demo1, demo2, demo3], ["Knowledge Base", "Web Search", "Formatter"])
-
-# Add MCP schema endpoint
-@app.get("/gradio_api/mcp/schema")
-async def mcp_schema():
-    return {
-        "tools": [
-            {
-                "name": "search_knowledge_base",
-                "description": "Search knowledge base for technical solutions",
-                "parameters": {
-                    "type": "object",
-                    "properties": {
-                        "issue": {"type": "string"}
-                    }
-                }
-            },
-            {
-                "name": "search_web",
-                "description": "Perform web search",
-                "parameters": {
-                    "type": "object",
-                    "properties": {
-                        "query": {"type": "string"}
-                    }
-                }
-            },
-            {
-                "name": "format_response",
-                "description": "Format steps into numbered list",
-                "parameters": {
-                    "type": "object",
-                    "properties": {
-                        "raw_steps": {"type": "string"}
-                    }
-                }
-            }
-        ]
-    }
-
-# Mount Gradio app
-app = gr.mount_gradio_app(app, demo, path="/")
-
 if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+    demo.launch()