Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| Demo script showing how to use the MCP Sentiment Analysis Server | |
| This script demonstrates various ways to interact with the MCP server: | |
| 1. Direct function calls (local testing) | |
| 2. HTTP requests to the MCP endpoint | |
| 3. Using the smolagents MCP client (if available) | |
| """ | |
| import json | |
| import requests | |
| import sys | |
| from typing import Dict, Any | |
| # Import local functions for testing | |
| try: | |
| from app import analyze_sentiment, get_sentiment_score, classify_emotion, batch_analyze | |
| LOCAL_FUNCTIONS_AVAILABLE = True | |
| except ImportError: | |
| LOCAL_FUNCTIONS_AVAILABLE = False | |
| print("⚠️ Local functions not available. Make sure you're in the docker-mcp-server directory.") | |
| def demo_local_functions(): | |
| """Demonstrate the sentiment analysis functions directly.""" | |
| print("🔧 Testing Local Functions") | |
| print("=" * 40) | |
| if not LOCAL_FUNCTIONS_AVAILABLE: | |
| print("❌ Local functions not available") | |
| return | |
| # Test texts | |
| texts = [ | |
| "I absolutely love this product! It's amazing!", | |
| "This is terrible and I hate it.", | |
| "The weather is okay today.", | |
| "I'm feeling confused about this situation." | |
| ] | |
| for i, text in enumerate(texts, 1): | |
| print(f"\n📝 Test {i}: '{text}'") | |
| # Analyze sentiment | |
| result = analyze_sentiment(text) | |
| data = json.loads(result) | |
| print(f" Sentiment: {data['sentiment']} (polarity: {data['polarity']})") | |
| # Get emotion | |
| emotion_result = classify_emotion(text) | |
| emotion_data = json.loads(emotion_result) | |
| print(f" Emotion: {emotion_data['emotion']} (confidence: {emotion_data['confidence']})") | |
| # Test batch analysis | |
| print(f"\n📊 Batch Analysis:") | |
| batch_text = "\n".join(texts) | |
| batch_result = batch_analyze(batch_text) | |
| batch_data = json.loads(batch_result) | |
| print(f" Total texts: {batch_data['summary']['total_texts']}") | |
| print(f" Positive: {batch_data['summary']['positive']}") | |
| print(f" Negative: {batch_data['summary']['negative']}") | |
| print(f" Neutral: {batch_data['summary']['neutral']}") | |
| print(f" Average polarity: {batch_data['summary']['average_polarity']}") | |
| def demo_http_requests(base_url: str = "http://localhost:7860"): | |
| """Demonstrate HTTP requests to the MCP server.""" | |
| print("\n🌐 Testing HTTP Requests") | |
| print("=" * 40) | |
| # Test if server is running | |
| try: | |
| response = requests.get(base_url, timeout=5) | |
| if response.status_code != 200: | |
| print(f"❌ Server not accessible at {base_url}") | |
| return | |
| except requests.exceptions.RequestException: | |
| print(f"❌ Cannot connect to server at {base_url}") | |
| print(" Make sure the server is running with: docker run -p 7860:7860 mcp-sentiment") | |
| return | |
| print(f"✅ Server accessible at {base_url}") | |
| # Test MCP endpoint | |
| mcp_url = f"{base_url}/gradio_api/mcp/sse" | |
| try: | |
| response = requests.get(mcp_url, timeout=5) | |
| print(f"✅ MCP endpoint accessible at {mcp_url}") | |
| except requests.exceptions.RequestException as e: | |
| print(f"❌ MCP endpoint not accessible: {e}") | |
| def demo_mcp_client(server_url: str = "http://localhost:7860/gradio_api/mcp/sse"): | |
| """Demonstrate using the smolagents MCP client.""" | |
| print("\n🤖 Testing MCP Client") | |
| print("=" * 40) | |
| try: | |
| from smolagents.mcp_client import MCPClient | |
| print(f"Connecting to MCP server at {server_url}...") | |
| with MCPClient({"url": server_url}) as tools: | |
| print(f"✅ Connected! Available tools:") | |
| for tool in tools: | |
| print(f" - {tool.name}: {tool.description}") | |
| # Test a tool | |
| if tools: | |
| print(f"\n🧪 Testing first tool...") | |
| # This would require more specific implementation | |
| # depending on how the MCP client works | |
| except ImportError: | |
| print("❌ smolagents not available") | |
| print(" Install with: pip install smolagents") | |
| except Exception as e: | |
| print(f"❌ MCP client error: {e}") | |
| def demo_gradio_api(base_url: str = "http://localhost:7860"): | |
| """Demonstrate using the Gradio API directly.""" | |
| print("\n🎨 Testing Gradio API") | |
| print("=" * 40) | |
| # This is a simplified example - actual Gradio API usage | |
| # would require knowing the specific endpoint structure | |
| try: | |
| # Test basic connectivity | |
| response = requests.get(f"{base_url}/api/", timeout=5) | |
| if response.status_code == 200: | |
| print("✅ Gradio API accessible") | |
| else: | |
| print(f"⚠️ Gradio API returned status {response.status_code}") | |
| except requests.exceptions.RequestException: | |
| print("❌ Gradio API not accessible") | |
| def main(): | |
| """Run all demos.""" | |
| print("🎭 MCP Sentiment Analysis Server Demo") | |
| print("=" * 50) | |
| # Parse command line arguments | |
| server_url = "http://localhost:7860" | |
| mcp_url = f"{server_url}/gradio_api/mcp/sse" | |
| for arg in sys.argv[1:]: | |
| if arg.startswith("--url="): | |
| server_url = arg.split("=", 1)[1] | |
| mcp_url = f"{server_url}/gradio_api/mcp/sse" | |
| # Run demos | |
| if LOCAL_FUNCTIONS_AVAILABLE: | |
| demo_local_functions() | |
| demo_http_requests(server_url) | |
| demo_gradio_api(server_url) | |
| demo_mcp_client(mcp_url) | |
| print("\n🎉 Demo completed!") | |
| print("\nNext steps:") | |
| print("1. Deploy to Hugging Face Spaces using: python deploy_to_hf.py") | |
| print("2. Connect your MCP clients to the deployed endpoint") | |
| print("3. Use the sentiment analysis tools in your AI applications") | |
| if __name__ == "__main__": | |
| if "--help" in sys.argv or "-h" in sys.argv: | |
| print("Usage: python demo.py [--url=SERVER_URL]") | |
| print("\nOptions:") | |
| print(" --url=URL Server URL (default: http://localhost:7860)") | |
| print(" --help, -h Show this help message") | |
| print("\nExamples:") | |
| print(" python demo.py") | |
| print(" python demo.py --url=https://myspace.hf.space") | |
| sys.exit(0) | |
| main() | |