initial commit

README.md

@@ -0,0 +1,360 @@
+# MCP Sentiment Analysis Server
+
+A complete **Model Context Protocol (MCP) Server** implementation for sentiment analysis with integrated **Gradio Web Interface**. This project provides both AI model integration through JSON-RPC 2.0 protocol and human-friendly web interface for sentiment analysis using TextBlob and Transformers.
+
+## 🚀 Features
+
+### Core Capabilities
+- **Dual Backend Support**: TextBlob (fast) and Transformers (accurate)
+- **MCP Protocol Compliance**: Full JSON-RPC 2.0 implementation
+- **Modern Web Interface**: Responsive Gradio UI with real-time analysis
+- **Batch Processing**: Analyze multiple texts simultaneously
+- **Analysis History**: Track and visualize analysis results
+- **Confidence Scoring**: Detailed confidence metrics for all predictions
+
+### Technical Features
+- **Async Architecture**: Non-blocking request processing
+- **Auto Backend Selection**: Intelligent fallback between analysis engines
+- **Comprehensive Error Handling**: Graceful failure management
+- **Performance Optimization**: Model caching and uvloop integration
+- **Type Safety**: Full type annotations throughout codebase
+
+## 📋 Requirements
+
+- Python 3.8+
+- 4GB+ RAM (for Transformers models)
+- Internet connection (for model downloads)
+
+## 🛠️ Installation
+
+### 1. Clone Repository
+```bash
+git clone <repository-url>
+cd mcp-sentiment
+```
+
+### 2. Install Dependencies
+```bash
+pip install -r requirements.txt
+```
+
+### 3. Download TextBlob Corpora (Optional)
+```bash
+python -c "import nltk; nltk.download('punkt'); nltk.download('brown')"
+```
+
+## 🎯 Quick Start
+
+### Run Tests
+```bash
+python main.py --mode test
+```
+
+### Launch Web Interface
+```bash
+python main.py --mode gradio
+```
+Visit `http://localhost:7860` to access the web interface.
+
+### Start MCP Server
+```bash
+python main.py --mode mcp
+```
+
+### Combined Mode (Default)
+```bash
+python main.py --mode combined
+```
+Runs both MCP server and Gradio interface simultaneously.
+
+## 📖 Usage Examples
+
+### Command Line Options
+
+```bash
+# Different execution modes
+python main.py --mode mcp                    # MCP server only
+python main.py --mode gradio                 # Web interface only
+python main.py --mode combined               # Both services (default)
+python main.py --mode test                   # Run functionality tests
+
+# Gradio customization
+python main.py --mode gradio --port 8080     # Custom port
+python main.py --mode gradio --share         # Enable public sharing
+python main.py --mode gradio --debug         # Debug mode
+
+# Logging control
+python main.py --log-level DEBUG             # Detailed logging
+```
+
+### Python API Usage
+
+```python
+import asyncio
+from src import get_analyzer, analyze_sentiment
+
+async def example():
+    # Simple analysis
+    result = await analyze_sentiment("I love this product!", backend="auto")
+    print(f"Sentiment: {result['label']} (confidence: {result['confidence']:.2%})")
+    
+    # Advanced usage
+    analyzer = await get_analyzer("transformers")
+    result = await analyzer.analyze("This is amazing!")
+    print(f"Result: {result.to_dict()}")
+    
+    # Batch processing
+    texts = ["Great product!", "Terrible service", "It's okay"]
+    results = await analyzer.analyze_batch(texts)
+    for text, result in zip(texts, results):
+        print(f"'{text}' -> {result.label.value}")
+
+asyncio.run(example())
+```
+
+### MCP Tool Integration
+
+The server provides these MCP tools:
+
+#### `analyze_sentiment`
+```json
+{
+  "name": "analyze_sentiment",
+  "arguments": {
+    "text": "I love this new feature!",
+    "backend": "auto"
+  }
+}
+```
+
+#### `analyze_sentiment_batch`
+```json
+{
+  "name": "analyze_sentiment_batch", 
+  "arguments": {
+    "texts": ["Great!", "Terrible!", "Okay."],
+    "backend": "transformers"
+  }
+}
+```
+
+#### `health_check`
+```json
+{
+  "name": "health_check",
+  "arguments": {}
+}
+```
+
+## 🏗️ Architecture
+
+### Project Structure
+```
+mcp-sentiment/
+├── src/
+│   ├── __init__.py              # Package initialization
+│   ├── sentiment_analyzer.py   # Core sentiment analysis engine
+│   ├── mcp_server.py           # JSON-RPC 2.0 MCP server
+│   ├── tools.py                # MCP tool definitions
+│   └── gradio_interface.py     # Web UI implementation
+├── requirements.txt             # Python dependencies
+├── main.py                     # Application entry point
+└── README.md                   # This file
+```
+
+### Component Overview
+
+| Component | Purpose | Technology |
+|-----------|---------|------------|
+| **MCP Server** | AI model communication | JSON-RPC 2.0, asyncio |
+| **Gradio UI** | Human web interface | Gradio framework |
+| **Sentiment Engine** | Text analysis core | TextBlob/Transformers |
+| **Tool Registry** | MCP tool definitions | Python decorators |
+
+## 🔧 Configuration
+
+### Backend Selection
+- **`auto`**: Automatically selects best available backend
+- **`textblob`**: Fast, rule-based sentiment analysis
+- **`transformers`**: AI-powered analysis with higher accuracy
+
+### Environment Variables
+```bash
+# Optional configuration
+export MCP_LOG_LEVEL=INFO
+export GRADIO_SERVER_PORT=7860
+export TRANSFORMERS_CACHE=/path/to/cache
+```
+
+## 📊 Web Interface Features
+
+### Main Analysis Tab
+- Real-time sentiment analysis
+- Confidence visualization with gauge charts
+- Backend selection (TextBlob/Transformers/Auto)
+- Quick example buttons for testing
+
+### Batch Analysis Tab
+- Process multiple texts simultaneously
+- Results table with sentiment and confidence
+- Summary visualization with pie charts
+- Export capabilities
+
+### History Tab
+- Track all analysis results
+- Statistical summaries
+- Sentiment distribution charts
+- Clear/refresh functionality
+
+### Settings & Info Tab
+- Backend status and configuration
+- Model information and availability
+- Usage tips and documentation
+- System health monitoring
+
+## 🧪 Testing
+
+### Run All Tests
+```bash
+python main.py --mode test
+```
+
+### Manual Testing
+```python
+# Test individual components
+python -m src.sentiment_analyzer
+python -m src.tools
+python -m src.mcp_server
+```
+
+### Performance Testing
+```bash
+# Benchmark different backends
+python -c "
+import asyncio
+from src import get_analyzer
+import time
+
+async def benchmark():
+    analyzer = await get_analyzer('textblob')
+    start = time.time()
+    await analyzer.analyze('Test message')
+    print(f'TextBlob: {time.time() - start:.3f}s')
+
+asyncio.run(benchmark())
+"
+```
+
+## 🚨 Troubleshooting
+
+### Common Issues
+
+#### "No sentiment analysis backend available"
+```bash
+pip install textblob
+python -c "import nltk; nltk.download('punkt')"
+```
+
+#### "Failed to load transformer model"
+```bash
+pip install torch transformers
+# Ensure sufficient RAM (4GB+)
+```
+
+#### "Gradio not available"
+```bash
+pip install gradio plotly pandas
+```
+
+#### Port already in use
+```bash
+python main.py --mode gradio --port 8080
+```
+
+### Debug Mode
+```bash
+python main.py --mode combined --debug --log-level DEBUG
+```
+
+## 🔒 Security Considerations
+
+- Input validation and sanitization
+- Rate limiting for API endpoints
+- No persistent data storage by default
+- Configurable access controls
+- Error message sanitization
+
+## 🚀 Performance Optimization
+
+### Model Caching
+- Transformers models cached after first load
+- Shared analyzer instances across requests
+- Memory-efficient batch processing
+
+### Async Processing
+- Non-blocking request handling
+- Concurrent batch analysis
+- Thread pool for CPU-bound operations
+
+### Resource Management
+- Automatic cleanup on shutdown
+- Memory monitoring and limits
+- Configurable worker threads
+
+## 📈 Monitoring & Metrics
+
+### Built-in Metrics
+- Request count and success rate
+- Average response times
+- Error rates by type
+- Backend usage statistics
+
+### Health Checks
+```bash
+curl -X POST http://localhost:7860/health \
+  -H "Content-Type: application/json" \
+  -d '{"method": "health_check", "params": {}}'
+```
+
+## 🤝 Contributing
+
+1. Fork the repository
+2. Create feature branch (`git checkout -b feature/amazing-feature`)
+3. Commit changes (`git commit -m 'Add amazing feature'`)
+4. Push to branch (`git push origin feature/amazing-feature`)
+5. Open Pull Request
+
+### Development Setup
+```bash
+pip install -r requirements.txt
+pip install pytest black flake8 mypy
+```
+
+### Code Quality
+```bash
+black src/                    # Format code
+flake8 src/                   # Lint code
+mypy src/                     # Type checking
+pytest tests/                 # Run tests
+```
+
+## 📄 License
+
+This project is licensed under the MIT License - see the LICENSE file for details.
+
+## 🙏 Acknowledgments
+
+- [Hugging Face Transformers](https://huggingface.co/transformers/) for state-of-the-art NLP models
+- [TextBlob](https://textblob.readthedocs.io/) for simple sentiment analysis
+- [Gradio](https://gradio.app/) for the amazing web interface framework
+- [Model Context Protocol](https://modelcontextprotocol.io/) for the integration standard
+
+## 📞 Support
+
+- **Issues**: [GitHub Issues](https://github.com/your-repo/issues)
+- **Discussions**: [GitHub Discussions](https://github.com/your-repo/discussions)
+- **Documentation**: [Wiki](https://github.com/your-repo/wiki)
+
+---
+
+**Made with ❤️ for the AI community** 

main.py

@@ -0,0 +1,356 @@
+#!/usr/bin/env python3
+"""
+Main entry point for MCP Sentiment Analysis Server.
+
+This script provides multiple modes of operation:
+1. MCP Server mode - JSON-RPC 2.0 server for AI model integration
+2. Gradio Interface mode - Web UI for human interaction
+3. Combined mode - Both MCP server and Gradio interface
+4. Test mode - Run basic functionality tests
+
+Usage:
+    python main.py --mode mcp          # Run MCP server only
+    python main.py --mode gradio       # Run Gradio interface only
+    python main.py --mode combined     # Run both (default)
+    python main.py --mode test         # Run tests
+"""
+
+import asyncio
+import argparse
+import logging
+import sys
+import signal
+import threading
+from typing import Optional
+from concurrent.futures import ThreadPoolExecutor
+
+try:
+    import uvloop
+    UVLOOP_AVAILABLE = True
+except ImportError:
+    UVLOOP_AVAILABLE = False
+
+from src import (
+    create_server,
+    MCPServerRunner,
+    create_gradio_interface,
+    get_analyzer
+)
+
+
+class ApplicationRunner:
+    """
+    Main application runner that manages different execution modes.
+    
+    Supports running MCP server, Gradio interface, or both simultaneously
+    with proper resource management and graceful shutdown.
+    """
+    
+    def __init__(self):
+        """Initialize application runner."""
+        self.logger = logging.getLogger(__name__)
+        self.mcp_server = None
+        self.gradio_interface = None
+        self.running = False
+        self.executor = ThreadPoolExecutor(max_workers=2)
+        
+        # Setup signal handlers
+        signal.signal(signal.SIGINT, self._signal_handler)
+        signal.signal(signal.SIGTERM, self._signal_handler)
+    
+    def _signal_handler(self, signum, frame):
+        """Handle shutdown signals."""
+        self.logger.info(f"Received signal {signum}, shutting down...")
+        self.running = False
+    
+    async def run_mcp_server(self) -> None:
+        """Run MCP server only."""
+        self.logger.info("Starting MCP server mode")
+        
+        try:
+            # Create and run MCP server
+            self.mcp_server = await create_server()
+            runner = MCPServerRunner(self.mcp_server)
+            
+            self.running = True
+            await runner.run()
+            
+        except Exception as e:
+            self.logger.error(f"MCP server failed: {e}")
+            raise
+        finally:
+            if self.mcp_server:
+                await self.mcp_server.stop()
+    
+    def run_gradio_interface(self, **kwargs) -> None:
+        """Run Gradio interface only."""
+        self.logger.info("Starting Gradio interface mode")
+        
+        try:
+            # Create and launch Gradio interface
+            self.gradio_interface = create_gradio_interface()
+            
+            # Default launch parameters
+            launch_params = {
+                "server_name": "0.0.0.0",
+                "server_port": 7860,
+                "share": False,
+                "debug": False,
+                "show_error": True,
+                "quiet": False
+            }
+            launch_params.update(kwargs)
+            
+            self.running = True
+            self.gradio_interface.launch(**launch_params)
+            
+        except Exception as e:
+            self.logger.error(f"Gradio interface failed: {e}")
+            raise
+    
+    async def run_combined(self, **gradio_kwargs) -> None:
+        """Run both MCP server and Gradio interface."""
+        self.logger.info("Starting combined mode (MCP server + Gradio interface)")
+        
+        try:
+            # Create MCP server
+            self.mcp_server = await create_server()
+            
+            # Create Gradio interface
+            self.gradio_interface = create_gradio_interface()
+            
+            # Default Gradio launch parameters
+            launch_params = {
+                "server_name": "0.0.0.0",
+                "server_port": 7860,
+                "share": False,
+                "debug": False,
+                "show_error": True,
+                "quiet": False
+            }
+            launch_params.update(gradio_kwargs)
+            
+            self.running = True
+            
+            # Run Gradio in thread pool
+            gradio_future = self.executor.submit(
+                self.gradio_interface.launch, **launch_params
+            )
+            
+            # Run MCP server in main thread
+            runner = MCPServerRunner(self.mcp_server)
+            
+            # Start both services
+            self.logger.info("Both services starting...")
+            
+            # Wait for either to complete or fail
+            try:
+                await runner.run()
+            except Exception as e:
+                self.logger.error(f"MCP server error: {e}")
+                raise
+            finally:
+                # Cleanup
+                if gradio_future:
+                    gradio_future.cancel()
+                
+        except Exception as e:
+            self.logger.error(f"Combined mode failed: {e}")
+            raise
+        finally:
+            if self.mcp_server:
+                await self.mcp_server.stop()
+    
+    async def run_tests(self) -> bool:
+        """Run basic functionality tests."""
+        self.logger.info("Running functionality tests...")
+        
+        try:
+            # Test 1: Sentiment analyzer initialization
+            self.logger.info("Test 1: Initializing sentiment analyzer...")
+            analyzer = await get_analyzer("textblob")
+            self.logger.info(f"✓ Analyzer initialized with backend: {analyzer.backend}")
+            
+            # Test 2: Basic sentiment analysis
+            self.logger.info("Test 2: Basic sentiment analysis...")
+            test_texts = [
+                "I love this product!",
+                "This is terrible.",
+                "It's okay, nothing special."
+            ]
+            
+            for text in test_texts:
+                result = await analyzer.analyze(text)
+                self.logger.info(f"✓ '{text}' -> {result.label.value} ({result.confidence:.2f})")
+            
+            # Test 3: Batch analysis
+            self.logger.info("Test 3: Batch analysis...")
+            batch_results = await analyzer.analyze_batch(test_texts)
+            self.logger.info(f"✓ Batch analysis completed: {len(batch_results)} results")
+            
+            # Test 4: MCP tools
+            self.logger.info("Test 4: MCP tools...")
+            from src.tools import get_tools
+            tools = get_tools()
+            available_tools = tools.get_tools()
+            self.logger.info(f"✓ {len(available_tools)} MCP tools available")
+            
+            # Test 5: Tool execution
+            self.logger.info("Test 5: Tool execution...")
+            result = await tools.call_tool("analyze_sentiment", {
+                "text": "This is a test message",
+                "backend": "textblob"
+            })
+            self.logger.info(f"✓ Tool execution successful: {result.get('success', False)}")
+            
+            # Test 6: Health check
+            self.logger.info("Test 6: Health check...")
+            health_result = await tools.call_tool("health_check", {})
+            self.logger.info(f"✓ Health check: {health_result.get('status', 'unknown')}")
+            
+            # Cleanup
+            await analyzer.cleanup()
+            
+            self.logger.info("🎉 All tests passed!")
+            return True
+            
+        except Exception as e:
+            self.logger.error(f"❌ Test failed: {e}")
+            return False
+
+
+def setup_logging(level: str = "INFO") -> None:
+    """
+    Setup logging configuration.
+    
+    Args:
+        level: Logging level
+    """
+    logging.basicConfig(
+        level=getattr(logging, level.upper()),
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+        handlers=[
+            logging.StreamHandler(sys.stderr)
+        ]
+    )
+
+
+def parse_arguments() -> argparse.Namespace:
+    """
+    Parse command line arguments.
+    
+    Returns:
+        Parsed arguments
+    """
+    parser = argparse.ArgumentParser(
+        description="MCP Sentiment Analysis Server",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+Examples:
+  python main.py --mode mcp                    # MCP server only
+  python main.py --mode gradio                 # Gradio interface only  
+  python main.py --mode combined               # Both services
+  python main.py --mode test                   # Run tests
+  python main.py --mode gradio --port 8080     # Custom port
+  python main.py --mode gradio --share         # Public sharing
+        """
+    )
+    
+    parser.add_argument(
+        "--mode",
+        choices=["mcp", "gradio", "combined", "test"],
+        default="combined",
+        help="Execution mode (default: combined)"
+    )
+    
+    parser.add_argument(
+        "--log-level",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
+        default="INFO",
+        help="Logging level (default: INFO)"
+    )
+    
+    # Gradio-specific options
+    parser.add_argument(
+        "--port",
+        type=int,
+        default=7860,
+        help="Gradio server port (default: 7860)"
+    )
+    
+    parser.add_argument(
+        "--host",
+        default="0.0.0.0",
+        help="Gradio server host (default: 0.0.0.0)"
+    )
+    
+    parser.add_argument(
+        "--share",
+        action="store_true",
+        help="Enable Gradio public sharing"
+    )
+    
+    parser.add_argument(
+        "--debug",
+        action="store_true",
+        help="Enable debug mode"
+    )
+    
+    return parser.parse_args()
+
+
+async def main() -> None:
+    """Main application entry point."""
+    # Parse arguments
+    args = parse_arguments()
+    
+    # Setup logging
+    setup_logging(args.log_level)
+    logger = logging.getLogger(__name__)
+    
+    # Use uvloop if available for better performance
+    if UVLOOP_AVAILABLE and args.mode in ["mcp", "combined"]:
+        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+        logger.info("Using uvloop for better performance")
+    
+    # Create application runner
+    runner = ApplicationRunner()
+    
+    try:
+        if args.mode == "mcp":
+            await runner.run_mcp_server()
+            
+        elif args.mode == "gradio":
+            # Gradio runs in sync mode
+            gradio_kwargs = {
+                "server_name": args.host,
+                "server_port": args.port,
+                "share": args.share,
+                "debug": args.debug
+            }
+            runner.run_gradio_interface(**gradio_kwargs)
+            
+        elif args.mode == "combined":
+            gradio_kwargs = {
+                "server_name": args.host,
+                "server_port": args.port,
+                "share": args.share,
+                "debug": args.debug
+            }
+            await runner.run_combined(**gradio_kwargs)
+            
+        elif args.mode == "test":
+            success = await runner.run_tests()
+            sys.exit(0 if success else 1)
+            
+    except KeyboardInterrupt:
+        logger.info("Application interrupted by user")
+    except Exception as e:
+        logger.error(f"Application error: {e}")
+        sys.exit(1)
+    finally:
+        logger.info("Application shutdown complete")
+
+
+if __name__ == "__main__":
+    asyncio.run(main()) 

requirements.txt

@@ -0,0 +1,13 @@
+gradio>=4.0.0
+textblob>=0.17.1
+transformers>=4.30.0
+torch>=2.0.0
+mcp>=1.0.0
+asyncio>=3.4.3
+aiohttp>=3.8.0
+pydantic>=2.0.0
+typing-extensions>=4.0.0
+plotly>=5.0.0
+pandas>=2.0.0
+pytest>=7.0.0
+uvloop>=0.17.0 

src/__init__.py

@@ -0,0 +1,60 @@
+"""
+MCP Sentiment Analysis Server Package.
+
+This package provides a complete Model Context Protocol server implementation
+for sentiment analysis with Gradio web interface integration.
+"""
+
+from .sentiment_analyzer import (
+    SentimentAnalyzer,
+    SentimentResult,
+    SentimentLabel,
+    get_analyzer,
+    analyze_sentiment
+)
+
+from .tools import (
+    MCPTools,
+    get_tools,
+    list_tools,
+    call_tool
+)
+
+from .mcp_server import (
+    SentimentMCPServer,
+    MCPServerRunner,
+    create_server
+)
+
+from .gradio_interface import (
+    GradioInterface,
+    create_gradio_interface
+)
+
+__version__ = "1.0.0"
+__author__ = "MCP Sentiment Analysis Team"
+__description__ = "Model Context Protocol server for sentiment analysis with Gradio interface"
+
+__all__ = [
+    # Sentiment Analysis
+    "SentimentAnalyzer",
+    "SentimentResult", 
+    "SentimentLabel",
+    "get_analyzer",
+    "analyze_sentiment",
+    
+    # MCP Tools
+    "MCPTools",
+    "get_tools",
+    "list_tools",
+    "call_tool",
+    
+    # MCP Server
+    "SentimentMCPServer",
+    "MCPServerRunner",
+    "create_server",
+    
+    # Gradio Interface
+    "GradioInterface",
+    "create_gradio_interface",
+] 

src/gradio_interface.py

@@ -0,0 +1,534 @@
+"""
+Gradio web interface for sentiment analysis.
+
+This module provides a modern, responsive web interface using Gradio
+for human interaction with the sentiment analysis system, including
+real-time analysis, confidence visualization, and history tracking.
+"""
+
+import asyncio
+import logging
+import json
+from typing import Dict, Any, List, Tuple, Optional
+from datetime import datetime
+import pandas as pd
+import plotly.graph_objects as go
+import plotly.express as px
+
+try:
+    import gradio as gr
+    GRADIO_AVAILABLE = True
+except ImportError:
+    GRADIO_AVAILABLE = False
+    logging.error("Gradio not available. Install with: pip install gradio")
+
+from .sentiment_analyzer import get_analyzer, SentimentResult, SentimentLabel
+
+
+class SentimentHistory:
+    """Manages sentiment analysis history."""
+    
+    def __init__(self, max_entries: int = 100):
+        self.max_entries = max_entries
+        self.entries: List[Dict[str, Any]] = []
+        self.logger = logging.getLogger(__name__)
+    
+    def add_entry(self, text: str, result: SentimentResult, backend: str) -> None:
+        entry = {
+            "timestamp": datetime.now().isoformat(),
+            "text": text[:100] + "..." if len(text) > 100 else text,
+            "full_text": text,
+            "label": result.label.value,
+            "confidence": result.confidence,
+            "backend": backend,
+            "raw_scores": result.raw_scores
+        }
+        
+        self.entries.append(entry)
+        
+        if len(self.entries) > self.max_entries:
+            self.entries = self.entries[-self.max_entries:]
+    
+    def get_recent_entries(self, count: int = 10) -> List[Dict[str, Any]]:
+        return self.entries[-count:] if self.entries else []
+    
+    def get_statistics(self) -> Dict[str, Any]:
+        if not self.entries:
+            return {
+                "total_analyses": 0,
+                "label_distribution": {},
+                "average_confidence": 0.0,
+                "backend_usage": {}
+            }
+        
+        labels = [entry["label"] for entry in self.entries]
+        confidences = [entry["confidence"] for entry in self.entries]
+        backends = [entry["backend"] for entry in self.entries]
+        
+        label_counts = {
+            "positive": labels.count("positive"),
+            "negative": labels.count("negative"),
+            "neutral": labels.count("neutral")
+        }
+        
+        backend_counts = {}
+        for backend in backends:
+            backend_counts[backend] = backend_counts.get(backend, 0) + 1
+        
+        return {
+            "total_analyses": len(self.entries),
+            "label_distribution": label_counts,
+            "average_confidence": sum(confidences) / len(confidences),
+            "backend_usage": backend_counts
+        }
+
+
+class GradioInterface:
+    """Gradio web interface for sentiment analysis."""
+    
+    def __init__(self, title: str = "Sentiment Analysis Server", 
+                 description: str = "Analyze text sentiment using TextBlob or Transformers"):
+        self.title = title
+        self.description = description
+        self.logger = logging.getLogger(__name__)
+        self.history = SentimentHistory()
+        self.interface = None
+        self._setup_interface()
+    
+    def _setup_interface(self) -> None:
+        if not GRADIO_AVAILABLE:
+            raise RuntimeError("Gradio not available")
+        
+        with gr.Blocks(
+            theme=gr.themes.Soft(),
+            title=self.title
+        ) as interface:
+            
+            gr.Markdown(f"# {self.title}")
+            gr.Markdown(f"*{self.description}*")
+            
+            with gr.Tabs():
+                with gr.TabItem("Sentiment Analysis"):
+                    with gr.Row():
+                        with gr.Column(scale=2):
+                            text_input = gr.Textbox(
+                                label="Text to Analyze",
+                                placeholder="Enter text here to analyze its sentiment...",
+                                lines=4
+                            )
+                            
+                            with gr.Row():
+                                backend_choice = gr.Dropdown(
+                                    choices=["auto", "textblob", "transformers"],
+                                    value="auto",
+                                    label="Analysis Backend"
+                                )
+                                
+                                analyze_btn = gr.Button(
+                                    "Analyze Sentiment",
+                                    variant="primary"
+                                )
+                        
+                        with gr.Column(scale=1):
+                            result_display = gr.HTML(
+                                value="<p>Enter text and click 'Analyze Sentiment' to see results.</p>"
+                            )
+                            
+                            confidence_plot = gr.Plot(visible=False)
+                    
+                    gr.Markdown("### Quick Examples")
+                    with gr.Row():
+                        pos_btn = gr.Button("😊 Positive", size="sm")
+                        neu_btn = gr.Button("😐 Neutral", size="sm")
+                        neg_btn = gr.Button("😞 Negative", size="sm")
+                        mix_btn = gr.Button("📝 Mixed", size="sm")
+                
+                with gr.TabItem("Batch Analysis"):
+                    with gr.Row():
+                        with gr.Column():
+                            batch_input = gr.Textbox(
+                                label="Texts to Analyze (one per line)",
+                                placeholder="Enter multiple texts, one per line...",
+                                lines=8
+                            )
+                            
+                            with gr.Row():
+                                batch_backend = gr.Dropdown(
+                                    choices=["auto", "textblob", "transformers"],
+                                    value="auto",
+                                    label="Analysis Backend"
+                                )
+                                
+                                batch_analyze_btn = gr.Button(
+                                    "Analyze Batch",
+                                    variant="primary"
+                                )
+                        
+                        with gr.Column():
+                            batch_results = gr.DataFrame(
+                                label="Batch Results",
+                                headers=["Text", "Sentiment", "Confidence"]
+                            )
+                            
+                            batch_summary_plot = gr.Plot(visible=False)
+                
+                with gr.TabItem("Analysis History"):
+                    with gr.Row():
+                        refresh_history_btn = gr.Button("Refresh History", variant="secondary")
+                        clear_history_btn = gr.Button("Clear History", variant="stop")
+                    
+                    with gr.Row():
+                        with gr.Column(scale=2):
+                            history_table = gr.DataFrame(
+                                label="Recent Analyses",
+                                headers=["Time", "Text", "Sentiment", "Confidence", "Backend"]
+                            )
+                        
+                        with gr.Column(scale=1):
+                            stats_display = gr.HTML(value="<p>No analyses yet.</p>")
+                            history_plot = gr.Plot(visible=False)
+                
+                with gr.TabItem("Settings & Info"):
+                    with gr.Row():
+                        with gr.Column():
+                            gr.Markdown("### Backend Information")
+                            backend_info = gr.HTML(value="<p>Loading backend information...</p>")
+                            refresh_info_btn = gr.Button("Refresh Info", variant="secondary")
+                        
+                        with gr.Column():
+                            gr.Markdown("### Usage Tips")
+                            gr.Markdown("""
+                            - **Auto**: Automatically selects the best available backend
+                            - **TextBlob**: Fast, simple sentiment analysis
+                            - **Transformers**: More accurate, AI-powered analysis
+                            - **Batch Analysis**: Process multiple texts at once
+                            - **History**: Track your analysis results over time
+                            """)
+            
+            # Event handlers
+            def analyze_sentiment(text: str, backend: str) -> Tuple[str, gr.Plot]:
+                return asyncio.run(self._analyze_sentiment_async(text, backend))
+            
+            def analyze_batch(texts: str, backend: str) -> Tuple[pd.DataFrame, gr.Plot]:
+                return asyncio.run(self._analyze_batch_async(texts, backend))
+            
+            def refresh_history() -> Tuple[pd.DataFrame, str, gr.Plot]:
+                return self._get_history_data()
+            
+            def clear_history() -> Tuple[pd.DataFrame, str, gr.Plot]:
+                self.history.entries.clear()
+                return self._get_history_data()
+            
+            def get_backend_info() -> str:
+                return asyncio.run(self._get_backend_info_async())
+            
+            # Example texts
+            examples = [
+                "I absolutely love this new feature! It's incredible and makes everything so much easier.",
+                "The weather is okay today, nothing particularly special about it.",
+                "This is terrible and frustrating. I hate how complicated this has become.",
+                "The movie had great visuals but the plot was disappointing. Mixed feelings overall."
+            ]
+            
+            # Wire up events
+            analyze_btn.click(
+                analyze_sentiment,
+                inputs=[text_input, backend_choice],
+                outputs=[result_display, confidence_plot]
+            )
+            
+            batch_analyze_btn.click(
+                analyze_batch,
+                inputs=[batch_input, batch_backend],
+                outputs=[batch_results, batch_summary_plot]
+            )
+            
+            refresh_history_btn.click(
+                refresh_history,
+                outputs=[history_table, stats_display, history_plot]
+            )
+            
+            clear_history_btn.click(
+                clear_history,
+                outputs=[history_table, stats_display, history_plot]
+            )
+            
+            refresh_info_btn.click(
+                get_backend_info,
+                outputs=[backend_info]
+            )
+            
+            # Example buttons
+            pos_btn.click(lambda: examples[0], outputs=[text_input])
+            neu_btn.click(lambda: examples[1], outputs=[text_input])
+            neg_btn.click(lambda: examples[2], outputs=[text_input])
+            mix_btn.click(lambda: examples[3], outputs=[text_input])
+            
+            # Load initial data
+            interface.load(get_backend_info, outputs=[backend_info])
+            interface.load(refresh_history, outputs=[history_table, stats_display, history_plot])
+        
+        self.interface = interface
+    
+    async def _analyze_sentiment_async(self, text: str, backend: str) -> Tuple[str, gr.Plot]:
+        try:
+            if not text.strip():
+                return "<p>Please enter some text to analyze.</p>", gr.Plot(visible=False)
+            
+            analyzer = await get_analyzer(backend)
+            result = await analyzer.analyze(text)
+            
+            self.history.add_entry(text, result, analyzer.backend)
+            
+            sentiment_class = f"sentiment-{result.label.value}"
+            confidence_class = (
+                "confidence-high" if result.confidence > 0.7 
+                else "confidence-medium" if result.confidence > 0.4 
+                else "confidence-low"
+            )
+            
+            html_result = f"""
+            <div style="padding: 1rem; border-radius: 0.5rem; background: #f8fafc; border-left: 4px solid #3b82f6;">
+                <h3>Analysis Result</h3>
+                <p><strong>Sentiment:</strong> <span style="color: {'#22c55e' if result.label.value == 'positive' else '#ef4444' if result.label.value == 'negative' else '#6b7280'}; font-weight: bold;">{result.label.value.title()}</span></p>
+                <p><strong>Confidence:</strong> <span style="color: {'#059669' if result.confidence > 0.7 else '#d97706' if result.confidence > 0.4 else '#dc2626'};">{result.confidence:.2%}</span></p>
+                <p><strong>Backend:</strong> {analyzer.backend}</p>
+                <p><strong>Text Length:</strong> {len(text)} characters</p>
+            </div>
+            """
+            
+            plot = self._create_confidence_plot(result)
+            return html_result, plot
+            
+        except Exception as e:
+            self.logger.error(f"Analysis failed: {e}")
+            error_html = f"""
+            <div style="padding: 1rem; border-radius: 0.5rem; background: #fef2f2; border-left: 4px solid #ef4444;">
+                <h3>Analysis Error</h3>
+                <p><strong>Error:</strong> {str(e)}</p>
+                <p>Please try again or check your input.</p>
+            </div>
+            """
+            return error_html, gr.Plot(visible=False)
+    
+    async def _analyze_batch_async(self, texts: str, backend: str) -> Tuple[pd.DataFrame, gr.Plot]:
+        try:
+            if not texts.strip():
+                return pd.DataFrame(), gr.Plot(visible=False)
+            
+            text_list = [t.strip() for t in texts.split('\n') if t.strip()]
+            
+            if not text_list:
+                return pd.DataFrame(), gr.Plot(visible=False)
+            
+            analyzer = await get_analyzer(backend)
+            results = await analyzer.analyze_batch(text_list)
+            
+            data = []
+            for text, result in zip(text_list, results):
+                self.history.add_entry(text, result, analyzer.backend)
+                
+                data.append({
+                    "Text": text[:50] + "..." if len(text) > 50 else text,
+                    "Sentiment": result.label.value.title(),
+                    "Confidence": f"{result.confidence:.2%}"
+                })
+            
+            df = pd.DataFrame(data)
+            plot = self._create_batch_summary_plot(results)
+            
+            return df, plot
+            
+        except Exception as e:
+            self.logger.error(f"Batch analysis failed: {e}")
+            return pd.DataFrame([{"Error": str(e)}]), gr.Plot(visible=False)
+    
+    def _create_confidence_plot(self, result: SentimentResult) -> gr.Plot:
+        try:
+            fig = go.Figure(go.Indicator(
+                mode="gauge+number",
+                value=result.confidence * 100,
+                domain={'x': [0, 1], 'y': [0, 1]},
+                title={'text': f"Confidence - {result.label.value.title()}"},
+                gauge={
+                    'axis': {'range': [None, 100]},
+                    'bar': {'color': "darkblue"},
+                    'steps': [
+                        {'range': [0, 40], 'color': "lightgray"},
+                        {'range': [40, 70], 'color': "yellow"},
+                        {'range': [70, 100], 'color': "green"}
+                    ]
+                }
+            ))
+            
+            fig.update_layout(height=300, margin=dict(l=20, r=20, t=40, b=20))
+            return gr.Plot(value=fig, visible=True)
+            
+        except Exception as e:
+            self.logger.error(f"Failed to create confidence plot: {e}")
+            return gr.Plot(visible=False)
+    
+    def _create_batch_summary_plot(self, results: List[SentimentResult]) -> gr.Plot:
+        try:
+            labels = [result.label.value for result in results]
+            label_counts = {
+                "Positive": labels.count("positive"),
+                "Negative": labels.count("negative"),
+                "Neutral": labels.count("neutral")
+            }
+            
+            fig = px.pie(
+                values=list(label_counts.values()),
+                names=list(label_counts.keys()),
+                title="Sentiment Distribution",
+                color_discrete_map={
+                    "Positive": "#22c55e",
+                    "Negative": "#ef4444",
+                    "Neutral": "#6b7280"
+                }
+            )
+            
+            fig.update_layout(height=300, margin=dict(l=20, r=20, t=40, b=20))
+            return gr.Plot(value=fig, visible=True)
+            
+        except Exception as e:
+            self.logger.error(f"Failed to create batch summary plot: {e}")
+            return gr.Plot(visible=False)
+    
+    def _get_history_data(self) -> Tuple[pd.DataFrame, str, gr.Plot]:
+        try:
+            entries = self.history.get_recent_entries(20)
+            
+            if not entries:
+                empty_df = pd.DataFrame(columns=["Time", "Text", "Sentiment", "Confidence", "Backend"])
+                return empty_df, "<p>No analyses yet.</p>", gr.Plot(visible=False)
+            
+            data = []
+            for entry in reversed(entries):
+                data.append({
+                    "Time": entry["timestamp"][:19].replace("T", " "),
+                    "Text": entry["text"],
+                    "Sentiment": entry["label"].title(),
+                    "Confidence": f"{entry['confidence']:.2%}",
+                    "Backend": entry["backend"]
+                })
+            
+            df = pd.DataFrame(data)
+            stats = self.history.get_statistics()
+            
+            stats_html = f"""
+            <div style="background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; padding: 1rem; border-radius: 0.5rem;">
+                <h3>📊 Analysis Statistics</h3>
+                <p><strong>Total Analyses:</strong> {stats['total_analyses']}</p>
+                <p><strong>Average Confidence:</strong> {stats['average_confidence']:.2%}</p>
+                <h4>Sentiment Distribution:</h4>
+                <ul>
+                    <li>Positive: {stats['label_distribution'].get('positive', 0)}</li>
+                    <li>Negative: {stats['label_distribution'].get('negative', 0)}</li>
+                    <li>Neutral: {stats['label_distribution'].get('neutral', 0)}</li>
+                </ul>
+            </div>
+            """
+            
+            plot = self._create_history_plot(stats) if stats['total_analyses'] > 0 else gr.Plot(visible=False)
+            return df, stats_html, plot
+            
+        except Exception as e:
+            self.logger.error(f"Failed to get history data: {e}")
+            error_df = pd.DataFrame([{"Error": str(e)}])
+            return error_df, f"<p>Error loading history: {e}</p>", gr.Plot(visible=False)
+    
+    def _create_history_plot(self, stats: Dict[str, Any]) -> gr.Plot:
+        try:
+            labels = list(stats['label_distribution'].keys())
+            values = list(stats['label_distribution'].values())
+            
+            fig = px.bar(
+                x=[label.title() for label in labels],
+                y=values,
+                title="Historical Sentiment Distribution",
+                color=labels,
+                color_discrete_map={
+                    "positive": "#22c55e",
+                    "negative": "#ef4444",
+                    "neutral": "#6b7280"
+                }
+            )
+            
+            fig.update_layout(height=300, margin=dict(l=20, r=20, t=40, b=20), showlegend=False)
+            return gr.Plot(value=fig, visible=True)
+            
+        except Exception as e:
+            self.logger.error(f"Failed to create history plot: {e}")
+            return gr.Plot(visible=False)
+    
+    async def _get_backend_info_async(self) -> str:
+        try:
+            analyzer = await get_analyzer("auto")
+            info = analyzer.get_info()
+            
+            html = f"""
+            <div style="padding: 1rem; border-radius: 0.5rem; background: #f0f9ff; border-left: 4px solid #0ea5e9;">
+                <h3>🔧 Backend Information</h3>
+                <p><strong>Current Backend:</strong> {info['backend']}</p>
+                <p><strong>Model Loaded:</strong> {'Yes' if info['model_loaded'] else 'No'}</p>
+                <p><strong>TextBlob Available:</strong> {'Yes' if info['textblob_available'] else 'No'}</p>
+                <p><strong>Transformers Available:</strong> {'Yes' if info['transformers_available'] else 'No'}</p>
+                <p><strong>CUDA Available:</strong> {'Yes' if info.get('cuda_available', False) else 'No'}</p>
+                {f"<p><strong>Model Name:</strong> {info['model_name']}</p>" if info.get('model_name') else ""}
+            </div>
+            """
+            return html
+            
+        except Exception as e:
+            self.logger.error(f"Failed to get backend info: {e}")
+            return f"""
+            <div style="padding: 1rem; border-radius: 0.5rem; background: #fef2f2; border-left: 4px solid #ef4444;">
+                <h3>❌ Backend Error</h3>
+                <p>Failed to load backend information: {str(e)}</p>
+            </div>
+            """
+    
+    def launch(self, **kwargs) -> None:
+        if not self.interface:
+            raise RuntimeError("Interface not initialized")
+        
+        launch_params = {
+            "server_name": "0.0.0.0",
+            "server_port": 7860,
+            "share": False,
+            "debug": False,
+            "show_error": True,
+            "quiet": False
+        }
+        
+        launch_params.update(kwargs)
+        
+        self.logger.info(f"Launching Gradio interface on {launch_params['server_name']}:{launch_params['server_port']}")
+        
+        try:
+            self.interface.launch(**launch_params)
+        except Exception as e:
+            self.logger.error(f"Failed to launch interface: {e}")
+            raise
+
+
+def create_gradio_interface(**kwargs) -> GradioInterface:
+    if not GRADIO_AVAILABLE:
+        raise RuntimeError("Gradio not available. Install with: pip install gradio")
+    
+    return GradioInterface(**kwargs)
+
+
+async def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    )
+    
+    interface = create_gradio_interface()
+    interface.launch(debug=True)
+
+
+if __name__ == "__main__":
+    asyncio.run(main()) 

src/mcp_server.py

@@ -0,0 +1,572 @@
+"""
+MCP Server implementation for sentiment analysis.
+
+This module implements a Model Context Protocol server that provides
+sentiment analysis capabilities through JSON-RPC 2.0 protocol with
+async request handling and comprehensive error management.
+"""
+
+import asyncio
+import json
+import logging
+import sys
+import traceback
+from typing import Dict, Any, List, Optional, Union
+from dataclasses import dataclass
+from enum import Enum
+import uuid
+from datetime import datetime
+
+try:
+    import uvloop
+    UVLOOP_AVAILABLE = True
+except ImportError:
+    UVLOOP_AVAILABLE = False
+
+from .tools import get_tools, list_tools, call_tool
+from .sentiment_analyzer import get_analyzer
+
+
+class MCPMessageType(Enum):
+    """MCP message types."""
+    REQUEST = "request"
+    RESPONSE = "response"
+    NOTIFICATION = "notification"
+
+
+@dataclass
+class MCPRequest:
+    """MCP request message structure."""
+    jsonrpc: str
+    method: str
+    params: Optional[Dict[str, Any]] = None
+    id: Optional[Union[str, int]] = None
+
+
+@dataclass
+class MCPResponse:
+    """MCP response message structure."""
+    jsonrpc: str
+    id: Optional[Union[str, int]]
+    result: Optional[Dict[str, Any]] = None
+    error: Optional[Dict[str, Any]] = None
+
+
+@dataclass
+class MCPError:
+    """MCP error structure."""
+    code: int
+    message: str
+    data: Optional[Dict[str, Any]] = None
+
+
+class MCPErrorCodes:
+    """Standard JSON-RPC 2.0 error codes."""
+    PARSE_ERROR = -32700
+    INVALID_REQUEST = -32600
+    METHOD_NOT_FOUND = -32601
+    INVALID_PARAMS = -32602
+    INTERNAL_ERROR = -32603
+    
+    # Custom error codes
+    TOOL_ERROR = -32000
+    ANALYZER_ERROR = -32001
+    VALIDATION_ERROR = -32002
+
+
+class SentimentMCPServer:
+    """
+    Model Context Protocol server for sentiment analysis.
+    
+    Implements JSON-RPC 2.0 protocol with async request handling,
+    tool registration, and comprehensive error management.
+    """
+    
+    def __init__(self, name: str = "sentiment-analyzer", version: str = "1.0.0"):
+        """
+        Initialize MCP server.
+        
+        Args:
+            name: Server name
+            version: Server version
+        """
+        self.name = name
+        self.version = version
+        self.logger = logging.getLogger(__name__)
+        
+        # Server state
+        self.running = False
+        self.request_count = 0
+        self.error_count = 0
+        self.start_time = None
+        
+        # Request handlers
+        self._handlers = {
+            "initialize": self._handle_initialize,
+            "tools/list": self._handle_list_tools,
+            "tools/call": self._handle_call_tool,
+            "ping": self._handle_ping,
+            "server/info": self._handle_server_info,
+            "server/stats": self._handle_server_stats
+        }
+        
+        self.logger.info(f"Initialized MCP server '{name}' v{version}")
+    
+    async def start(self) -> None:
+        """Start the MCP server."""
+        self.running = True
+        self.start_time = datetime.now()
+        self.logger.info(f"MCP server '{self.name}' started")
+        
+        # Pre-load analyzer to improve first request performance
+        try:
+            await get_analyzer("auto")
+            self.logger.info("Sentiment analyzer pre-loaded successfully")
+        except Exception as e:
+            self.logger.warning(f"Failed to pre-load analyzer: {e}")
+    
+    async def stop(self) -> None:
+        """Stop the MCP server."""
+        self.running = False
+        self.logger.info(f"MCP server '{self.name}' stopped")
+        
+        # Cleanup resources
+        try:
+            analyzer = await get_analyzer("auto")
+            await analyzer.cleanup()
+        except Exception as e:
+            self.logger.error(f"Error during cleanup: {e}")
+    
+    def _create_error_response(self, request_id: Optional[Union[str, int]], 
+                              code: int, message: str, 
+                              data: Optional[Dict[str, Any]] = None) -> MCPResponse:
+        """
+        Create error response.
+        
+        Args:
+            request_id: Request ID
+            code: Error code
+            message: Error message
+            data: Additional error data
+            
+        Returns:
+            MCPResponse with error
+        """
+        error = {
+            "code": code,
+            "message": message
+        }
+        if data:
+            error["data"] = data
+        
+        return MCPResponse(
+            jsonrpc="2.0",
+            id=request_id,
+            error=error
+        )
+    
+    def _create_success_response(self, request_id: Optional[Union[str, int]], 
+                               result: Dict[str, Any]) -> MCPResponse:
+        """
+        Create success response.
+        
+        Args:
+            request_id: Request ID
+            result: Response result
+            
+        Returns:
+            MCPResponse with result
+        """
+        return MCPResponse(
+            jsonrpc="2.0",
+            id=request_id,
+            result=result
+        )
+    
+    def _parse_request(self, message: str) -> MCPRequest:
+        """
+        Parse JSON-RPC request message.
+        
+        Args:
+            message: JSON message string
+            
+        Returns:
+            Parsed MCPRequest
+            
+        Raises:
+            ValueError: If parsing fails
+        """
+        try:
+            data = json.loads(message)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Invalid JSON: {e}")
+        
+        # Validate required fields
+        if not isinstance(data, dict):
+            raise ValueError("Request must be a JSON object")
+        
+        if data.get("jsonrpc") != "2.0":
+            raise ValueError("Invalid JSON-RPC version")
+        
+        if "method" not in data:
+            raise ValueError("Missing 'method' field")
+        
+        return MCPRequest(
+            jsonrpc=data["jsonrpc"],
+            method=data["method"],
+            params=data.get("params"),
+            id=data.get("id")
+        )
+    
+    async def process_request(self, message: str) -> str:
+        """
+        Process incoming JSON-RPC request.
+        
+        Args:
+            message: JSON-RPC request message
+            
+        Returns:
+            JSON-RPC response message
+        """
+        request_id = None
+        
+        try:
+            # Parse request
+            try:
+                request = self._parse_request(message)
+                request_id = request.id
+            except ValueError as e:
+                response = self._create_error_response(
+                    None, MCPErrorCodes.PARSE_ERROR, str(e)
+                )
+                return json.dumps(response.__dict__)
+            
+            # Update stats
+            self.request_count += 1
+            
+            # Log request
+            self.logger.debug(f"Processing request: {request.method} (ID: {request_id})")
+            
+            # Handle request
+            if request.method in self._handlers:
+                handler = self._handlers[request.method]
+                result = await handler(request.params or {})
+                response = self._create_success_response(request_id, result)
+            else:
+                response = self._create_error_response(
+                    request_id, 
+                    MCPErrorCodes.METHOD_NOT_FOUND,
+                    f"Method '{request.method}' not found"
+                )
+            
+        except Exception as e:
+            self.error_count += 1
+            self.logger.error(f"Request processing failed: {e}")
+            self.logger.debug(traceback.format_exc())
+            
+            response = self._create_error_response(
+                request_id,
+                MCPErrorCodes.INTERNAL_ERROR,
+                "Internal server error",
+                {"error": str(e), "type": type(e).__name__}
+            )
+        
+        # Convert response to JSON
+        response_dict = response.__dict__
+        # Remove None values
+        response_dict = {k: v for k, v in response_dict.items() if v is not None}
+        
+        return json.dumps(response_dict)
+    
+    async def _handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle initialize request.
+        
+        Args:
+            params: Initialize parameters
+            
+        Returns:
+            Server capabilities
+        """
+        client_info = params.get("clientInfo", {})
+        self.logger.info(f"Client connected: {client_info}")
+        
+        return {
+            "protocolVersion": "2024-11-05",
+            "capabilities": {
+                "tools": {
+                    "listChanged": False
+                },
+                "logging": {},
+                "prompts": {
+                    "listChanged": False
+                },
+                "resources": {
+                    "subscribe": False,
+                    "listChanged": False
+                }
+            },
+            "serverInfo": {
+                "name": self.name,
+                "version": self.version,
+                "description": "Sentiment analysis server using TextBlob and Transformers"
+            }
+        }
+    
+    async def _handle_list_tools(self, params: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle tools/list request.
+        
+        Args:
+            params: List tools parameters
+            
+        Returns:
+            Available tools
+        """
+        try:
+            tools = await list_tools()
+            return {"tools": tools}
+        except Exception as e:
+            raise RuntimeError(f"Failed to list tools: {e}")
+    
+    async def _handle_call_tool(self, params: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle tools/call request.
+        
+        Args:
+            params: Tool call parameters
+            
+        Returns:
+            Tool execution result
+        """
+        try:
+            name = params.get("name")
+            arguments = params.get("arguments", {})
+            
+            if not name:
+                raise ValueError("Tool name is required")
+            
+            result = await call_tool(name, arguments)
+            
+            return {
+                "content": [
+                    {
+                        "type": "text",
+                        "text": json.dumps(result, indent=2)
+                    }
+                ],
+                "isError": not result.get("success", True)
+            }
+            
+        except Exception as e:
+            self.logger.error(f"Tool call failed: {e}")
+            return {
+                "content": [
+                    {
+                        "type": "text",
+                        "text": json.dumps({
+                            "success": False,
+                            "error": str(e),
+                            "error_type": type(e).__name__
+                        }, indent=2)
+                    }
+                ],
+                "isError": True
+            }
+    
+    async def _handle_ping(self, params: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle ping request.
+        
+        Args:
+            params: Ping parameters
+            
+        Returns:
+            Pong response
+        """
+        return {
+            "pong": True,
+            "timestamp": datetime.now().isoformat(),
+            "server": self.name,
+            "version": self.version
+        }
+    
+    async def _handle_server_info(self, params: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle server/info request.
+        
+        Args:
+            params: Server info parameters
+            
+        Returns:
+            Server information
+        """
+        try:
+            analyzer = await get_analyzer("auto")
+            analyzer_info = analyzer.get_info()
+        except Exception as e:
+            analyzer_info = {"error": str(e)}
+        
+        return {
+            "server": {
+                "name": self.name,
+                "version": self.version,
+                "running": self.running,
+                "start_time": self.start_time.isoformat() if self.start_time else None
+            },
+            "analyzer": analyzer_info,
+            "capabilities": {
+                "sentiment_analysis": True,
+                "batch_processing": True,
+                "multiple_backends": True,
+                "async_processing": True
+            }
+        }
+    
+    async def _handle_server_stats(self, params: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle server/stats request.
+        
+        Args:
+            params: Server stats parameters
+            
+        Returns:
+            Server statistics
+        """
+        uptime = None
+        if self.start_time:
+            uptime = (datetime.now() - self.start_time).total_seconds()
+        
+        return {
+            "requests_processed": self.request_count,
+            "errors_encountered": self.error_count,
+            "success_rate": (
+                (self.request_count - self.error_count) / self.request_count 
+                if self.request_count > 0 else 0
+            ),
+            "uptime_seconds": uptime,
+            "running": self.running
+        }
+
+
+class MCPServerRunner:
+    """
+    Runner for MCP server with stdio communication.
+    
+    Handles stdin/stdout communication for MCP protocol.
+    """
+    
+    def __init__(self, server: SentimentMCPServer):
+        """
+        Initialize server runner.
+        
+        Args:
+            server: MCP server instance
+        """
+        self.server = server
+        self.logger = logging.getLogger(__name__)
+    
+    async def run(self) -> None:
+        """Run the MCP server with stdio communication."""
+        self.logger.info("Starting MCP server with stdio communication")
+        
+        # Start server
+        await self.server.start()
+        
+        try:
+            # Set up event loop optimization
+            if UVLOOP_AVAILABLE:
+                self.logger.info("Using uvloop for better performance")
+            
+            # Handle stdin/stdout communication
+            reader = asyncio.StreamReader()
+            protocol = asyncio.StreamReaderProtocol(reader)
+            await asyncio.get_event_loop().connect_read_pipe(
+                lambda: protocol, sys.stdin
+            )
+            
+            writer_transport, writer_protocol = await asyncio.get_event_loop().connect_write_pipe(
+                asyncio.streams.FlowControlMixin, sys.stdout
+            )
+            writer = asyncio.StreamWriter(writer_transport, writer_protocol, reader, asyncio.get_event_loop())
+            
+            self.logger.info("MCP server ready for requests")
+            
+            # Process requests
+            while self.server.running:
+                try:
+                    # Read request
+                    line = await reader.readline()
+                    if not line:
+                        break
+                    
+                    message = line.decode().strip()
+                    if not message:
+                        continue
+                    
+                    # Process request
+                    response = await self.server.process_request(message)
+                    
+                    # Send response
+                    writer.write((response + '\n').encode())
+                    await writer.drain()
+                    
+                except asyncio.CancelledError:
+                    break
+                except Exception as e:
+                    self.logger.error(f"Communication error: {e}")
+                    break
+        
+        finally:
+            await self.server.stop()
+            self.logger.info("MCP server stopped")
+
+
+async def create_server(name: str = "sentiment-analyzer", 
+                       version: str = "1.0.0") -> SentimentMCPServer:
+    """
+    Create and configure MCP server.
+    
+    Args:
+        name: Server name
+        version: Server version
+        
+    Returns:
+        Configured MCP server
+    """
+    # Configure logging
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+        handlers=[
+            logging.StreamHandler(sys.stderr)  # Use stderr to avoid interfering with stdio protocol
+        ]
+    )
+    
+    # Create server
+    server = SentimentMCPServer(name, version)
+    
+    return server
+
+
+async def main() -> None:
+    """Main entry point for MCP server."""
+    # Use uvloop if available for better performance
+    if UVLOOP_AVAILABLE:
+        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+    
+    # Create and run server
+    server = await create_server()
+    runner = MCPServerRunner(server)
+    
+    try:
+        await runner.run()
+    except KeyboardInterrupt:
+        logging.info("Server interrupted by user")
+    except Exception as e:
+        logging.error(f"Server error: {e}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main()) 

src/sentiment_analyzer.py

@@ -0,0 +1,383 @@
+"""
+Core sentiment analysis engine for MCP server.
+
+This module provides sentiment analysis functionality using both TextBlob
+for simplicity and Transformers for accuracy, with confidence scoring
+and comprehensive error handling.
+"""
+
+import logging
+from typing import Dict, Any, Optional, Tuple
+from enum import Enum
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+
+try:
+    from textblob import TextBlob
+    TEXTBLOB_AVAILABLE = True
+except ImportError:
+    TEXTBLOB_AVAILABLE = False
+    logging.warning("TextBlob not available. Install with: pip install textblob")
+
+try:
+    from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
+    import torch
+    TRANSFORMERS_AVAILABLE = True
+except ImportError:
+    TRANSFORMERS_AVAILABLE = False
+    logging.warning("Transformers not available. Install with: pip install transformers torch")
+
+
+class SentimentLabel(Enum):
+    """Sentiment classification labels."""
+    POSITIVE = "positive"
+    NEGATIVE = "negative"
+    NEUTRAL = "neutral"
+
+
+class SentimentResult:
+    """Container for sentiment analysis results."""
+    
+    def __init__(self, label: SentimentLabel, confidence: float, raw_scores: Optional[Dict[str, float]] = None):
+        self.label = label
+        self.confidence = confidence
+        self.raw_scores = raw_scores or {}
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert result to dictionary format."""
+        return {
+            "label": self.label.value,
+            "confidence": round(self.confidence, 4),
+            "raw_scores": self.raw_scores
+        }
+
+
+class SentimentAnalyzer:
+    """
+    Advanced sentiment analysis engine supporting multiple backends.
+    
+    Supports both TextBlob (simple) and Transformers (accurate) for sentiment analysis
+    with confidence scoring and async processing capabilities.
+    """
+    
+    def __init__(self, backend: str = "auto", model_name: str = "cardiffnlp/twitter-roberta-base-sentiment-latest"):
+        """
+        Initialize sentiment analyzer.
+        
+        Args:
+            backend: Analysis backend ("textblob", "transformers", or "auto")
+            model_name: Hugging Face model name for transformers backend
+        """
+        self.backend = backend
+        self.model_name = model_name
+        self.logger = logging.getLogger(__name__)
+        self.executor = ThreadPoolExecutor(max_workers=2)
+        
+        # Model caching
+        self._transformer_pipeline = None
+        self._model_loaded = False
+        
+        # Initialize backend
+        self._initialize_backend()
+    
+    def _initialize_backend(self) -> None:
+        """Initialize the selected backend."""
+        if self.backend == "auto":
+            if TRANSFORMERS_AVAILABLE:
+                self.backend = "transformers"
+                self.logger.info("Auto-selected Transformers backend")
+            elif TEXTBLOB_AVAILABLE:
+                self.backend = "textblob"
+                self.logger.info("Auto-selected TextBlob backend")
+            else:
+                raise RuntimeError("No sentiment analysis backend available. Install textblob or transformers.")
+        
+        if self.backend == "transformers" and not TRANSFORMERS_AVAILABLE:
+            raise RuntimeError("Transformers backend requested but not available")
+        
+        if self.backend == "textblob" and not TEXTBLOB_AVAILABLE:
+            raise RuntimeError("TextBlob backend requested but not available")
+    
+    async def _load_transformer_model(self) -> None:
+        """Load transformer model asynchronously."""
+        if self._model_loaded:
+            return
+        
+        try:
+            self.logger.info(f"Loading transformer model: {self.model_name}")
+            
+            # Load model in thread pool to avoid blocking
+            loop = asyncio.get_event_loop()
+            self._transformer_pipeline = await loop.run_in_executor(
+                self.executor,
+                lambda: pipeline(
+                    "sentiment-analysis",
+                    model=self.model_name,
+                    tokenizer=self.model_name,
+                    device=0 if torch.cuda.is_available() else -1,
+                    return_all_scores=True
+                )
+            )
+            
+            self._model_loaded = True
+            self.logger.info("Transformer model loaded successfully")
+            
+        except Exception as e:
+            self.logger.error(f"Failed to load transformer model: {e}")
+            raise RuntimeError(f"Model loading failed: {e}")
+    
+    def _validate_input(self, text: str) -> str:
+        """
+        Validate and sanitize input text.
+        
+        Args:
+            text: Input text to validate
+            
+        Returns:
+            Sanitized text
+            
+        Raises:
+            ValueError: If text is invalid
+        """
+        if not isinstance(text, str):
+            raise ValueError("Input must be a string")
+        
+        text = text.strip()
+        
+        if not text:
+            raise ValueError("Input text cannot be empty")
+        
+        if len(text) > 10000:  # Reasonable limit
+            raise ValueError("Input text too long (max 10,000 characters)")
+        
+        # Basic sanitization
+        text = text.replace('\x00', '')  # Remove null bytes
+        
+        return text
+    
+    def _analyze_with_textblob(self, text: str) -> SentimentResult:
+        """
+        Analyze sentiment using TextBlob.
+        
+        Args:
+            text: Text to analyze
+            
+        Returns:
+            SentimentResult with classification and confidence
+        """
+        try:
+            blob = TextBlob(text)
+            polarity = blob.sentiment.polarity
+            
+            # Convert polarity to label and confidence
+            if polarity > 0.1:
+                label = SentimentLabel.POSITIVE
+                confidence = min(polarity, 1.0)
+            elif polarity < -0.1:
+                label = SentimentLabel.NEGATIVE
+                confidence = min(abs(polarity), 1.0)
+            else:
+                label = SentimentLabel.NEUTRAL
+                confidence = 1.0 - abs(polarity)
+            
+            raw_scores = {
+                "polarity": polarity,
+                "subjectivity": blob.sentiment.subjectivity
+            }
+            
+            return SentimentResult(label, confidence, raw_scores)
+            
+        except Exception as e:
+            self.logger.error(f"TextBlob analysis failed: {e}")
+            raise RuntimeError(f"Sentiment analysis failed: {e}")
+    
+    async def _analyze_with_transformers(self, text: str) -> SentimentResult:
+        """
+        Analyze sentiment using Transformers.
+        
+        Args:
+            text: Text to analyze
+            
+        Returns:
+            SentimentResult with classification and confidence
+        """
+        try:
+            await self._load_transformer_model()
+            
+            # Run inference in thread pool
+            loop = asyncio.get_event_loop()
+            results = await loop.run_in_executor(
+                self.executor,
+                lambda: self._transformer_pipeline(text)
+            )
+            
+            # Process results
+            scores = {result['label'].lower(): result['score'] for result in results[0]}
+            
+            # Map model labels to our labels
+            label_mapping = {
+                'positive': SentimentLabel.POSITIVE,
+                'negative': SentimentLabel.NEGATIVE,
+                'neutral': SentimentLabel.NEUTRAL,
+                'label_0': SentimentLabel.NEGATIVE,  # Some models use numeric labels
+                'label_1': SentimentLabel.NEUTRAL,
+                'label_2': SentimentLabel.POSITIVE
+            }
+            
+            # Find best match
+            best_score = 0
+            best_label = SentimentLabel.NEUTRAL
+            
+            for model_label, score in scores.items():
+                if model_label in label_mapping and score > best_score:
+                    best_score = score
+                    best_label = label_mapping[model_label]
+            
+            return SentimentResult(best_label, best_score, scores)
+            
+        except Exception as e:
+            self.logger.error(f"Transformers analysis failed: {e}")
+            raise RuntimeError(f"Sentiment analysis failed: {e}")
+    
+    async def analyze(self, text: str) -> SentimentResult:
+        """
+        Analyze sentiment of input text.
+        
+        Args:
+            text: Text to analyze
+            
+        Returns:
+            SentimentResult with label, confidence, and raw scores
+            
+        Raises:
+            ValueError: If input is invalid
+            RuntimeError: If analysis fails
+        """
+        # Validate input
+        text = self._validate_input(text)
+        
+        try:
+            if self.backend == "transformers":
+                return await self._analyze_with_transformers(text)
+            elif self.backend == "textblob":
+                # Run TextBlob in thread pool since it's CPU-bound
+                loop = asyncio.get_event_loop()
+                return await loop.run_in_executor(
+                    self.executor,
+                    self._analyze_with_textblob,
+                    text
+                )
+            else:
+                raise RuntimeError(f"Unknown backend: {self.backend}")
+                
+        except Exception as e:
+            self.logger.error(f"Sentiment analysis failed for text: {text[:100]}... Error: {e}")
+            raise
+    
+    async def analyze_batch(self, texts: list[str]) -> list[SentimentResult]:
+        """
+        Analyze sentiment for multiple texts concurrently.
+        
+        Args:
+            texts: List of texts to analyze
+            
+        Returns:
+            List of SentimentResult objects
+        """
+        if not texts:
+            return []
+        
+        # Analyze all texts concurrently
+        tasks = [self.analyze(text) for text in texts]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        
+        # Handle exceptions
+        processed_results = []
+        for i, result in enumerate(results):
+            if isinstance(result, Exception):
+                self.logger.error(f"Failed to analyze text {i}: {result}")
+                # Return neutral result for failed analysis
+                processed_results.append(
+                    SentimentResult(SentimentLabel.NEUTRAL, 0.0, {"error": str(result)})
+                )
+            else:
+                processed_results.append(result)
+        
+        return processed_results
+    
+    def get_info(self) -> Dict[str, Any]:
+        """Get information about the analyzer configuration."""
+        return {
+            "backend": self.backend,
+            "model_name": self.model_name if self.backend == "transformers" else None,
+            "model_loaded": self._model_loaded,
+            "textblob_available": TEXTBLOB_AVAILABLE,
+            "transformers_available": TRANSFORMERS_AVAILABLE,
+            "cuda_available": torch.cuda.is_available() if TRANSFORMERS_AVAILABLE else False
+        }
+    
+    async def cleanup(self) -> None:
+        """Clean up resources."""
+        self.executor.shutdown(wait=True)
+        self.logger.info("Sentiment analyzer cleaned up")
+
+
+# Global analyzer instance for reuse
+_global_analyzer: Optional[SentimentAnalyzer] = None
+
+
+async def get_analyzer(backend: str = "auto") -> SentimentAnalyzer:
+    """
+    Get or create global sentiment analyzer instance.
+    
+    Args:
+        backend: Analysis backend to use
+        
+    Returns:
+        SentimentAnalyzer instance
+    """
+    global _global_analyzer
+    
+    if _global_analyzer is None:
+        _global_analyzer = SentimentAnalyzer(backend=backend)
+    
+    return _global_analyzer
+
+
+async def analyze_sentiment(text: str, backend: str = "auto") -> Dict[str, Any]:
+    """
+    Convenience function for sentiment analysis.
+    
+    Args:
+        text: Text to analyze
+        backend: Analysis backend to use
+        
+    Returns:
+        Dictionary with sentiment analysis results
+    """
+    analyzer = await get_analyzer(backend)
+    result = await analyzer.analyze(text)
+    return result.to_dict()
+
+
+if __name__ == "__main__":
+    # Example usage
+    async def main():
+        analyzer = SentimentAnalyzer(backend="textblob")
+        
+        test_texts = [
+            "I love this product! It's amazing!",
+            "This is terrible and I hate it.",
+            "It's okay, nothing special.",
+            "The weather is nice today."
+        ]
+        
+        for text in test_texts:
+            result = await analyzer.analyze(text)
+            print(f"Text: {text}")
+            print(f"Result: {result.to_dict()}")
+            print("-" * 50)
+        
+        await analyzer.cleanup()
+    
+    asyncio.run(main()) 

src/tools.py

@@ -0,0 +1,421 @@
+"""
+MCP tool definitions for sentiment analysis server.
+
+This module defines the tools available through the Model Context Protocol,
+including sentiment analysis, batch processing, and analyzer information tools.
+"""
+
+import logging
+from typing import Dict, Any, List, Optional
+from pydantic import BaseModel, Field
+import asyncio
+
+from .sentiment_analyzer import get_analyzer, SentimentAnalyzer
+
+
+class SentimentAnalysisInput(BaseModel):
+    """Input schema for sentiment analysis tool."""
+    text: str = Field(..., description="Text to analyze for sentiment", min_length=1, max_length=10000)
+    backend: Optional[str] = Field("auto", description="Analysis backend: 'textblob', 'transformers', or 'auto'")
+
+
+class BatchSentimentAnalysisInput(BaseModel):
+    """Input schema for batch sentiment analysis tool."""
+    texts: List[str] = Field(..., description="List of texts to analyze for sentiment", min_items=1, max_items=100)
+    backend: Optional[str] = Field("auto", description="Analysis backend: 'textblob', 'transformers', or 'auto'")
+
+
+class AnalyzerInfoInput(BaseModel):
+    """Input schema for analyzer info tool."""
+    backend: Optional[str] = Field("auto", description="Backend to get info for")
+
+
+class MCPTools:
+    """
+    MCP tool registry and handlers for sentiment analysis.
+    
+    This class manages the registration and execution of tools available
+    through the Model Context Protocol interface.
+    """
+    
+    def __init__(self):
+        self.logger = logging.getLogger(__name__)
+        self._tools = {}
+        self._register_tools()
+    
+    def _register_tools(self) -> None:
+        """Register all available MCP tools."""
+        self._tools = {
+            "analyze_sentiment": {
+                "name": "analyze_sentiment",
+                "description": "Analyze the sentiment of a given text",
+                "inputSchema": {
+                    "type": "object",
+                    "properties": {
+                        "text": {
+                            "type": "string",
+                            "description": "Text to analyze for sentiment",
+                            "minLength": 1,
+                            "maxLength": 10000
+                        },
+                        "backend": {
+                            "type": "string",
+                            "description": "Analysis backend: 'textblob', 'transformers', or 'auto'",
+                            "enum": ["textblob", "transformers", "auto"],
+                            "default": "auto"
+                        }
+                    },
+                    "required": ["text"]
+                },
+                "handler": self._handle_analyze_sentiment
+            },
+            
+            "analyze_sentiment_batch": {
+                "name": "analyze_sentiment_batch",
+                "description": "Analyze sentiment for multiple texts in batch",
+                "inputSchema": {
+                    "type": "object",
+                    "properties": {
+                        "texts": {
+                            "type": "array",
+                            "description": "List of texts to analyze for sentiment",
+                            "items": {
+                                "type": "string",
+                                "minLength": 1,
+                                "maxLength": 10000
+                            },
+                            "minItems": 1,
+                            "maxItems": 100
+                        },
+                        "backend": {
+                            "type": "string",
+                            "description": "Analysis backend: 'textblob', 'transformers', or 'auto'",
+                            "enum": ["textblob", "transformers", "auto"],
+                            "default": "auto"
+                        }
+                    },
+                    "required": ["texts"]
+                },
+                "handler": self._handle_analyze_sentiment_batch
+            },
+            
+            "get_analyzer_info": {
+                "name": "get_analyzer_info",
+                "description": "Get information about the sentiment analyzer configuration",
+                "inputSchema": {
+                    "type": "object",
+                    "properties": {
+                        "backend": {
+                            "type": "string",
+                            "description": "Backend to get info for",
+                            "enum": ["textblob", "transformers", "auto"],
+                            "default": "auto"
+                        }
+                    },
+                    "required": []
+                },
+                "handler": self._handle_get_analyzer_info
+            },
+            
+            "health_check": {
+                "name": "health_check",
+                "description": "Check the health status of the sentiment analysis service",
+                "inputSchema": {
+                    "type": "object",
+                    "properties": {},
+                    "required": []
+                },
+                "handler": self._handle_health_check
+            }
+        }
+        
+        self.logger.info(f"Registered {len(self._tools)} MCP tools")
+    
+    def get_tools(self) -> List[Dict[str, Any]]:
+        """
+        Get list of available tools for MCP protocol.
+        
+        Returns:
+            List of tool definitions
+        """
+        return [
+            {
+                "name": tool["name"],
+                "description": tool["description"],
+                "inputSchema": tool["inputSchema"]
+            }
+            for tool in self._tools.values()
+        ]
+    
+    async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Call a registered tool with given arguments.
+        
+        Args:
+            name: Tool name
+            arguments: Tool arguments
+            
+        Returns:
+            Tool execution result
+            
+        Raises:
+            ValueError: If tool not found or arguments invalid
+            RuntimeError: If tool execution fails
+        """
+        if name not in self._tools:
+            raise ValueError(f"Tool '{name}' not found. Available tools: {list(self._tools.keys())}")
+        
+        tool = self._tools[name]
+        handler = tool["handler"]
+        
+        try:
+            self.logger.info(f"Calling tool '{name}' with arguments: {arguments}")
+            result = await handler(arguments)
+            self.logger.info(f"Tool '{name}' completed successfully")
+            return result
+            
+        except Exception as e:
+            self.logger.error(f"Tool '{name}' failed: {e}")
+            raise RuntimeError(f"Tool execution failed: {e}")
+    
+    async def _handle_analyze_sentiment(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle sentiment analysis tool call.
+        
+        Args:
+            arguments: Tool arguments containing text and optional backend
+            
+        Returns:
+            Sentiment analysis result
+        """
+        try:
+            # Validate input
+            input_data = SentimentAnalysisInput(**arguments)
+            
+            # Get analyzer and perform analysis
+            analyzer = await get_analyzer(input_data.backend)
+            result = await analyzer.analyze(input_data.text)
+            
+            return {
+                "success": True,
+                "result": result.to_dict(),
+                "metadata": {
+                    "backend": analyzer.backend,
+                    "text_length": len(input_data.text),
+                    "model_info": analyzer.get_info()
+                }
+            }
+            
+        except Exception as e:
+            return {
+                "success": False,
+                "error": str(e),
+                "error_type": type(e).__name__
+            }
+    
+    async def _handle_analyze_sentiment_batch(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle batch sentiment analysis tool call.
+        
+        Args:
+            arguments: Tool arguments containing texts and optional backend
+            
+        Returns:
+            Batch sentiment analysis results
+        """
+        try:
+            # Validate input
+            input_data = BatchSentimentAnalysisInput(**arguments)
+            
+            # Get analyzer and perform batch analysis
+            analyzer = await get_analyzer(input_data.backend)
+            results = await analyzer.analyze_batch(input_data.texts)
+            
+            # Convert results to dictionaries
+            result_dicts = [result.to_dict() for result in results]
+            
+            # Calculate summary statistics
+            labels = [result.label.value for result in results]
+            label_counts = {
+                "positive": labels.count("positive"),
+                "negative": labels.count("negative"),
+                "neutral": labels.count("neutral")
+            }
+            
+            avg_confidence = sum(result.confidence for result in results) / len(results)
+            
+            return {
+                "success": True,
+                "results": result_dicts,
+                "summary": {
+                    "total_texts": len(input_data.texts),
+                    "label_distribution": label_counts,
+                    "average_confidence": round(avg_confidence, 4)
+                },
+                "metadata": {
+                    "backend": analyzer.backend,
+                    "model_info": analyzer.get_info()
+                }
+            }
+            
+        except Exception as e:
+            return {
+                "success": False,
+                "error": str(e),
+                "error_type": type(e).__name__
+            }
+    
+    async def _handle_get_analyzer_info(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle analyzer info tool call.
+        
+        Args:
+            arguments: Tool arguments containing optional backend
+            
+        Returns:
+            Analyzer configuration information
+        """
+        try:
+            # Validate input
+            input_data = AnalyzerInfoInput(**arguments)
+            
+            # Get analyzer info
+            analyzer = await get_analyzer(input_data.backend)
+            info = analyzer.get_info()
+            
+            return {
+                "success": True,
+                "info": info,
+                "available_backends": ["textblob", "transformers", "auto"],
+                "recommended_backend": "transformers" if info.get("transformers_available") else "textblob"
+            }
+            
+        except Exception as e:
+            return {
+                "success": False,
+                "error": str(e),
+                "error_type": type(e).__name__
+            }
+    
+    async def _handle_health_check(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Handle health check tool call.
+        
+        Args:
+            arguments: Tool arguments (empty for health check)
+            
+        Returns:
+            Health status information
+        """
+        try:
+            # Test basic functionality
+            test_text = "This is a test message for health check."
+            analyzer = await get_analyzer("auto")
+            result = await analyzer.analyze(test_text)
+            
+            return {
+                "success": True,
+                "status": "healthy",
+                "test_result": result.to_dict(),
+                "analyzer_info": analyzer.get_info(),
+                "timestamp": asyncio.get_event_loop().time()
+            }
+            
+        except Exception as e:
+            return {
+                "success": False,
+                "status": "unhealthy",
+                "error": str(e),
+                "error_type": type(e).__name__,
+                "timestamp": asyncio.get_event_loop().time()
+            }
+
+
+# Global tools instance
+_global_tools: Optional[MCPTools] = None
+
+
+def get_tools() -> MCPTools:
+    """
+    Get or create global MCP tools instance.
+    
+    Returns:
+        MCPTools instance
+    """
+    global _global_tools
+    
+    if _global_tools is None:
+        _global_tools = MCPTools()
+    
+    return _global_tools
+
+
+async def list_tools() -> List[Dict[str, Any]]:
+    """
+    Get list of available MCP tools.
+    
+    Returns:
+        List of tool definitions
+    """
+    tools = get_tools()
+    return tools.get_tools()
+
+
+async def call_tool(name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Call an MCP tool with given arguments.
+    
+    Args:
+        name: Tool name
+        arguments: Tool arguments
+        
+    Returns:
+        Tool execution result
+    """
+    tools = get_tools()
+    return await tools.call_tool(name, arguments)
+
+
+if __name__ == "__main__":
+    # Example usage
+    async def main():
+        tools = get_tools()
+        
+        # List available tools
+        available_tools = tools.get_tools()
+        print("Available tools:")
+        for tool in available_tools:
+            print(f"- {tool['name']}: {tool['description']}")
+        
+        print("\n" + "="*50 + "\n")
+        
+        # Test sentiment analysis tool
+        result = await tools.call_tool("analyze_sentiment", {
+            "text": "I love this new feature! It's absolutely amazing!",
+            "backend": "textblob"
+        })
+        print("Sentiment analysis result:")
+        print(result)
+        
+        print("\n" + "="*50 + "\n")
+        
+        # Test batch analysis
+        batch_result = await tools.call_tool("analyze_sentiment_batch", {
+            "texts": [
+                "This is great!",
+                "I hate this.",
+                "It's okay, I guess."
+            ],
+            "backend": "textblob"
+        })
+        print("Batch analysis result:")
+        print(batch_result)
+        
+        print("\n" + "="*50 + "\n")
+        
+        # Test health check
+        health_result = await tools.call_tool("health_check", {})
+        print("Health check result:")
+        print(health_result)
+    
+    asyncio.run(main())