gradio_app.py

#!/usr/bin/env python3
"""
Gradio MCP Server for Stack Overflow Search
A web interface and MCP server that provides Stack Overflow search capabilities.
"""

import asyncio
import os
from typing import List, Optional, Tuple
import gradio as gr
from datetime import datetime

from stackoverflow_mcp.api import StackExchangeAPI
from stackoverflow_mcp.formatter import format_response
from stackoverflow_mcp.env import STACK_EXCHANGE_API_KEY

# Initialize a default API client (can be overridden with user's key)
default_api = StackExchangeAPI(api_key=STACK_EXCHANGE_API_KEY)

def get_api_client(api_key: str = "") -> StackExchangeAPI:
    """Get API client with user's key or fallback to default."""
    if api_key and api_key.strip():
        return StackExchangeAPI(api_key=api_key.strip())
    return default_api

def search_by_query_sync(
    query: str,
    tags: str = "",
    min_score: int = 0,
    has_accepted_answer: bool = False,
    limit: int = 5,
    response_format: str = "markdown",
    api_key: str = ""
) -> str:
    """
    Search Stack Overflow for questions matching a query.

    Args:
        query (str): The search query
        tags (str): Comma-separated list of tags to filter by (e.g., "python,pandas")
        min_score (int): Minimum score threshold for questions
        has_accepted_answer (bool): Whether questions must have an accepted answer
        limit (int): Maximum number of results to return (1-20)
        response_format (str): Format of response ("json" or "markdown")

    Returns:
        str: Formatted search results
    """
    if not query.strip():
        return "❌ Please enter a search query."
    
    # Convert tags string to list
    tags_list = [tag.strip() for tag in tags.split(",") if tag.strip()] if tags else None
    
    # Limit the range
    limit = max(1, min(limit, 20))
    
    try:
        # Get API client with user's key
        api = get_api_client(api_key)
        
        # Run the async function safely
        try:
            loop = asyncio.get_event_loop()
            if loop.is_closed():
                raise RuntimeError("Event loop is closed")
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        
        results = loop.run_until_complete(
            api.search_by_query(
                query=query,
                tags=tags_list,
                min_score=min_score if min_score > 0 else None,
                has_accepted_answer=has_accepted_answer if has_accepted_answer else None,
                limit=limit
            )
        )
        
        if not results:
            return f"🔍 No results found for query: '{query}'"
        
        return format_response(results, response_format)
    
    except Exception as e:
        return f"❌ Error searching Stack Overflow: {str(e)}"


def search_by_error_sync(
    error_message: str,
    language: str = "",
    technologies: str = "",
    min_score: int = 0,
    has_accepted_answer: bool = False,
    limit: int = 5,
    response_format: str = "markdown",
    api_key: str = ""
) -> str:
    """
    Search Stack Overflow for solutions to an error message.

    Args:
        error_message (str): The error message to search for
        language (str): Programming language (e.g., "python", "javascript")
        technologies (str): Comma-separated related technologies (e.g., "react,django")
        min_score (int): Minimum score threshold for questions
        has_accepted_answer (bool): Whether questions must have an accepted answer
        limit (int): Maximum number of results to return (1-20)
        response_format (str): Format of response ("json" or "markdown")

    Returns:
        str: Formatted search results
    """
    if not error_message.strip():
        return "❌ Please enter an error message."
    
    # Build tags list
    tags = []
    if language.strip():
        tags.append(language.strip().lower())
    if technologies.strip():
        tags.extend([tech.strip().lower() for tech in technologies.split(",") if tech.strip()])
    
    # Limit the range
    limit = max(1, min(limit, 20))
    
    try:
        # Get API client with user's key
        api = get_api_client(api_key)
        
        # Run the async function safely
        try:
            loop = asyncio.get_event_loop()
            if loop.is_closed():
                raise RuntimeError("Event loop is closed")
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        
        results = loop.run_until_complete(
            api.search_by_query(
                query=error_message,
                tags=tags if tags else None,
                min_score=min_score if min_score > 0 else None,
                has_accepted_answer=has_accepted_answer if has_accepted_answer else None,
                limit=limit
            )
        )
        
        if not results:
            return f"🔍 No results found for error: '{error_message}'"
        
        return format_response(results, response_format)
    
    except Exception as e:
        return f"❌ Error searching Stack Overflow: {str(e)}"


def get_question_sync(
    question_id: str,
    include_comments: bool = True,
    response_format: str = "markdown",
    api_key: str = ""
) -> str:
    """
    Get a specific Stack Overflow question by ID.

    Args:
        question_id (str): The Stack Overflow question ID
        include_comments (bool): Whether to include comments in results
        response_format (str): Format of response ("json" or "markdown")

    Returns:
        str: Formatted question details
    """
    if not question_id.strip():
        return "❌ Please enter a question ID."
    
    try:
        # Convert to int
        q_id = int(question_id.strip())
        
        # Get API client with user's key
        api = get_api_client(api_key)
        
        # Run the async function safely
        try:
            loop = asyncio.get_event_loop()
            if loop.is_closed():
                raise RuntimeError("Event loop is closed")
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        
        result = loop.run_until_complete(
            api.get_question(
                question_id=q_id,
                include_comments=include_comments
            )
        )
        
        return format_response([result], response_format)
    
    except ValueError:
        return "❌ Question ID must be a number."
    except Exception as e:
        return f"❌ Error fetching question: {str(e)}"


def analyze_stack_trace_sync(
    stack_trace: str,
    language: str,
    min_score: int = 0,
    has_accepted_answer: bool = False,
    limit: int = 3,
    response_format: str = "markdown",
    api_key: str = ""
) -> str:
    """
    Analyze a stack trace and find relevant solutions on Stack Overflow.

    Args:
        stack_trace (str): The stack trace to analyze
        language (str): Programming language of the stack trace
        min_score (int): Minimum score threshold for questions
        has_accepted_answer (bool): Whether questions must have an accepted answer
        limit (int): Maximum number of results to return (1-10)
        response_format (str): Format of response ("json" or "markdown")

    Returns:
        str: Formatted search results
    """
    if not stack_trace.strip():
        return "❌ Please enter a stack trace."
    
    if not language.strip():
        return "❌ Please specify the programming language."
    
    # Limit the range
    limit = max(1, min(limit, 10))
    
    # Extract the first line as the main error
    error_lines = stack_trace.strip().split("\n")
    error_message = error_lines[0]
    
    try:
        # Get API client with user's key
        api = get_api_client(api_key)
        
        # Run the async function safely
        try:
            loop = asyncio.get_event_loop()
            if loop.is_closed():
                raise RuntimeError("Event loop is closed")
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        
        results = loop.run_until_complete(
            api.search_by_query(
                query=error_message,
                tags=[language.strip().lower()],
                min_score=min_score if min_score > 0 else None,
                has_accepted_answer=has_accepted_answer if has_accepted_answer else None,
                limit=limit
            )
        )
        
        if not results:
            return f"🔍 No results found for stack trace error: '{error_message}'"
        
        return format_response(results, response_format)
    
    except Exception as e:
        return f"❌ Error analyzing stack trace: {str(e)}"


def advanced_search_sync(
    query: str = "",
    tags: str = "",
    excluded_tags: str = "",
    min_score: int = 0,
    title: str = "",
    body: str = "",
    min_answers: int = 0,
    has_accepted_answer: bool = False,
    min_views: int = 0,
    sort_by: str = "votes",
    limit: int = 5,
    response_format: str = "markdown",
    api_key: str = ""
) -> str:
    """
    Advanced search for Stack Overflow questions with comprehensive filters.

    Args:
        query (str): Free-form search query
        tags (str): Comma-separated list of tags to filter by
        excluded_tags (str): Comma-separated list of tags to exclude
        min_score (int): Minimum score threshold
        title (str): Text that must appear in the title
        body (str): Text that must appear in the body
        min_answers (int): Minimum number of answers
        has_accepted_answer (bool): Whether questions must have an accepted answer
        min_views (int): Minimum number of views
        sort_by (str): Field to sort by (activity, creation, votes, relevance)
        limit (int): Maximum number of results to return (1-20)
        response_format (str): Format of response ("json" or "markdown")

    Returns:
        str: Formatted search results
    """
    if not query.strip() and not tags.strip() and not title.strip() and not body.strip():
        return "❌ Please provide at least one search criteria (query, tags, title, or body)."
    
    # Convert tags strings to lists
    tags_list = [tag.strip() for tag in tags.split(",") if tag.strip()] if tags else None
    excluded_tags_list = [tag.strip() for tag in excluded_tags.split(",") if tag.strip()] if excluded_tags else None
    
    # Limit the range
    limit = max(1, min(limit, 20))
    
    try:
        # Get API client with user's key
        api = get_api_client(api_key)
        
        # Run the async function safely
        try:
            loop = asyncio.get_event_loop()
            if loop.is_closed():
                raise RuntimeError("Event loop is closed")
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        
        results = loop.run_until_complete(
            api.advanced_search(
                query=query.strip() if query.strip() else None,
                tags=tags_list,
                excluded_tags=excluded_tags_list,
                min_score=min_score if min_score > 0 else None,
                title=title.strip() if title.strip() else None,
                body=body.strip() if body.strip() else None,
                answers=min_answers if min_answers > 0 else None,
                has_accepted_answer=has_accepted_answer if has_accepted_answer else None,
                views=min_views if min_views > 0 else None,
                sort_by=sort_by,
                limit=limit
            )
        )
        
        if not results:
            return "🔍 No results found with the specified criteria."
        
        return format_response(results, response_format)
    
    except Exception as e:
        return f"❌ Error performing advanced search: {str(e)}"


# Helper functions for example buttons (not exposed as MCP tools)
def _set_django_example():
    return ("Django pagination best practices", "python,django", 5, True, 5, "markdown")

def _set_async_example():
    return ("Python asyncio concurrency patterns", "python,asyncio", 10, True, 5, "markdown")

def _set_react_example():
    return ("React hooks useState useEffect", "javascript,reactjs", 15, True, 5, "markdown")

def _set_sql_example():
    return ("SQL INNER JOIN vs LEFT JOIN performance", "sql,join", 20, True, 5, "markdown")


# Create the Gradio interface with multiple tabs
with gr.Blocks(
    title="Stack Overflow MCP Server",
    theme=gr.themes.Soft(),
    css="""
    .gradio-container {
        max-width: 1200px !important;
    }
    .tab-nav button {
        font-size: 16px !important;
    }
    """
) as demo:
    
    gr.Markdown("""
    # 🔍 Stack Overflow MCP Server
    
    **A powerful interface to search Stack Overflow and analyze programming errors**
    
    This application serves as both a web interface and an MCP (Model Context Protocol) server, 
    allowing AI assistants like Claude to search Stack Overflow programmatically.
    
    💡 **MCP Server URL**: Use this URL in your MCP client: `{SERVER_URL}/gradio_api/mcp/sse`
    
    ## 🚀 Quick Start Examples
    
    Try these example searches to get started:
    - **General Search**: "Django pagination best practices" with tags "python,django"
    - **Error Search**: "TypeError: 'NoneType' object has no attribute" in Python
    - **Question ID**: 11227809 (famous "Why is processing a sorted array faster?" question)
    - **Stack Trace**: JavaScript TypeError examples
    - **Advanced**: High-scored Python questions with accepted answers
    """)
    
    # Global API Key Input
    with gr.Row():
        with gr.Column(scale=3):
            gr.Markdown("### 🔑 Stack Exchange API Key (Optional)")
            gr.Markdown("""
            **Why provide an API key?**
            - Higher request quotas (10,000 vs 300 requests/day)
            - Faster responses and better reliability
            - API keys are **not secret** - safe to share publicly
            
            **How to get one:**
            1. Visit [Stack Apps OAuth Registration](https://stackapps.com/apps/oauth/register)
            2. Fill in basic info (name: "Stack Overflow MCP", domain: "localhost")
            3. Copy your API key from the results page
            """)
        
        with gr.Column(scale=2):
            api_key_input = gr.Textbox(
                label="Stack Exchange API Key",
                placeholder="Enter your API key here (optional)",
                value="",
                type="password",
                info="Optional: Provides higher quotas and better performance"
            )
    
    with gr.Tabs():
        
        # Tab 1: General Search
        with gr.Tab("🔍 General Search", id="search"):
            gr.Markdown("### Search Stack Overflow by query and filters")
            
            with gr.Row():
                with gr.Column(scale=2):
                    query_input = gr.Textbox(
                        label="Search Query",
                        placeholder="e.g., 'Django pagination best practices'",
                        value="python list comprehension"
                    )
                    
                with gr.Column(scale=1):
                    tags_input = gr.Textbox(
                        label="Tags (comma-separated)",
                        placeholder="e.g., python,pandas",
                        value=""
                    )
            
            with gr.Row():
                min_score_input = gr.Slider(
                    label="Minimum Score",
                    minimum=0,
                    maximum=100,
                    value=0,
                    step=1
                )
                
                has_accepted_input = gr.Checkbox(
                    label="Must have accepted answer",
                    value=False
                )
                
                limit_input = gr.Slider(
                    label="Number of Results",
                    minimum=1,
                    maximum=20,
                    value=5,
                    step=1
                )
                
                format_input = gr.Dropdown(
                    label="Response Format",
                    choices=["markdown", "json"],
                    value="markdown"
                )
            
            search_btn = gr.Button("🔍 Search", variant="primary", size="lg")
            
            # Example buttons
            with gr.Row():
                gr.Markdown("**Quick Examples:**")
            with gr.Row():
                example1_btn = gr.Button("Django Pagination", size="sm")
                example2_btn = gr.Button("Python Async", size="sm") 
                example3_btn = gr.Button("React Hooks", size="sm")
                example4_btn = gr.Button("SQL JOIN", size="sm")
            
            # Example button click handlers
            example1_btn.click(
                lambda: ("Django pagination best practices", "python,django", 5, True, 5, "markdown"),
                outputs=[query_input, tags_input, min_score_input, has_accepted_input, limit_input, format_input]
            )
            example2_btn.click(
                lambda: ("Python asyncio concurrency patterns", "python,asyncio", 10, True, 5, "markdown"),
                outputs=[query_input, tags_input, min_score_input, has_accepted_input, limit_input, format_input]
            )
            example3_btn.click(
                lambda: ("React hooks useState useEffect", "javascript,reactjs", 15, True, 5, "markdown"),
                outputs=[query_input, tags_input, min_score_input, has_accepted_input, limit_input, format_input]
            )
            example4_btn.click(
                lambda: ("SQL INNER JOIN vs LEFT JOIN performance", "sql,join", 20, True, 5, "markdown"),
                outputs=[query_input, tags_input, min_score_input, has_accepted_input, limit_input, format_input]
            )
            
            search_output = gr.Markdown(label="Search Results", height=400)
            
            search_btn.click(
                fn=search_by_query_sync,
                inputs=[query_input, tags_input, min_score_input, has_accepted_input, limit_input, format_input, api_key_input],
                outputs=search_output
            )
        
        # Tab 2: Error Search
        with gr.Tab("🐛 Error Search", id="error"):
            gr.Markdown("### Find solutions for specific error messages")
            
            with gr.Row():
                with gr.Column(scale=2):
                    error_input = gr.Textbox(
                        label="Error Message",
                        placeholder="e.g., 'TypeError: object of type 'NoneType' has no len()'",
                        value="TypeError: 'NoneType' object has no attribute"
                    )
                
                with gr.Column(scale=1):
                    language_input = gr.Textbox(
                        label="Programming Language",
                        placeholder="e.g., python",
                        value="python"
                    )
            
            tech_input = gr.Textbox(
                label="Related Technologies (comma-separated)",
                placeholder="e.g., django,flask",
                value=""
            )
            
            with gr.Row():
                error_min_score = gr.Slider(
                    label="Minimum Score",
                    minimum=0,
                    maximum=100,
                    value=0,
                    step=1
                )
                
                error_accepted = gr.Checkbox(
                    label="Must have accepted answer",
                    value=True
                )
                
                error_limit = gr.Slider(
                    label="Number of Results",
                    minimum=1,
                    maximum=20,
                    value=5,
                    step=1
                )
                
                error_format = gr.Dropdown(
                    label="Response Format",
                    choices=["markdown", "json"],
                    value="markdown"
                )
            
            error_search_btn = gr.Button("🐛 Search for Solutions", variant="primary", size="lg")
            error_output = gr.Markdown(label="Error Solutions", height=400)
            
            error_search_btn.click(
                fn=search_by_error_sync,
                inputs=[error_input, language_input, tech_input, error_min_score, error_accepted, error_limit, error_format, api_key_input],
                outputs=error_output
            )
        
        # Tab 3: Get Specific Question
        with gr.Tab("📄 Get Question", id="question"):
            gr.Markdown("### Retrieve a specific Stack Overflow question by ID")
            
            with gr.Row():
                question_id_input = gr.Textbox(
                    label="Question ID",
                    placeholder="e.g., 11227809",
                    value="11227809"
                )
                
                question_comments = gr.Checkbox(
                    label="Include Comments",
                    value=True
                )
                
                question_format = gr.Dropdown(
                    label="Response Format",
                    choices=["markdown", "json"],
                    value="markdown"
                )
            
            question_btn = gr.Button("📄 Get Question", variant="primary", size="lg")
            question_output = gr.Markdown(label="Question Details", height=400)
            
            question_btn.click(
                fn=get_question_sync,
                inputs=[question_id_input, question_comments, question_format, api_key_input],
                outputs=question_output
            )
        
        # Tab 4: Stack Trace Analysis
        with gr.Tab("📊 Stack Trace Analysis", id="trace"):
            gr.Markdown("### Analyze stack traces and find relevant solutions")
            
            stack_trace_input = gr.Textbox(
                label="Stack Trace",
                placeholder="Paste your full stack trace here...",
                lines=8,
                value="TypeError: Cannot read property 'length' of undefined\n    at Array.map (<anonymous>)\n    at Component.render (app.js:42:18)"
            )
            
            with gr.Row():
                trace_language = gr.Textbox(
                    label="Programming Language",
                    placeholder="e.g., javascript",
                    value="javascript"
                )
                
                trace_min_score = gr.Slider(
                    label="Minimum Score",
                    minimum=0,
                    maximum=100,
                    value=5,
                    step=1
                )
                
                trace_accepted = gr.Checkbox(
                    label="Must have accepted answer",
                    value=True
                )
                
                trace_limit = gr.Slider(
                    label="Number of Results",
                    minimum=1,
                    maximum=10,
                    value=3,
                    step=1
                )
                
                trace_format = gr.Dropdown(
                    label="Response Format",
                    choices=["markdown", "json"],
                    value="markdown"
                )
            
            trace_btn = gr.Button("📊 Analyze Stack Trace", variant="primary", size="lg")
            trace_output = gr.Markdown(label="Stack Trace Analysis", height=400)
            
            trace_btn.click(
                fn=analyze_stack_trace_sync,
                inputs=[stack_trace_input, trace_language, trace_min_score, trace_accepted, trace_limit, trace_format, api_key_input],
                outputs=trace_output
            )
        
        # Tab 5: Advanced Search
        with gr.Tab("⚙️ Advanced Search", id="advanced"):
            gr.Markdown("### Advanced search with comprehensive filtering options")
            
            with gr.Row():
                with gr.Column():
                    adv_query_input = gr.Textbox(
                        label="Search Query (optional)",
                        placeholder="e.g., 'memory management'",
                        value=""
                    )
                    
                    adv_title_input = gr.Textbox(
                        label="Title Contains (optional)",
                        placeholder="Text that must appear in the title",
                        value=""
                    )
                    
                    adv_body_input = gr.Textbox(
                        label="Body Contains (optional)",
                        placeholder="Text that must appear in the body",
                        value=""
                    )
                
                with gr.Column():
                    adv_tags_input = gr.Textbox(
                        label="Include Tags (comma-separated)",
                        placeholder="e.g., python,django,performance",
                        value=""
                    )
                    
                    adv_excluded_tags_input = gr.Textbox(
                        label="Exclude Tags (comma-separated)",
                        placeholder="e.g., beginner,homework",
                        value=""
                    )
                    
                    adv_sort_input = gr.Dropdown(
                        label="Sort By",
                        choices=["votes", "activity", "creation", "relevance"],
                        value="votes"
                    )
            
            with gr.Row():
                adv_min_score = gr.Slider(
                    label="Minimum Score",
                    minimum=0,
                    maximum=500,
                    value=10,
                    step=5
                )
                
                adv_min_answers = gr.Slider(
                    label="Minimum Answers",
                    minimum=0,
                    maximum=50,
                    value=1,
                    step=1
                )
                
                adv_min_views = gr.Slider(
                    label="Minimum Views",
                    minimum=0,
                    maximum=10000,
                    value=0,
                    step=100
                )
            
            with gr.Row():
                adv_accepted = gr.Checkbox(
                    label="Must have accepted answer",
                    value=False
                )
                
                adv_limit = gr.Slider(
                    label="Number of Results",
                    minimum=1,
                    maximum=20,
                    value=5,
                    step=1
                )
                
                adv_format = gr.Dropdown(
                    label="Response Format",
                    choices=["markdown", "json"],
                    value="markdown"
                )
            
            adv_search_btn = gr.Button("⚙️ Advanced Search", variant="primary", size="lg")
            adv_output = gr.Markdown(label="Advanced Search Results", height=400)
            
            adv_search_btn.click(
                fn=advanced_search_sync,
                inputs=[
                    adv_query_input, adv_tags_input, adv_excluded_tags_input, 
                    adv_min_score, adv_title_input, adv_body_input, adv_min_answers,
                    adv_accepted, adv_min_views, adv_sort_input, adv_limit, adv_format, api_key_input
                ],
                outputs=adv_output
            )
    
    # Footer with MCP information
    gr.Markdown("""
    ---
    
    ## 🤖 MCP Integration
    
    This app also functions as an **MCP (Model Context Protocol) Server**! 
    
    To use with AI assistants like Claude Desktop, add this configuration:
    
    ```json
    {
      "mcpServers": {
        "stackoverflow": {
          "url": "YOUR_DEPLOYED_URL/gradio_api/mcp/sse"
        }
      }
    }
    ```
    
    **Available MCP Tools:**
    - `search_by_query_sync` - General Stack Overflow search
    - `search_by_error_sync` - Error-specific search
    - `get_question_sync` - Get specific question by ID
    - `analyze_stack_trace_sync` - Analyze stack traces
    - `advanced_search_sync` - Advanced search with comprehensive filters
    
    **💡 Pro Tip:** Add your Stack Exchange API key above for higher quotas (10,000 vs 300 requests/day)!
    
    Built with ❤️ for the MCP Hackathon
    """)


if __name__ == "__main__":
    # Launch with MCP server enabled
    demo.launch(
        mcp_server=True,
        share=True,  # Create a public link for testing
        server_name="0.0.0.0",  # Allow external connections
        server_port=7860,  # Standard Gradio port
        show_error=True
    )