Migrate to package

src/knowledge_search_mcp/logging.py

@@ -0,0 +1,56 @@
+"""
+Centralized Cloud Logging setup.
+Uses CloudLoggingHandler (background thread) so logging does not add latency
+"""
+
+import logging
+from typing import Optional, Dict, Literal
+
+import google.cloud.logging
+from google.cloud.logging.handlers import CloudLoggingHandler
+
+from .config import get_config
+
+
+_eval_log: logging.Logger | None = None
+
+
+def _get_logger() -> logging.Logger:
+    """Get or create the singleton evaluation logger."""
+    global _eval_log
+    if _eval_log is not None:
+        return _eval_log
+
+    cfg = get_config()
+    logger = logging.getLogger(cfg.log_name)
+    if any(isinstance(h, CloudLoggingHandler) for h in logger.handlers):
+        _eval_log = logger
+        return logger
+
+    try:
+        client = google.cloud.logging.Client(project=cfg.project_id)
+        handler = CloudLoggingHandler(client, name=cfg.log_name)  # async transport
+        logger.addHandler(handler)
+        logger.setLevel(getattr(logging, cfg.log_level.upper()))
+    except Exception as e:
+        # Fallback to console if Cloud Logging is unavailable (local dev)
+        logging.basicConfig(level=getattr(logging, cfg.log_level.upper()))
+        logger = logging.getLogger(cfg.log_name)
+        logger.warning("Cloud Logging setup failed; using console. Error: %s", e)
+
+    _eval_log = logger
+    return logger
+
+
+def log_structured_entry(message: str, severity: Literal["INFO", "WARNING", "ERROR"], custom_log: Optional[Dict] = None) -> None:
+    """
+    Emit a JSON-structured log row.
+
+    Args:
+        message: Short label for the row (e.g., "Final agent turn").
+        severity: "INFO" | "WARNING" | "ERROR"
+        custom_log: A dict with your structured payload.
+    """
+    level = getattr(logging, severity.upper(), logging.INFO)
+    logger = _get_logger()
+    logger.log(level, message, extra={"json_fields": {"message": message, "custom": custom_log or {}}})