Spaces:
Sleeping
Sleeping
| import json | |
| from textblob import TextBlob | |
| from fastapi import FastAPI, HTTPException | |
| import uvicorn | |
| import gradio as gr | |
| import mcp | |
| app = FastAPI() | |
| def analyze_sentiment(text: str) -> dict: | |
| """Core sentiment analysis logic""" | |
| 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" | |
| } | |
| async def handle_mcp_request(data: dict): | |
| """ | |
| MCP-compatible endpoint | |
| Format: {"parameters": {"text": "your text"}} | |
| """ | |
| try: | |
| text = data.get("parameters", {}).get("text", "") | |
| if not text: | |
| raise HTTPException(status_code=400, detail="Missing 'text' parameter") | |
| return { | |
| "jsonrpc": "2.0", | |
| "result": analyze_sentiment(text), | |
| "id": "sentiment-response" | |
| } | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def health_check(): | |
| """Required for Hugging Face health checks""" | |
| return {"status": "OK"} | |
| demo = gr.Interface( | |
| fn=analyze_sentiment, | |
| inputs=gr.Textbox(placeholder="Enter text to analyze..."), | |
| outputs=gr.Textbox(), | |
| title="Text Sentiment Analysis", | |
| description="Analyze the sentiment of text using TextBlob" | |
| ) | |
| # Launch the interface and MCP server | |
| if __name__ == "__main__": | |
| demo.launch(mcp_server=True) | |
| uvicorn.run(app, host="0.0.0.0", port=7860) # HF uses 7860 |