Blame for app.py

import json
import os
import requests
import gradio as gr
from smolagents import Tool, CodeAgent, InferenceClientModel

class SentimentAnalysisTool(Tool):
    """
    Tool that connects to the MCP sentiment analysis server.
    Analyzes the sentiment of text using the remote MCP server.
    """
    name = "sentiment_analysis"
    description = """
    Analyze the sentiment of text to determine if it's positive, negative, or neutral.
    Use this tool when the user asks about sentiment, emotion, feeling, or opinion in text.
    Returns polarity score (-1 to 1), subjectivity (0 to 1), and assessment (positive/negative/neutral).
    This tool connects to the MCP sentiment analysis server.
    """

    inputs = {
        "text": {
            "type": "string",
            "description": "The text to analyze for sentiment",
        }
    }

    output_type = "string"

    def __init__(self, mcp_server_url: str = "https://s2v-general-mcp-sentiment.hf.space/gradio_api/mcp/"):
        """
        Initialize the sentiment analysis tool with MCP server URL.
        
        Args:
            mcp_server_url: The URL of the MCP sentiment analysis server
        """
        super().__init__()
        self.mcp_server_url = mcp_server_url

    def forward(self, text: str) -> str:
        """
        Call the MCP sentiment analysis server to analyze text.
        
        Args:
            text: The text to analyze
            
        Returns:
            JSON string with sentiment analysis results
        """
        try:
            # Call the Gradio MCP endpoint
            # Gradio MCP endpoints typically use SSE or JSON-RPC over HTTP
            # Try calling the function directly via the Gradio API
            api_url = self.mcp_server_url.rstrip('/')
            if not api_url.endswith('/api'):
                # Try the Gradio predict endpoint
                predict_url = api_url.replace('/gradio_api/mcp/', '/api/predict/')
            else:
                predict_url = api_url
            
            # Try calling via Gradio's predict API
            payload = {
                "data": [text],
                "fn_index": 0
            }
            
            response = requests.post(
                predict_url,
                json=payload,
                headers={"Content-Type": "application/json"},
                timeout=10
            )
            
            if response.status_code == 200:
                result = response.json()
                # Gradio returns {"data": [result]}
                if "data" in result and len(result["data"]) > 0:
                    return result["data"][0]
                else:
                    return json.dumps(result, indent=2)
            else:
                # Fallback: Use TextBlob directly (same as MCP server)
                from textblob import TextBlob
                blob = TextBlob(text)
                sentiment = blob.sentiment
                result = {
                    "polarity": round(sentiment.polarity, 2),
                    "subjectivity": round(sentiment.subjectivity, 2),
                    "assessment": "positive" if sentiment.polarity > 0 else "negative" if sentiment.polarity < 0 else "neutral"
                }
                return json.dumps(result, indent=2)
                
        except requests.exceptions.RequestException:
            # Fallback: Use TextBlob directly (same as MCP server)
            try:
                from textblob import TextBlob
                blob = TextBlob(text)
                sentiment = blob.sentiment
                result = {
                    "polarity": round(sentiment.polarity, 2),
                    "subjectivity": round(sentiment.subjectivity, 2),
                    "assessment": "positive" if sentiment.polarity > 0 else "negative" if sentiment.polarity < 0 else "neutral"
                }
                return json.dumps(result, indent=2)
            except Exception as e:
                return f"Error analyzing sentiment: {str(e)}"
        except Exception as e:
            return f"Error: {str(e)}"


# Initialize the sentiment analysis tool with MCP server connection
sentiment_tool = SentimentAnalysisTool()

# Create the agent with the MCP sentiment analysis tool
# ============================================================================
# LLM Configuration - Current: Qwen/Qwen2.5-72B-Instruct via HuggingFace Inference API
# ============================================================================
# 
# API KEY SETUP:
#   
#   HUGGINGFACE SPACE DEPLOYMENT:
#   - HF_TOKEN is automatically available in HuggingFace Spaces
#   - You can also set secrets in Space Settings → Secrets
#   - Add secret: HF_TOKEN with your token value
#
#   LOCAL SETUP:
#   1. Environment variable: export HF_TOKEN="your_huggingface_token"
#   2. Or login via: huggingface-cli login
#
#   Get your HuggingFace token from: https://huggingface.co/settings/tokens
#
# PRIMARY MODEL (Currently Active):
#   Model: Qwen/Qwen2.5-72B-Instruct
#   Provider: HuggingFace Inference API (default, no provider specified)
#   Usage: High-performance model via HuggingFace Inference API
#
# ALTERNATIVE MODELS (for rate limit fallbacks):
#   If you hit rate limits, uncomment one of these alternatives:
#
# Option 1: Other HuggingFace Inference API models (uses HF_TOKEN)
#   model=InferenceClientModel(model_id="meta-llama/Meta-Llama-3-8B-Instruct", token=api_token)
#   model=InferenceClientModel(model_id="mistralai/Mistral-7B-Instruct-v0.2", token=api_token)
#   model=InferenceClientModel(model_id="Qwen/Qwen2.5-7B-Instruct", token=api_token)
#   model=InferenceClientModel(model_id="google/gemma-2-2b-it", token=api_token)
#   model=InferenceClientModel(model_id="Qwen/Qwen2.5-32B-Instruct", token=api_token)
#   model=InferenceClientModel(model_id="meta-llama/Llama-3.1-70B-Instruct", token=api_token)
#
# Option 2: OpenAI (requires API key: export OPENAI_API_KEY="your_key")
#   from smolagents import OpenAIModel
#   model=OpenAIModel(model="gpt-4o-mini")  # or "gpt-4o", "gpt-3.5-turbo"
#
# Option 3: Anthropic Claude (requires API key: export ANTHROPIC_API_KEY="your_key")
#   from smolagents import AnthropicModel
#   model=AnthropicModel(model="claude-3-5-sonnet-20241022")  # or "claude-3-opus-20240229"
#
# Option 4: Local Ollama (for offline/local use)
#   from smolagents import OllamaModel
#   model=OllamaModel(model="llama3.1")  # or "mistral", "qwen", etc.
#
# ============================================================================

# Get API token from environment variables
# Priority: HF_TOKEN > HUGGINGFACE_TOKEN
# In HuggingFace Spaces, HF_TOKEN is automatically available
api_token = (
    os.getenv("HF_TOKEN") or 
    os.getenv("HUGGINGFACE_TOKEN")
)

# Check if running in HuggingFace Space
is_hf_space = os.getenv("SPACE_ID") is not None

# Initialize the model with API token
# Using HuggingFace Inference API (no provider specified = default HF Inference API)
if api_token:
    try:
        agent = CodeAgent(
            tools=[sentiment_tool],
            model=InferenceClientModel(
                model_id="Qwen/Qwen2.5-72B-Instruct",
                token=api_token  # Uses HuggingFace Inference API by default
            )
        )
        if is_hf_space:
            print("✅ Agent initialized successfully using HuggingFace Space token")
        else:
            print("✅ Agent initialized successfully using local API token")
    except Exception as e:
        print(f"❌ Error initializing agent with token: {e}")
        print("Falling back to tokenless initialization...")
        agent = CodeAgent(
            tools=[sentiment_tool],
            model=InferenceClientModel(
                model_id="Qwen/Qwen2.5-72B-Instruct"
            )
        )
else:
    # Try without explicit token (will use HuggingFace login if available)
    if is_hf_space:
        print("⚠️ Warning: No API token found in HuggingFace Space secrets.")
        print("Please add HF_TOKEN in Space Settings → Secrets")
    else:
        print("⚠️ Warning: No API token found. Attempting to use HuggingFace login...")
        print("If this fails, set one of these environment variables:")
        print("  - HF_TOKEN (HuggingFace token)")
        print("  - HUGGINGFACE_TOKEN (alternative HuggingFace token)")
        print("\nOr login via: huggingface-cli login")
    
    agent = CodeAgent(
        tools=[sentiment_tool],
        model=InferenceClientModel(
            model_id="Qwen/Qwen2.5-72B-Instruct"
        )
    )


def chat_with_agent(message, history, request: gr.Request):
    """
    Wrapper function to use the agent's run method with Gradio.
    The agent is stateful and manages history internally.
    """
    try:
        # Run the agent and get the response
        result = agent.run(message)
        
        # Convert result to string if it's not already
        if isinstance(result, dict):
            # Format dictionary results nicely
            if "polarity" in result and "subjectivity" in result and "assessment" in result:
                formatted_result = f"""Sentiment Analysis Results:
- **Polarity**: {result['polarity']} (range: -1 to 1)
- **Subjectivity**: {result['subjectivity']} (range: 0 to 1)
- **Assessment**: {result['assessment'].upper()}"""
                return formatted_result
            else:
                # Format other dictionaries as JSON
                import json
                return json.dumps(result, indent=2)
        elif isinstance(result, (list, tuple)):
            # Convert lists/tuples to string
            return str(result)
        else:
            # Already a string or other type that can be converted
            return str(result)
    except Exception as e:
        return f"Error: {str(e)}"


# Create and launch the Gradio Chat Interface
demo = gr.ChatInterface(
    fn=chat_with_agent,
    title="MCP Sentiment Analysis Agent",
    description="Chat with an AI agent that can analyze sentiment using the MCP sentiment analysis server.",
    chatbot=gr.Chatbot(layout="bubble", show_copy_button=True, type="messages"),
    type="messages"
)

if __name__ == "__main__":
    demo.launch()