src/mcp_server.py

"""
Sentiment Evolution Tracker MCP Server
Analyzes sentiment trajectories in conversations to detect opinion changes and predict risks.

Key MCP Protocol Requirements:
1. MUST use stdio_server() for communication with Claude Desktop
2. MUST NOT log to stdout (reserved for protocol messages)
3. MUST log to stderr or file only
4. MUST return TextContent with proper formatting
5. MUST handle async/await correctly
"""

import json
import logging
import asyncio
import sys
import os
from typing import Any

# MCP imports - CRITICAL: correct imports for MCP protocol
from mcp.server import Server
from mcp.server.lowlevel import NotificationOptions
from mcp.server.models import InitializationOptions
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server

# Import analysis modules
from sentiment_analyzer import SentimentAnalyzer
from pattern_detector import PatternDetector
from risk_predictor import RiskPredictor
from database_manager import AnalysisDatabase

# ============================================================================
# LOGGING SETUP - CRITICAL FOR DEBUGGING
# Log to file and stderr ONLY (never stdout - that's for MCP protocol)
# ============================================================================

# Get absolute path for log file
log_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'mcp_server.log')

# Configure logging to file
logging.basicConfig(
    filename=log_file,
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Also send errors to stderr (Claude Desktop will capture these)
stderr_handler = logging.StreamHandler(sys.stderr)
stderr_handler.setLevel(logging.ERROR)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
stderr_handler.setFormatter(formatter)
logger.addHandler(stderr_handler)

logger.info("=" * 80)
logger.info("MCP Server starting up")
logger.info(f"Python executable: {sys.executable}")
logger.info(f"Log file: {log_file}")
logger.info("=" * 80)

# ============================================================================
# INITIALIZE MCP SERVER
# ============================================================================

server = Server("sentiment-evolution-tracker")

# Initialize analysis modules - CRITICAL: do this before accepting connections
logger.info("Initializing analysis modules...")
try:
    sentiment_analyzer = SentimentAnalyzer()
    pattern_detector = PatternDetector()
    risk_predictor = RiskPredictor()
    db = AnalysisDatabase()
    logger.info("✓ All analysis modules initialized successfully")
    logger.info(f"✓ Database: {db.db_path}")
except Exception as e:
    error_msg = f"FATAL: Failed to initialize modules: {str(e)}"
    logger.error(error_msg, exc_info=True)
    sys.stderr.write(error_msg + "\n")
    sys.exit(1)

# ============================================================================
# TOOL DEFINITIONS
# ============================================================================

@server.list_tools()
async def list_tools() -> list[Tool]:
    """List all available tools."""
    logger.debug("list_tools() called by Claude")
    
    tools = [
        Tool(
            name="analyze_sentiment_evolution",
            description="Analyzes sentiment evolution across a series of messages to detect trending patterns (improving, declining, or stable sentiment)",
            inputSchema={
                "type": "object",
                "properties": {
                    "messages": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "List of messages to analyze, ordered chronologically"
                    }
                },
                "required": ["messages"]
            }
        ),
        Tool(
            name="detect_risk_signals",
            description="Detects risk signals in conversations (competitor mentions, frustration, disengagement, pricing concerns)",
            inputSchema={
                "type": "object",
                "properties": {
                    "messages": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "List of messages to analyze for risk signals"
                    },
                    "context_type": {
                        "type": "string",
                        "enum": ["customer", "employee", "email"],
                        "description": "Type of conversation context"
                    }
                },
                "required": ["messages", "context_type"]
            }
        ),
        Tool(
            name="predict_next_action",
            description="Predicts the likely next action or outcome based on sentiment and signals (CHURN, RESOLUTION, ESCALATION)",
            inputSchema={
                "type": "object",
                "properties": {
                    "messages": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "List of messages for analysis"
                    },
                    "context_type": {
                        "type": "string",
                        "enum": ["customer", "employee", "email"],
                        "description": "Type of conversation context"
                    }
                },
                "required": ["messages", "context_type"]
            }
        ),
        Tool(
            name="get_customer_history",
            description="Retrieves historical analysis data for a specific customer, including all previous analyses, trends, and active alerts. THIS REQUIRES DATABASE ACCESS - Claude cannot do this alone!",
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_id": {
                        "type": "string",
                        "description": "Unique customer identifier"
                    }
                },
                "required": ["customer_id"]
            }
        ),
        Tool(
            name="get_high_risk_customers",
            description="Returns list of all customers currently at high risk of churn. THIS REQUIRES DATABASE ACCESS - Claude cannot do this alone!",
            inputSchema={
                "type": "object",
                "properties": {
                    "threshold": {
                        "type": "number",
                        "description": "Risk threshold (0-1, default 0.75)",
                        "default": 0.75
                    }
                },
                "required": []
            }
        ),
        Tool(
            name="get_database_statistics",
            description="Returns overall statistics about analyzed customers and alerts. THIS REQUIRES DATABASE ACCESS - Claude cannot do this alone!",
            inputSchema={
                "type": "object",
                "properties": {}
            }
        ),
        Tool(
            name="save_analysis",
            description="Explicitly save a sentiment analysis with a customer name to the database. Use this to save analysis results with a specific customer identifier.",
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_id": {
                        "type": "string",
                        "description": "Unique customer identifier (e.g., 'LUIS_RAMIREZ', 'CUST_001_ACME')"
                    },
                    "customer_name": {
                        "type": "string",
                        "description": "Customer display name (optional)"
                    },
                    "messages": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "List of messages in the conversation"
                    },
                    "sentiment_score": {
                        "type": "number",
                        "description": "Overall sentiment score (0-100)"
                    },
                    "trend": {
                        "type": "string",
                        "enum": ["RISING", "DECLINING", "STABLE"],
                        "description": "Sentiment trend"
                    },
                    "risk_level": {
                        "type": "string",
                        "description": "Risk classification (LOW, MEDIUM, HIGH)"
                    },
                    "predicted_action": {
                        "type": "string",
                        "description": "Recommended action (CHURN, RESOLUTION, ESCALATION)"
                    },
                    "confidence": {
                        "type": "number",
                        "description": "Confidence level (0-1.0)"
                    },
                    "context_type": {
                        "type": "string",
                        "enum": ["customer", "employee", "email"],
                        "description": "Type of conversation",
                        "default": "customer"
                    }
                },
                "required": ["customer_id", "messages", "sentiment_score", "trend", "predicted_action", "confidence"]
            }
        )
    ]
    
    logger.info(f"✓ Returning {len(tools)} tools to Claude")
    return tools


# ============================================================================
# TOOL HANDLERS
# ============================================================================

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """
    Execute tool based on name and arguments.
    ALL ERRORS are logged to stderr and file.
    """
    
    try:
        logger.info(f"Tool call received: {name}")
        logger.debug(f"Arguments: {arguments}")
        
        if name == "analyze_sentiment_evolution":
            # Extract messages - must be non-empty
            messages = arguments.get("messages", [])
            if not messages or not isinstance(messages, list):
                error_msg = "Missing or invalid 'messages' parameter (must be non-empty array)"
                logger.warning(f"analyze_sentiment_evolution: {error_msg}")
                return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
            
            logger.info(f"Analyzing sentiment evolution for {len(messages)} messages")
            result = sentiment_analyzer.analyze_evolution(messages)
            
            # Save to database
            customer_id = arguments.get("customer_id", f"customer_{hash(str(messages))}")
            db.save_analysis(customer_id, "conversation", messages, result)
            logger.info(f"✓ analyze_sentiment_evolution completed and saved to database")
            
            return [TextContent(type="text", text=json.dumps(result))]
        
        elif name == "detect_risk_signals":
            messages = arguments.get("messages", [])
            context_type = arguments.get("context_type", "customer")
            
            if not messages or not isinstance(messages, list):
                error_msg = "Missing or invalid 'messages' parameter (must be non-empty array)"
                logger.warning(f"detect_risk_signals: {error_msg}")
                return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
            
            if context_type not in ["customer", "employee", "email"]:
                context_type = "customer"
                logger.info(f"Invalid context_type, defaulting to 'customer'")
            
            logger.info(f"Detecting risk signals for {len(messages)} messages (context: {context_type})")
            result = pattern_detector.detect_signals(messages, context_type)
            
            # Save to database
            customer_id = arguments.get("customer_id", f"customer_{hash(str(messages))}")
            db.save_analysis(customer_id, context_type, messages, result)
            logger.info(f"✓ detect_risk_signals completed and saved to database")
            
            return [TextContent(type="text", text=json.dumps(result))]
        
        elif name == "predict_next_action":
            messages = arguments.get("messages", [])
            context_type = arguments.get("context_type", "customer")
            
            if not messages or not isinstance(messages, list):
                error_msg = "Missing or invalid 'messages' parameter (must be non-empty array)"
                logger.warning(f"predict_next_action: {error_msg}")
                return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
            
            if context_type not in ["customer", "employee", "email"]:
                context_type = "customer"
                logger.info(f"Invalid context_type, defaulting to 'customer'")
            
            logger.info(f"Predicting next action for {len(messages)} messages (context: {context_type})")
            result = risk_predictor.predict_action(messages, context_type)
            
            # Save to database
            customer_id = arguments.get("customer_id", f"customer_{hash(str(messages))}")
            db.save_analysis(customer_id, context_type, messages, result)
            logger.info(f"✓ predict_next_action completed and saved to database")
            
            return [TextContent(type="text", text=json.dumps(result))]
        
        elif name == "get_customer_history":
            customer_id = arguments.get("customer_id", "")
            if not customer_id:
                error_msg = "Missing 'customer_id' parameter"
                logger.warning(f"get_customer_history: {error_msg}")
                return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
            
            logger.info(f"Retrieving history for customer: {customer_id}")
            result = db.get_customer_history(customer_id)
            logger.info(f"✓ get_customer_history completed - found {len(result.get('analyses', []))} analyses")
            
            return [TextContent(type="text", text=json.dumps(result))]
        
        elif name == "get_high_risk_customers":
            threshold = float(arguments.get("threshold", 0.75))
            
            logger.info(f"Retrieving high-risk customers (threshold: {threshold})")
            result = db.get_high_risk_customers(threshold)
            logger.info(f"✓ get_high_risk_customers completed - found {len(result)} at-risk customers")
            
            return [TextContent(type="text", text=json.dumps({
                'high_risk_customers': result,
                'count': len(result),
                'threshold': threshold
            }))]
        
        elif name == "get_database_statistics":
            logger.info("Retrieving database statistics")
            result = db.get_statistics()
            logger.info(f"✓ get_database_statistics completed")
            
            return [TextContent(type="text", text=json.dumps(result))]
        
        elif name == "save_analysis":
            """Save analysis results explicitly with customer identifier"""
            customer_id = arguments.get("customer_id", "")
            if not customer_id:
                error_msg = "Missing 'customer_id' parameter"
                logger.warning(f"save_analysis: {error_msg}")
                return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
            
            messages = arguments.get("messages", [])
            if not messages or not isinstance(messages, list):
                error_msg = "Missing or invalid 'messages' parameter (must be non-empty array)"
                logger.warning(f"save_analysis: {error_msg}")
                return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
            
            # Build analysis dictionary from parameters
            analysis = {
                "current_sentiment": arguments.get("sentiment_score", 50),
                "trend": arguments.get("trend", "STABLE"),
                "risk_level": arguments.get("risk_level", "MEDIUM"),
                "predicted_action": arguments.get("predicted_action", "UNKNOWN"),
                "confidence": arguments.get("confidence", 0.5)
            }
            
            context_type = arguments.get("context_type", "customer")
            if context_type not in ["customer", "employee", "email"]:
                context_type = "customer"
            
            logger.info(f"Saving analysis for customer: {customer_id}")
            logger.debug(f"Analysis data: {analysis}")
            
            # Save to database
            analysis_id = db.save_analysis(customer_id, context_type, messages, analysis)
            
            logger.info(f"✓ Analysis saved successfully - ID: {analysis_id}, Customer: {customer_id}")
            
            return [TextContent(type="text", text=json.dumps({
                "success": True,
                "analysis_id": analysis_id,
                "customer_id": customer_id,
                "message": f"Analysis saved for {customer_id} with {len(messages)} messages"
            }))]
        
        else:
            error_msg = f"Unknown tool: {name}"
            logger.error(error_msg)
            return [TextContent(type="text", text=json.dumps({"error": error_msg}))]
    
    except Exception as e:
        error_msg = f"Error in tool {name}: {str(e)}"
        logger.error(error_msg, exc_info=True)
        sys.stderr.write(f"ERROR: {error_msg}\n")
        return [TextContent(type="text", text=json.dumps({"error": error_msg}))]


# ============================================================================
# MAIN SERVER LOOP
# ============================================================================

async def main():
    """
    Run the MCP server with stdio transport.
    This is the CRITICAL function that handles protocol communication.
    
    IMPORTANT: stdio_server() yields a tuple (read_stream, write_stream)
    """
    logger.info("main() called - entering async loop")
    
    try:
        # Use stdio_server context manager for proper protocol handling
        async with stdio_server() as (read_stream, write_stream):
            logger.info("✓ stdio_server initialized - streams ready")
            logger.info("✓ Creating InitializationOptions...")
            
            # Create initialization options required by MCP protocol
            init_options = InitializationOptions(
                server_name="sentiment-evolution-tracker",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                )
            )
            
            logger.info("✓ Connecting to Claude Desktop...")
            logger.info(f"✓ Server capabilities: {init_options.capabilities}")
            
            # Start the server with stdin/stdout streams and initialization options
            # This blocks until connection is closed
            await server.run(read_stream, write_stream, init_options)
            
            logger.info("✓ Server loop completed (connection closed)")
    
    except Exception as e:
        error_msg = f"Server error in main(): {str(e)}"
        logger.error(error_msg, exc_info=True)
        sys.stderr.write(f"FATAL ERROR: {error_msg}\n")
        raise


# ============================================================================
# ENTRY POINT
# ============================================================================

if __name__ == "__main__":
    logger.info("=" * 80)
    logger.info("MCP Server Process Starting")
    logger.info("=" * 80)
    
    try:
        # Windows compatibility: set event loop policy
        if sys.platform == "win32":
            logger.info("Windows detected - setting WindowsSelectorEventLoopPolicy")
            asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
        
        # Run the server
        logger.info("Calling asyncio.run(main())")
        asyncio.run(main())
        logger.info("MCP Server exited normally")
    
    except KeyboardInterrupt:
        logger.info("Server stopped by user (KeyboardInterrupt)")
        sys.exit(0)
    
    except Exception as e:
        error_msg = f"FATAL ERROR in main process: {str(e)}"
        logger.critical(error_msg, exc_info=True)
        sys.stderr.write(f"\n{error_msg}\n")
        sys.exit(1)