MCP Log Analyzer

import asyncio
from mcp.server.fastmcp import FastMCP
from typing import List, Dict, Any
import json

mcp = FastMCP("LogAnalyzer")

def safe_str(value: Any) -> str:
    return str(value) if value is not None else ""

@mcp.tool()
def analyze_logs(logs: List[Dict]) -> Dict:
    """
    Analyze log entries to extract errors and warnings.
    """
    errors = []
    warnings = []

    for log in logs:
        level = safe_str(log.get("level")).upper()
        if level == "ERROR":
            errors.append({
                "timestamp": safe_str(log.get("timestamp")),
                "component": safe_str(log.get("component")),
                "message": safe_str(log.get("message")),
                "stack_trace": safe_str(log.get("stack_trace"))
            })
        elif level == "WARNING":
            warnings.append({
                "timestamp": safe_str(log.get("timestamp")),
                "component": safe_str(log.get("component")),
                "message": safe_str(log.get("message"))
            })

    result = {
        "summary": {
            "total_logs": len(logs),
            "error_count": len(errors),
            "warning_count": len(warnings)
        },
        "errors": errors,
        "warnings": warnings
    }
    return result


@mcp.tool()
def suggest_fix(errors_and_warnings: Dict) -> List[str]:
    """
    Suggest potential fixes for each issue based on the log message content.

    Args:
        errors_and_warnings (Dict): Dictionary containing analyzed errors and warnings.

    Returns:
        List of textual suges or actions to take for each issue.
    """
    suges = []

    # Process ERROR level issues
    for error in errors_and_warnings.get("errors", []):
        msg = safe_str(error.get("message")).lower()
        component = safe_str(error.get("component")).lower()
        stack_trace = safe_str(error.get("stack_trace")).lower()

        # Database-related errors
        if any(keyword in msg for keyword in ["database", "connection timeout", "deadlock", "sql"]):
            if "timeout" in msg:
                suges.append(
                    "🔧 Database: Increase connection timeout, optimize slow queries, and check network connectivity between application and database server.")
            elif "deadlock" in msg:
                suges.append(
                    "🔧 Database: Review transaction isolation levels, minimize transaction duration, and implement proper lock ordering to prevent deadlocks.")
            elif "pool" in msg and ("exhausted" in msg or "running low" in msg):
                suges.append(
                    "🔧 Database: Increase connection pool size, implement connection pooling best practices, and monitor connection leaks.")

        # Memory-related errors
        elif any(keyword in msg for keyword in ["outofmemoryerror", "heap space", "memory"]):
            if "heap space" in msg:
                suges.append(
                    "🔧 Memory: Increase JVM heap size (-Xmx), optimize memory usage in DataProcessor, and implement memory profiling to identify leaks.")
            elif "outofmemory" in msg:
                suges.append(
                    "🔧 Memory: Monitor memory usage patterns, implement garbage collection tuning, and consider using memory-efficient data structures.")

        # Authentication errors
        elif any(keyword in msg for keyword in ["nullpointer", "authentication", "oauth", "ldap", "token"]):
            if "nullpointer" in msg:
                suges.append(
                    "🔧 Authentication: Add comprehensive null checks in AuthService, implement proper error handling, and validate input parameters before processing.")
            elif "token expired" in msg or "oauth" in msg:
                suges.append(
                    "🔧 Authentication: Implement automatic token refresh mechanism, add token expiry validation, and configure proper token lifetime management.")
            elif "ldap" in msg:
                suges.append(
                    "🔧 Authentication: Check LDAP server connectivity, verify credentials, implement connection retry logic, and add fallback authentication methods.")

        # Network and timeout errors
        elif any(keyword in msg for keyword in ["timeout", "ssl", "connection", "network"]):
            if "timeout" in msg:
                if "payment" in msg or "gateway" in msg:
                    suges.append(
                        "🔧 Network: Increase PaymentGateway timeout settings, implement circuit breaker pattern, and add retry logic with exponential backoff.")
                else:
                    suges.append(
                        "🔧 Network: Check network connectivity, increase timeout values, implement connection pooling, and add health checks for external services.")
            elif "ssl" in msg:
                suges.append(
                    "🔧 Network: Verify SSL certificates, update certificate trust store, check TLS version compatibility, and implement proper SSL configuration.")

        # File and configuration errors
        elif any(keyword in msg for keyword in ["filenotfound", "config", "properties"]):
            suges.append(
                "🔧 Configuration: Verify file paths, ensure configuration files exist, implement configuration validation, and add default fallback configurations.")

        # Cache and Redis errors
        elif any(keyword in msg for keyword in ["redis", "cache", "pool exhausted"]):
            if "pool exhausted" in msg:
                suges.append(
                    "🔧 Cache: Increase Redis connection pool size, implement connection management, monitor cache hit ratios, and add connection health checks.")

        # External service errors
        elif any(keyword in msg for keyword in ["elasticsearch", "kafka", "service discovery"]):
            if "elasticsearch" in msg:
                suges.append(
                    "🔧 External Services: Check Elasticsearch cluster health, verify connectivity, implement retry mechanisms, and add service monitoring.")
            elif "kafka" in msg:
                suges.append(
                    "🔧 External Services: Verify Kafka broker connectivity, check network configuration, implement producer/consumer error handling.")
            elif "service discovery" in msg:
                suges.append(
                    "🔧 External Services: Check service registry health, implement service discovery fallbacks, and verify network connectivity to registry.")

        # Threading and execution errors
        elif any(keyword in msg for keyword in ["thread", "rejected execution", "circuit breaker"]):
            if "rejected execution" in msg:
                suges.append(
                    "🔧 Threading: Increase thread pool size, implement proper task queuing, monitor thread pool metrics, and add backpressure handling.")
            elif "circuit breaker" in msg:
                suges.append(
                    "🔧 Resilience: Review circuit breaker thresholds, implement proper fallback mechanisms, and monitor service health metrics.")

        # Serialization and data format errors
        elif any(keyword in msg for keyword in ["json", "serialization", "parsing"]):
            if "json" in msg:
                suges.append(
                    "🔧 Data Processing: Implement robust JSON validation, add proper error handling for malformed data, and use schema validation.")
            elif "serialization" in msg:
                suges.append(
                    "🔧 Data Processing: Ensure all objects implement Serializable, review serialization compatibility, and consider using alternative serialization formats.")

        # Validation and constraint errors
        elif any(keyword in msg for keyword in ["validation", "constraint", "integrity"]):
            suges.append(
                "🔧 Data Validation: Implement comprehensive input validation, review database constraints, and add proper error handling for validation failures.")

        # Container and deployment errors
        elif any(keyword in msg for keyword in ["docker", "container"]):
            suges.append(
                "🔧 Deployment: Check Docker configuration, verify resource limits, review container logs, and ensure proper image dependencies.")

        # WebSocket and real-time communication errors
        elif any(keyword in msg for keyword in ["websocket", "broken pipe"]):
            suges.append(
                "🔧 Real-time Communication: Implement WebSocket reconnection logic, add connection health monitoring, and handle network interruptions gracefully.")

        # GraphQL and API errors
        elif any(keyword in msg for keyword in ["graphql", "query execution"]):
            suges.append(
                "🔧 API: Optimize GraphQL query performance, implement query complexity analysis, add proper timeout handling, and monitor query execution times.")

        # Batch processing errors
        elif any(keyword in msg for keyword in ["batch", "job failed"]):
            suges.append(
                "🔧 Batch Processing: Implement proper error handling in batch jobs, add data validation, implement retry mechanisms, and monitor job execution status.")

        # Lock and concurrency errors
        elif any(keyword in msg for keyword in ["lock", "distributed lock"]):
            suges.append(
                "🔧 Concurrency: Review lock timeout settings, implement proper lock release mechanisms, add lock monitoring, and consider lock-free alternatives where possible.")

        # Cloud service errors (AWS, etc.)
        elif any(keyword in msg for keyword in ["s3", "access denied", "aws"]):
            suges.append(
                "🔧 Cloud Services: Verify IAM permissions, check AWS credentials, review bucket policies, and implement proper error handling for cloud service failures.")

        # Generic error handling
        else:
            suges.append(
                f"🔧 General: Review {component} component logs, implement proper error handling, add monitoring and alerting for this error type.")

    # Process WARNING level issues
    for warning in errors_and_warnings.get("warnings", []):
        msg = safe_str(warning.get("message")).lower()
        component = safe_str(warning.get("component")).lower()

        # System resource warnings
        if any(keyword in msg for keyword in ["disk usage", "disk"]):
            suges.append(
                "💡 System Resources: Set up automated log rotation, clean old temporary files, monitor disk usage proactively, and implement disk space alerts.")
        elif any(keyword in msg for keyword in ["memory usage", "memory"]):
            suges.append(
                "💡 System Resources: Monitor memory usage patterns, implement memory optimization, consider increasing available memory, and add memory usage alerts.")
        elif any(keyword in msg for keyword in ["cpu usage", "cpu"]):
            suges.append(
                "💡 System Resources: Optimize CPU-intensive operations, implement load balancing, monitor CPU usage trends, and consider scaling resources.")

        # Performance warnings
        elif any(keyword in msg for keyword in ["response time", "performance", "degradation"]):
            suges.append(
                "💡 Performance: Optimize slow operations, implement caching strategies, review database query performance, and add performance monitoring.")
        elif any(keyword in msg for keyword in ["latency", "network"]):
            suges.append(
                "💡 Performance: Optimize network calls, implement connection pooling, review network infrastructure, and add latency monitoring.")

        # Queue and throughput warnings
        elif any(keyword in msg for keyword in ["queue", "limit", "approaching"]):
            suges.append(
                "💡 Capacity: Monitor queue sizes, implement auto-scaling, optimize message processing, and add queue depth alerts.")
        elif any(keyword in msg for keyword in ["rate limit", "requests"]):
            suges.append(
                "💡 Capacity: Review rate limiting policies, implement request throttling, consider increasing limits, and add rate limiting alerts.")

        # Cache performance warnings
        elif any(keyword in msg for keyword in ["cache", "hit ratio"]):
            suges.append(
                "💡 Caching: Optimize cache configuration, review cache key strategies, implement cache warming, and monitor cache performance metrics.")

        # Scheduled task warnings
        elif any(keyword in msg for keyword in ["scheduled", "task", "delayed"]):
            suges.append(
                "💡 Scheduling: Review task scheduling configuration, optimize task execution time, implement task monitoring, and add scheduling alerts.")

        # Garbage collection warnings
        elif any(keyword in msg for keyword in ["garbage collection", "gc"]):
            suges.append(
                "💡 JVM: Tune garbage collection parameters, review heap configuration, implement GC monitoring, and optimize memory allocation patterns.")

        # Connection pool warnings
        elif any(keyword in msg for keyword in ["connection pool", "connections available"]):
            suges.append(
                "💡 Connection Management: Increase connection pool size, implement connection monitoring, optimize connection usage, and add pool health checks.")

        # Generic warning handling
        else:
            suges.append(
                f"💡 Monitoring: Monitor {component} component closely, implement alerting for this warning type, and review system metrics regularly.")

    # Handle empty results
    if not suges:
        if errors_and_warnings.get("errors") or errors_and_warnings.get("warnings"):
            suges.append(
                "✅ Review the identified issues above and implement appropriate monitoring and logging practices.")
        else:
            suges.append(
                "✅ No critical issues found. Continue monitoring system health and maintain current operational practices.")

    # Add general recommendations if there are multiple issues
    error_count = len(errors_and_warnings.get("errors", []))
    warning_count = len(errors_and_warnings.get("warnings", []))

    if error_count > 3:
        suges.append(
            "🚨 High Error Volume: Consider implementing comprehensive error tracking, alerting systems, and automated incident response procedures.")

    if warning_count > 5:
        suges.append(
            "⚠️ Multiple Warnings: Review system capacity planning, implement proactive monitoring, and consider performance optimization initiatives.")

    return suges


if __name__ == "__main__":
    mcp.run(transport='stdio')