app.py

import json
import warnings
import asyncio
from textblob import TextBlob
import gradio as gr
from mcp import Tool
from mcp.server import Server
from mcp.types import TextContent
import uvicorn
import threading

# ============================================================
# Suppress harmless asyncio warnings from Gradio
# ============================================================
warnings.filterwarnings("ignore", message=".*Invalid file descriptor.*")

# ============================================================
# Shared sentiment analysis function
# ============================================================
def sentiment_analysis(text: str) -> dict:
    """Analyze the sentiment of the given text."""
    if not text or not text.strip():
        return {
            "error": "No text provided",
            "polarity": 0,
            "subjectivity": 0,
            "assessment": "neutral"
        }

    blob = TextBlob(text)
    sentiment = blob.sentiment

    return {
        "polarity": round(sentiment.polarity, 2),
        "subjectivity": round(sentiment.subjectivity, 2),
        "assessment": (
            "positive" if sentiment.polarity > 0 else
            "negative" if sentiment.polarity < 0 else
            "neutral"
        )
    }

# ============================================================
# Gradio wrapper for JSON output
# ============================================================
def sentiment_analysis_gradio(text: str) -> str:
    result = sentiment_analysis(text)
    return json.dumps(result, indent=2)

# ============================================================
# Create MCP server
# ============================================================
mcp_server = Server("sentiment-analysis")

@mcp_server.list_tools()
async def handle_list_tools() -> list[Tool]:
    return [
        Tool(
            name="analyze_sentiment",
            description=(
                "Analyze the sentiment of text using TextBlob. "
                "Returns polarity (-1 to 1), subjectivity (0 to 1), "
                "and assessment (positive/negative/neutral)."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {
                        "type": "string",
                        "description": "The text to analyze for sentiment"
                    }
                },
                "required": ["text"]
            }
        )
    ]

@mcp_server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "analyze_sentiment":
        text = arguments.get("text", "")
        result = sentiment_analysis(text)
        return [TextContent(type="text", text=json.dumps(result, indent=2))]
    return [TextContent(type="text", text=f"Unknown tool: {name}")]

# ============================================================
# MCP ASGI app (Server object is already ASGI-compatible)
# ============================================================
mcp_asgi_app = mcp_server

# ============================================================
# Gradio interface
# ============================================================
demo = gr.Interface(
    fn=sentiment_analysis_gradio,
    inputs=gr.Textbox(
        placeholder="Enter text to analyze...",
        label="Input Text",
        lines=5
    ),
    outputs=gr.Textbox(
        label="Sentiment Analysis Result (JSON)",
        lines=10
    ),
    title="Text Sentiment Analysis",
    description="Analyze the sentiment of text using TextBlob. MCP server available at /sse",
    examples=[
        ["I absolutely love this product! It's amazing and works perfectly."],
        ["This is the worst experience I've ever had. Terrible service."],
        ["The weather today is cloudy with a chance of rain."],
    ]
)

# # ============================================================
# # Main async runner for MCP + Gradio
# # ============================================================
# async def main():
#     # Run MCP server
#     print("main line 0")
#     config = uvicorn.Config(mcp_asgi_app, host="0.0.0.0", port=7860, log_level="info")
#     print("main line 1")
#     server = uvicorn.Server(config)
#     print("main line 2")
#     mcp_task = asyncio.create_task(server.serve())
#     print("main line 3")

#     # Launch Gradio UI
#     demo.launch(
#         server_name="0.0.0.0",
#         server_port=8000,
#         prevent_thread_lock=True,
#         ssr_mode=False
#     )
#     print("main line 4")

#     # Keep MCP server alive
#     await mcp_task
#     print("main line 5")

# # ============================================================
# # Entry point
# # ============================================================
# if __name__ == "__main__":
#     # print("=" * 60)
#     # print("Starting Sentiment Analysis Services")
#     # print("=" * 60)
#     # print("🔌 MCP Server: http://0.0.0.0:7860/sse")
#     # print("📊 Gradio UI : http://0.0.0.0:8000")
#     # print("=" * 60)
#     asyncio.run(main())

def run_mcp_server():
    async def start():
        config = uvicorn.Config(mcp_asgi_app, host="0.0.0.0", port=8000)
        server = uvicorn.Server(config)
        await server.serve()
    
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(start())

if __name__ == "__main__":    
    threading.Thread(target=run_mcp_server, daemon=True).start()
    demo.launch(server_name="0.0.0.0", server_port=7860, ssr_mode=False)