backend/mcp_server/common/logging.py

from __future__ import annotations

import logging
import os
from typing import Any, Dict, Optional

logger = logging.getLogger("integrachat.mcp")
if not logger.handlers:
    handler = logging.StreamHandler()
    formatter = logging.Formatter(
        "[%(asctime)s] %(levelname)s %(name)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)

logger.setLevel(os.getenv("LOG_LEVEL", "INFO").upper())

try:
    from backend.api.storage.analytics_store import AnalyticsStore
except Exception:  # pragma: no cover - analytics storage is optional during tests
    AnalyticsStore = None  # type: ignore
    _analytics_store = None
else:
    _analytics_store = AnalyticsStore()


def log_tool_usage(
    tool_name: str,
    tenant_id: Optional[str],
    *,
    success: bool,
    latency_ms: Optional[int] = None,
    metadata: Optional[Dict[str, Any]] = None,
    error_message: Optional[str] = None,
    user_id: Optional[str] = None,
):
    log_data = {
        "tool": tool_name,
        "tenant_id": tenant_id,
        "success": success,
        "latency_ms": latency_ms,
        "user_id": user_id,
        "metadata": metadata or {},
    }
    if error_message:
        log_data["error"] = error_message

    if success:
        logger.info("tool_completed %s", log_data)
    else:
        logger.warning("tool_failed %s", log_data)

    if _analytics_store and tenant_id:
        try:
            _analytics_store.log_tool_usage(
                tenant_id=tenant_id,
                tool_name=tool_name,
                latency_ms=latency_ms,
                success=success,
                error_message=error_message,
                metadata=metadata,
                user_id=user_id,
            )
        except Exception as exc:  # pragma: no cover - analytics failures shouldn't crash tools
            logger.debug("analytics logging failed: %s", exc)


def log_rag_search_metrics(
    tenant_id: str,
    query: str,
    hits_count: int,
    avg_score: Optional[float],
    top_score: Optional[float],
    latency_ms: Optional[int] = None,
):
    if _analytics_store:
        try:
            _analytics_store.log_rag_search(
                tenant_id=tenant_id,
                query=query,
                hits_count=hits_count,
                avg_score=avg_score,
                top_score=top_score,
                latency_ms=latency_ms,
            )
        except Exception as exc:  # pragma: no cover
            logger.debug("rag analytics logging failed: %s", exc)


def log_redflag_violation(
    tenant_id: str,
    rule_id: str,
    rule_pattern: str,
    severity: str,
    matched_text: str,
    *,
    confidence: Optional[float] = None,
    message_preview: Optional[str] = None,
    user_id: Optional[str] = None,
):
    if _analytics_store:
        try:
            _analytics_store.log_redflag_violation(
                tenant_id=tenant_id,
                rule_id=rule_id,
                rule_pattern=rule_pattern,
                severity=severity,
                matched_text=matched_text,
                confidence=confidence,
                message_preview=message_preview,
                user_id=user_id,
            )
        except Exception as exc:  # pragma: no cover
            logger.debug("redflag logging failed: %s", exc)