update logic

README.md

@@ -0,0 +1,431 @@
+---
+title:  Sentiment Analysis Server
+emoji: 🎭
+colorFrom: blue
+colorTo: purple
+sdk: gradio
+sdk_version: 5.33.2
+app_file: app.py
+pinned: false
+---
+
+# 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
+python3 -m pip install -r requirements.txt
+```
+
+### 3. Download TextBlob Corpora (Optional)
+```bash
+python3 -c "import nltk; nltk.download('punkt'); nltk.download('brown')"
+```
+
+## 🎯 Quick Start
+
+### Run Tests
+```bash
+python3 app.py --mode test
+```
+
+### Launch Web Interface
+```bash
+python3 app.py --mode gradio
+```
+Visit `http://localhost:7860` to access the web interface.
+
+### Launch Web Interface with MCP Server
+```bash
+python3 app.py --mode gradio --mcp-server
+```
+Enables both web interface and MCP server functionality in a single Gradio app.
+
+### Start MCP Server
+```bash
+python3 app.py --mode mcp
+```
+
+### Combined Mode (Default)
+```bash
+python3 app.py --mode combined
+```
+Runs both MCP server and Gradio interface simultaneously.
+
+## 📖 Usage Examples
+
+### Command Line Options
+
+```bash
+# Different execution modes
+python3 app.py --mode mcp                    # MCP server only
+python3 app.py --mode gradio                 # Web interface only
+python3 app.py --mode combined               # Both services (default)
+python3 app.py --mode test                   # Run functionality tests
+
+# Gradio customization
+python3 app.py --mode gradio --port 8080     # Custom port
+python3 app.py --mode gradio --share         # Enable public sharing
+python3 app.py --mode gradio --debug         # Debug mode
+python3 app.py --mode gradio --mcp-server    # Enable MCP server in Gradio
+
+# Logging control
+python3 app.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": {}
+}
+```
+
+## 🔗 Gradio MCP Server Integration
+
+The Gradio web interface can also serve as an MCP server, providing API endpoints that correspond to each sentiment analysis function. This allows the same interface to serve both human users through the web UI and AI models through the MCP protocol.
+
+### Enable MCP Server in Gradio
+
+#### Method 1: Command Line Flag
+```bash
+python3 app.py --mode gradio --mcp-server
+```
+
+#### Method 2: Environment Variable
+```bash
+export GRADIO_MCP_SERVER=true
+python3 app.py --mode gradio
+```
+
+#### Method 3: Programmatic (Python API)
+```python
+from src import create_gradio_interface
+
+interface = create_gradio_interface()
+interface.launch(mcp_server=True)
+```
+
+### Available MCP Endpoints
+
+When MCP server is enabled, the Gradio interface automatically exposes these API endpoints:
+
+- **POST /api/analyze_sentiment** - Single text analysis
+- **POST /api/analyze_sentiment_batch** - Batch text analysis  
+- **POST /api/health_check** - Server health status
+
+### MCP API Usage Examples
+
+```bash
+# Single sentiment analysis
+curl -X POST http://localhost:7860/api/analyze_sentiment \
+  -H "Content-Type: application/json" \
+  -d '{"text": "I love this product!", "backend": "auto"}'
+
+# Batch analysis
+curl -X POST http://localhost:7860/api/analyze_sentiment_batch \
+  -H "Content-Type: application/json" \
+  -d '{"texts": ["Great!", "Terrible!", "Okay."], "backend": "transformers"}'
+
+# Health check
+curl -X POST http://localhost:7860/api/health_check \
+  -H "Content-Type: application/json" \
+  -d '{}'
+```
+
+## 🏗️ 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
+├── app.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 GRADIO_MCP_SERVER=true              # Enable MCP server in Gradio interface
+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 app.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 app.py --mode gradio --port 8080
+```
+
+### Debug Mode
+```bash
+python app.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** 

app.py

@@ -0,0 +1,368 @@
+#!/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 app.py --mode mcp          # Run MCP server only
+    python app.py --mode gradio       # Run Gradio interface only
+    python app.py --mode combined     # Run both (default)
+    python app.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 app.py --mode mcp                    # MCP server only
+  python app.py --mode gradio                 # Gradio interface only  
+  python app.py --mode combined               # Both services
+  python app.py --mode test                   # Run tests
+  python app.py --mode gradio --port 8080     # Custom port
+  python app.py --mode gradio --share         # Public sharing
+  python app.py --mode gradio --mcp-server    # Gradio with MCP server
+  
+Environment Variables:
+  GRADIO_MCP_SERVER=true                      # Enable MCP server in Gradio
+        """
+    )
+    
+    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"
+    )
+    
+    parser.add_argument(
+        "--mcp-server",
+        action="store_true",
+        help="Enable MCP server functionality in Gradio interface"
+    )
+    
+    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,
+                "mcp_server": args.mcp_server
+            }
+            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,
+                "mcp_server": args.mcp_server
+            }
+            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,576 @@
+"""
+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
+import os
+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
+from .tools import list_tools
+
+
+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())
+            
+            def get_mcp_schema() -> str:
+                """Get MCP tools schema as JSON."""
+                return asyncio.run(self._get_mcp_schema_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>
+            """
+    
+    async def _get_mcp_schema_async(self) -> str:
+        """Get MCP tools schema as formatted JSON."""
+        try:
+            tools = await list_tools()
+            schema = {
+                "mcp_version": "2024-11-05",
+                "server_info": {
+                    "name": "sentiment-analyzer",
+                    "version": "1.0.0",
+                    "description": "Sentiment analysis server using TextBlob and Transformers"
+                },
+                "tools": tools,
+                "total_tools": len(tools)
+            }
+            return json.dumps(schema, indent=2)
+            
+        except Exception as e:
+            self.logger.error(f"Failed to get MCP schema: {e}")
+            return json.dumps({
+                "error": str(e),
+                "error_type": type(e).__name__
+            }, indent=2)
+    
+    def launch(self, **kwargs) -> None:
+        if not self.interface:
+            raise RuntimeError("Interface not initialized")
+        
+        # Check for MCP server mode from environment variable or parameter
+        mcp_server_enabled = (
+            kwargs.get("mcp_server", False) or 
+            os.getenv("GRADIO_MCP_SERVER", "").lower() in ("true", "1", "yes", "on")
+        )
+        
+        launch_params = {
+            "server_name": "0.0.0.0",
+            "server_port": 7860,
+            "share": False,
+            "debug": False,
+            "show_error": True,
+            "quiet": False
+        }
+        
+        # Add MCP server parameter if enabled
+        if mcp_server_enabled:
+            launch_params["mcp_server"] = True
+            self.logger.info("MCP server functionality enabled for Gradio interface")
+        
+        launch_params.update(kwargs)
+        
+        self.logger.info(f"Launching Gradio interface on {launch_params['server_name']}:{launch_params['server_port']}")
+        if mcp_server_enabled:
+            self.logger.info("Gradio interface will also serve as MCP server with API endpoints")
+        
+        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())