diff --git a/main.py b/main.py index 2fcab4c..c70a6c9 100644 --- a/main.py +++ b/main.py @@ -15,7 +15,7 @@ from google import genai from google.genai import types as genai_types from mcp.server.fastmcp import Context, FastMCP -from utils import Settings, _args, log_structured_entry +from utils import Settings, _args, cfg, log_structured_entry HTTP_TOO_MANY_REQUESTS = 429 HTTP_SERVER_ERROR = 500 @@ -624,8 +624,6 @@ async def lifespan(_server: FastMCP) -> AsyncIterator[AppContext]: log_structured_entry("MCP server lifespan ending", "INFO") -cfg = Settings.model_validate({}) - mcp = FastMCP( "knowledge-search", host=_args.host, diff --git a/utils/__init__.py b/utils/__init__.py index dcbcb26..17f1feb 100644 --- a/utils/__init__.py +++ b/utils/__init__.py @@ -1,4 +1,4 @@ -from .config import Settings, _args +from .config import Settings, _args, cfg from .logging_setup import log_structured_entry -__all__ = ['Settings', '_args', 'log_structured_entry'] +__all__ = ['Settings', '_args', 'cfg', 'log_structured_entry'] diff --git a/utils/config.py b/utils/config.py index 07b5c01..04f81d4 100644 --- a/utils/config.py +++ b/utils/config.py @@ -35,6 +35,8 @@ class Settings(BaseSettings): endpoint_domain: str embedding_model: str = "gemini-embedding-001" search_limit: int = 10 + log_name: str = "va_agent_evaluation_logs" + log_level: str = "INFO" @classmethod def settings_customise_sources( @@ -52,3 +54,7 @@ class Settings(BaseSettings): YamlConfigSettingsSource(settings_cls), file_secret_settings, ) + + +# Singleton instance of Settings +cfg = Settings.model_validate({}) diff --git a/utils/logging_setup.py b/utils/logging_setup.py index 686da1f..fb1519a 100644 --- a/utils/logging_setup.py +++ b/utils/logging_setup.py @@ -9,26 +9,24 @@ from typing import Optional, Dict, Literal import google.cloud.logging from google.cloud.logging.handlers import CloudLoggingHandler -from .config import Settings +from .config import cfg def _setup_logger() -> logging.Logger: """Create or return the singleton evaluation logger.""" - log_name = "va_agent-evaluation-logs" - logger = logging.getLogger(log_name) - cfg = Settings.model_validate({}) + logger = logging.getLogger(cfg.log_name) if any(isinstance(h, CloudLoggingHandler) for h in logger.handlers): return logger try: client = google.cloud.logging.Client(project=cfg.project_id) - handler = CloudLoggingHandler(client, name=log_name) # async transport + handler = CloudLoggingHandler(client, name=cfg.log_name) # async transport logger.addHandler(handler) - logger.setLevel(logging.INFO) + 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=logging.INFO) - logger = logging.getLogger(log_name) + 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) return logger @@ -43,7 +41,7 @@ def log_structured_entry(message: str, severity: Literal["INFO", "WARNING", "ERR Args: message: Short label for the row (e.g., "Final agent turn"). - severity: "INFO" | "WARNING" | "ERROR" etc. + severity: "INFO" | "WARNING" | "ERROR" custom_log: A dict with your structured payload. """ level = getattr(logging, severity.upper(), logging.INFO)