app.py

#!/usr/bin/env python3
"""Enhanced MCP Server using Gradio's built-in MCP functionality.
This server exposes 5 specialized tools via Gradio MCP integration.
"""
import gradio as gr
import json
import os
from typing import Any, Dict, List, Optional
import argparse

# Configuration
RIDB_API_KEY = "d3e5b760-226d-4e09-99db-1f9fc514775c"

# Set environment variable for tools that need it
os.environ['RIDB_API_KEY'] = RIDB_API_KEY

# Import our enhanced tools
from tools.sentiment_tool import SentimentTool
from tools.place_tool import PlaceSearchTool
from tools.restaurant_tool import RestaurantSearchTool
from tools.hiking_tool import HikingSearchTool
from tools.web_search_tool import WebSearchTool

# Import the formatting service
from services.formatting_service import ResponseFormatter

# Initialize tools
def initialize_tools() -> Dict[str, Any]:
    """Initialize all available tools."""
    tools = {}
    
    try:
        tools['sentiment_analysis'] = SentimentTool()
        print("✅ Sentiment analysis tool initialized")
    except Exception as e:
        print(f"❌ Failed to initialize sentiment tool: {e}")
        
    try:
        tools['place_search'] = PlaceSearchTool()
        print("✅ Place search tool initialized")
    except Exception as e:
        print(f"❌ Failed to initialize place search tool: {e}")
        
    try:
        tools['restaurant_search'] = RestaurantSearchTool()
        print("✅ Restaurant search tool initialized")
    except Exception as e:
        print(f"❌ Failed to initialize restaurant search tool: {e}")
        
    try:
        tools['hiking_search'] = HikingSearchTool()
        print("✅ Hiking search tool initialized")
    except Exception as e:
        print(f"❌ Failed to initialize hiking search tool: {e}")
        
    try:
        tools['web_search'] = WebSearchTool()
        print("✅ Web search tool initialized")
    except Exception as e:
        print(f"❌ Failed to initialize web search tool: {e}")
    
    return tools

# Tool wrapper functions for Gradio MCP
def sentiment_analysis(text: str) -> str:
    """Analyze the sentiment of the given text using TextBlob."""
    tool = tools['sentiment_analysis']
    raw_result = tool.forward(text)
    
    # Handle error cases
    if 'error' in raw_result:
        return ResponseFormatter.format_error(raw_result['error'])
    
    # Extract the required arguments from results
    text_analyzed = raw_result.get('text', text)
    polarity = raw_result.get('polarity', 0.0)
    subjectivity = raw_result.get('subjectivity', 0.5)
    assessment = raw_result.get('assessment', 'neutral')
    
    # Format the result with correct arguments
    return ResponseFormatter.format_sentiment_response(text_analyzed, polarity, subjectivity, assessment)

def place_search(query: str, max_distance: int = 20) -> dict:
    """Search for hotels and accommodations in a specified location."""
    tool = tools['place_search']
    results = tool.forward(location=query, max_distance=max_distance)
    
    # Handle error cases
    if 'error' in results:
        return ResponseFormatter.format_error(results['error'])
    
    # Extract location and places from results
    location = results.get('location', query)
    places = results.get('places', [])
    
    # Format the response with correct arguments
    return ResponseFormatter.format_place_response(location, places)

def restaurant_search(query: str, cuisine: str = None) -> dict:
    """Find restaurants and dining options in a specified location."""
    tool = tools['restaurant_search']
    results = tool.forward(location=query, cuisine=cuisine)
    
    # Handle error cases
    if 'error' in results:
        return ResponseFormatter.format_error(results['error'])
    
    # Extract the required arguments from results
    location = results.get('location', query)
    # The service returns distance as a string like "1.0 miles", extract the number
    distance_str = results.get('distance', '1.0 miles')
    try:
        distance = float(distance_str.split()[0])  # Extract number from "1.0 miles"
    except (ValueError, IndexError):
        distance = 1.0  # Default
    
    # The service returns 'top_restaurants', but formatter expects 'restaurants'
    restaurants = results.get('top_restaurants', [])
    
    # Format the response with correct arguments
    return ResponseFormatter.format_restaurant_response(location, distance, restaurants)

def hiking_search(location: str, difficulty: str = None, max_distance: int = 50) -> dict:
    """Discover hiking trails and outdoor activities in a specified location."""
    tool = tools['hiking_search']
    results = tool.forward(location, difficulty, max_distance)
    
    # Handle error cases
    if 'error' in results:
        return ResponseFormatter.format_error(results['error'])
    
    # Extract the required arguments from results
    location_name = results.get('location', location)
    max_dist = max_distance  # We already have this from the function parameter
    difficulty_filter = results.get('difficulty_filter', difficulty or 'All')
    trails = results.get('trails', [])
    stats = results.get('stats', {})
    
    # Format the response with correct arguments
    return ResponseFormatter.format_hiking_response(location_name, max_dist, difficulty_filter, trails, stats)

def web_search(query: str, max_results: int = 5) -> dict:
    """Search the web for information with intelligent category matching."""
    tool = tools['web_search']
    raw_result = tool.forward(query, max_results)
    
    # Format the result beautifully using the formatting service
    if 'error' not in raw_result:
        formatted_result = ResponseFormatter.format_web_search_results(raw_result)
        return {
            "formatted_output": formatted_result,
            "raw_data": raw_result
        }
    else:
        # Return error with formatting
        formatted_result = ResponseFormatter.format_web_search_results(raw_result)
        return {
            "formatted_output": formatted_result,
            "raw_data": raw_result
        }

def create_gradio_mcp_server():
    """Create Gradio MCP server with individual tool interfaces."""
    
    # Create individual interfaces for each tool
    # These will be automatically exposed as MCP tools
    sentiment_interface = gr.Interface(
        fn=sentiment_analysis,
        inputs=[gr.Textbox(placeholder="Enter text to analyze", label="text")],
        outputs=[gr.Markdown(label="Sentiment Analysis Result")],
        title="🎭 Sentiment Analysis",
        description="Analyze the sentiment of the given text using TextBlob."
    )
    
    place_interface = gr.Interface(
        fn=place_search,
        inputs=[
            gr.Textbox(placeholder="Enter location", label="query"),
            gr.Slider(minimum=1, maximum=100, value=20, label="max_distance")
        ],
        outputs=[gr.Markdown(label="result")],
        title="🏨 Place Search",
        description="Search for hotels and accommodations in a specified location."
    )
    
    restaurant_interface = gr.Interface(
        fn=restaurant_search,
        inputs=[
            gr.Textbox(placeholder="Enter location", label="query"),
            gr.Textbox(placeholder="Enter cuisine type (optional)", label="cuisine", value="")
        ],
        outputs=[gr.Markdown(label="result")],
        title="🍽️ Restaurant Search",
        description="Find restaurants and dining options in a specified location."
    )
    
    hiking_interface = gr.Interface(
        fn=hiking_search,
        inputs=[
            gr.Textbox(placeholder="Enter location", label="location"),
            gr.Textbox(placeholder="Enter difficulty (optional)", label="difficulty", value=""),
            gr.Slider(minimum=1, maximum=200, value=50, label="max_distance")
        ],
        outputs=[gr.Markdown(label="result")],
        title="🥾 Hiking Search",
        description="Discover hiking trails and outdoor activities in a specified location."
    )
    
    web_interface = gr.Interface(
        fn=web_search,
        inputs=[
            gr.Textbox(placeholder="Enter search query", label="query"),
            gr.Slider(minimum=1, maximum=10, value=5, label="max_results")
        ],
        outputs=[gr.JSON(label="Search Results")],
        title="🌐 Web Search",
        description="Search the web for information with intelligent category matching."
    )
    
    # Create a combined tabbed interface
    demo = gr.TabbedInterface(
        [sentiment_interface, place_interface, restaurant_interface, hiking_interface, web_interface],
        ["🎭 Sentiment Analysis", "🏨 Place Search", "🍽️ Restaurant Search", "🥾 Hiking Search", "🌐 Web Search"],
        title="🚀 Enhanced MCP Server with 5 Specialized Tools"
    )
    
    return demo

def main():
    """Main entry point for the Enhanced MCP Server."""
    parser = argparse.ArgumentParser(description="Enhanced MCP Server with Gradio Integration")
    parser.add_argument(
        "--port",
        type=int,
        default=7860,
        help="Port to run the server (default: 7861)"
    )
    parser.add_argument(
        "--host",
        default="0.0.0.0",
        help="Host to bind the server (default: 0.0.0.0)"
    )
    
    args = parser.parse_args()
    
    print("🚀 Starting Enhanced MCP Server with Gradio Integration")
    print("✨ Using Gradio's built-in MCP functionality")
    print("")
    print("🔧 Initializing Tools:")
    
    # Initialize tools globally
    global tools
    tools = initialize_tools()
    
    print(f"✅ {len(tools)} tool instances initialized successfully")
    print("")
    print("🔧 Server Features:")
    for tool_name, tool_instance in tools.items():
        print(f"  • {tool_instance.description}")
    print("")
    print(f"📡 Server URL: http://{args.host}:{args.port}")
    print(f"🔗 MCP Endpoint: http://{args.host}:{args.port}/gradio_api/mcp/sse")
    print("═" * 60)
    
    # Create and launch Gradio MCP server
    demo = create_gradio_mcp_server()
    
    print("🌟 Server ready! Tools available via MCP protocol and web interface!")
    
    # Launch with MCP server enabled
    demo.launch(
        server_name=args.host,
        server_port=args.port,
        share=False,
        show_error=True,
        mcp_server=True  # Enable Gradio's MCP functionality
    )

if __name__ == "__main__":
    main()