Add structured logging (JSONL) + optional Prometheus /metrics

Two independent outputs from the same instrumentation events, no overlap:

1. Logs to stderr. Text formatter (default) for dev; JSON Lines
   (ESET_MCP_LOG_FORMAT=json) for prod log shippers. Every layer emits a
   typed 'event' record with low-cardinality fields - tool_call,
   token_refresh, http_request_complete, http_request_retry,
   server_starting, onprem_default.

2. Prometheus metrics at /metrics (opt-in via ESET_MCP_METRICS_ENABLED=true).
   Mounted on the same uvicorn server but bypasses the basic-auth
   middleware - scrapers can't speak Basic, and /metrics carries no
   secrets. Protect at the network layer.

   Counters:  esetmcp_tool_calls_total{tool,deployment,status}
              esetmcp_token_refresh_total{deployment,reason}
              esetmcp_http_retries_total{deployment,status}
              esetmcp_responses_capped_total{tool}
   Histograms: esetmcp_tool_duration_seconds{tool,deployment}
               esetmcp_response_bytes{tool,deployment}
   Gauge:      esetmcp_client_pool_size

   prometheus_client is an OPTIONAL dependency (pip install eset-mcp[metrics]).
   When missing, all metric helpers are no-ops and /metrics returns a 503
   with a hint.

Defensive secrets gate (the whole point):

The log_event() helper applies a hard-coded deny-list before any field
reaches a formatter. Anything whose key contains 'password', 'secret',
'token', 'authorization', 'cookie', 'api_key', 'bearer', 'credentials',
or 'cf-access-client-*' is dropped entirely - no '****' redaction, no
'value present but hidden' marker. Even if a future caller accidentally
passes Authorization=..., it never appears in any log artefact under any
circumstance.

Path templates from the OpenAPI spec are logged ('/v1/devices/{deviceUuid}')
but the SUBSTITUTED path (with the actual UUID) is never logged - that
would leak tenant data through request-id correlation.

Username (api@firma.tld) is logged - useful for per-tenant debugging,
not a secret. NEVER appears as a metric label (would explode cardinality).

Test coverage: tests/test_observability.py adds 36 unit tests:
- Sensitive-key matcher against 15 representative keys
- _scrub() removes sensitive keys, keeps safe ones
- log_event() drops sensitive fields end-to-end (asserts no leakage in
  the serialised line text, not just the parsed record)
- JSON formatter produces parseable lines with ISO-Z timestamps
- Text formatter appends scrubbed extras
- configure_logging() is idempotent (no handler stacking)
- All metric helpers callable; /metrics returns Prometheus content-type
  and contains the expected metric names/labels
- /metrics ignores lifespan scope without crashing

Snyk Code: 0 issues in eset_mcp/observability/. Ruff: clean. Tests:
136/136 unit + 11 RO smoke against real ESET passed.

Author: maciekaz

.env.example

@@ -72,6 +72,21 @@ ESET_MCP_RESPONSE_BYTES_MAX=100000
 # ─── Optional logging level ────────────────────────────────────────────────
 ESET_LOG_LEVEL=INFO
 
+# ─── Observability (logs + metrics) ────────────────────────────────────────
+# Log output format:
+#   text - human-readable single line per record (default, good for dev)
+#   json - JSON Lines, one self-contained object per line (recommended in
+#          prod - Loki/Vector/Fluent Bit/Datadog all parse it out of the box)
+ESET_MCP_LOG_FORMAT=text
+
+# Prometheus /metrics endpoint (HTTP transport only). Opt-in. When true,
+# the metrics route is mounted on the same uvicorn server but bypasses
+# basic-auth (Prometheus scrapers don't speak Basic auth; /metrics carries
+# no secrets). Protect at the network/firewall layer instead.
+# Requires `pip install eset-mcp[metrics]` for the prometheus_client dep.
+ESET_MCP_METRICS_ENABLED=false
+ESET_MCP_METRICS_PATH=/metrics
+
 # ─── Production deployment (docker compose --profile prod) ─────────────────
 # Public domain Caddy will request a Let's Encrypt cert for. Must point at
 # the host running docker-compose. Only used by the `prod` profile.

README.md

@@ -133,6 +133,24 @@ HTTP errors are mapped to readable hints: 403 → check Permission Sets in
 ESET PROTECT Hub; 401 → server refreshes the token automatically; 429 →
 back off; 5xx → retry shortly.
 
+### Observability (logs + Prometheus)
+
+- **Structured logs** to stderr. Text (default) for dev, JSON Lines for
+  prod log shippers via `ESET_MCP_LOG_FORMAT=json`. Every tool call,
+  token refresh, HTTP retry and pool eviction emits a typed `event`
+  record with low-cardinality fields (tool, deployment, status,
+  duration_ms, response_bytes, ...).
+- **Prometheus metrics** at an opt-in `/metrics` endpoint
+  (`ESET_MCP_METRICS_ENABLED=true`, requires
+  `pip install eset-mcp[metrics]`). Counters for tool calls, token
+  refreshes, HTTP retries, cap hits; histograms for tool duration and
+  response sizes; gauge for client pool size.
+- **What never enters logs or metrics**: passwords, `Authorization`
+  headers, CF Access secrets, request/response bodies, query strings,
+  substituted path parameters (which can leak UUIDs). A defensive
+  deny-list in the logger strips known-sensitive keys before any
+  formatter sees them.
+
 ---
 
 ## Security
@@ -273,6 +291,9 @@ All settings live in `.env`. Required fields are marked in
 | `ESET_MCP_HTTP_PORT`           | `8765`        | HTTP port                                                     |
 | `ESET_MCP_RESPONSE_BYTES_MAX`  | `100000`      | Per-call response byte cap; `0` disables                      |
 | `ESET_LOG_LEVEL`               | `INFO`        | `DEBUG` / `INFO` / `WARNING` / `ERROR`                        |
+| `ESET_MCP_LOG_FORMAT`          | `text`        | `text` (dev, human-readable) or `json` (prod log shippers)    |
+| `ESET_MCP_METRICS_ENABLED`     | `false`       | Mount Prometheus `/metrics`; requires `eset-mcp[metrics]`     |
+| `ESET_MCP_METRICS_PATH`        | `/metrics`    | Where to mount the metrics endpoint                           |
 | `ESET_DEPLOYMENT`              | `cloud`       | `cloud` (ESET Connect) or `onprem` (customer-hosted PROTECT)  |
 | `ESET_ONPREM_SERVER_URL`       | -             | `https://host[:port]` of the on-prem console (req. in env+onprem) |
 | `ESET_ONPREM_VERIFY_SSL`       | `true`        | Set `false` for on-prem consoles with self-signed certs       |

eset_mcp/__main__.py

@@ -1,32 +1,36 @@
-"""Server entrypoint — stdio or Streamable HTTP transport."""
+"""Server entrypoint - stdio or Streamable HTTP transport."""
 from __future__ import annotations
 
 import asyncio
 import logging
-import sys
 
 from .client_pool import ClientPool
 from .config import Settings
 from .credentials import BasicAuthCredentialResolver, EnvCredentialResolver
+from .observability import configure_logging, log_event
 from .server import build_server
 
 
 def main() -> None:
     settings = Settings.from_env()
-    logging.basicConfig(
-        level=settings.log_level,
-        format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
-        stream=sys.stderr,  # MCP stdio uses stdout for JSON-RPC — logs MUST go to stderr.
-    )
+    configure_logging(level=settings.log_level, fmt=settings.log_format)
     log = logging.getLogger("eset_mcp")
-    log.info(
-        "ESET-MCP starting — mode=%s region=%s transport=%s auth=%s deployment=%s",
-        settings.mode, settings.region, settings.transport, settings.auth_mode, settings.deployment,
+    log_event(
+        log, "server_starting",
+        mode=settings.mode,
+        region=settings.region,
+        transport=settings.transport,
+        auth_mode=settings.auth_mode,
+        deployment=settings.deployment,
+        metrics_enabled=settings.metrics_enabled,
+        log_format=settings.log_format,
     )
     if settings.deployment == "onprem":
-        log.info("ESET-MCP on-prem default URL: %s (verify_ssl=%s)",
-                 settings.onprem_server_url or "(none — must be supplied per request)",
-                 settings.onprem_verify_ssl)
+        log_event(
+            log, "onprem_default",
+            server_url=settings.onprem_server_url or "(none - must be supplied per request)",
+            verify_ssl=settings.onprem_verify_ssl,
+        )
 
     asyncio.run(_run(settings))
 
@@ -74,15 +78,34 @@ async def _lifespan(_app):
         async with manager.run():
             yield
 
-    app = Starlette(
-        routes=[Mount("/mcp", app=manager.handle_request)],
-        lifespan=_lifespan,
-    )
-
-    # Basic-auth middleware sits in front of everything in `basic` mode.
+    routes = [Mount("/mcp", app=manager.handle_request)]
+    # Optional /metrics endpoint - mounted BEFORE the basic-auth middleware
+    # is added so that Prometheus scrapers don't need to send Basic auth
+    # to pull metrics. The endpoint carries no secrets and is intended to
+    # be protected at the network layer (private subnet / VPN / Caddy ACL).
+    if settings.metrics_enabled:
+        from .observability.metrics import metrics_asgi_app, metrics_available
+        routes.append(Mount(settings.metrics_path, app=metrics_asgi_app()))
+        if not metrics_available():
+            logging.getLogger("eset_mcp").warning(
+                "ESET_MCP_METRICS_ENABLED=true but prometheus_client is not "
+                "installed - %s will return 503. Install via "
+                "'pip install eset-mcp[metrics]'.",
+                settings.metrics_path,
+            )
+
+    app = Starlette(routes=routes, lifespan=_lifespan)
+
+    # Basic-auth middleware sits in front of everything in `basic` mode -
+    # EXCEPT the /metrics route, which was added above. We skip the
+    # middleware for that path so scrapers don't get 401'd.
     if settings.auth_mode == "basic":
         from .middleware import BasicAuthCredentialsMiddleware
-        app.add_middleware(BasicAuthCredentialsMiddleware, settings=settings)
+        app.add_middleware(
+            BasicAuthCredentialsMiddleware,
+            settings=settings,
+            skip_paths=(settings.metrics_path,) if settings.metrics_enabled else (),
+        )
 
     config = uvicorn.Config(
         app, host=settings.http_host, port=settings.http_port, log_level=settings.log_level.lower()

eset_mcp/auth.py

@@ -20,6 +20,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import time
 from dataclasses import dataclass
 from typing import Protocol
@@ -28,8 +29,11 @@
 
 from .credentials import Credentials
 from .errors import map_http_error
+from .observability import inc_token_refresh, log_event
 from .regions import resolve_auth_url
 
+_LOG = logging.getLogger("eset_mcp.auth")
+
 # Safety margin — refresh the token N seconds before its nominal expiry.
 # Cloud uses 5 min (1h tokens, plenty of headroom); on-prem mirrors the
 # reference TypeScript client at 60 s.
@@ -71,16 +75,29 @@ def _expired_or_expiring(self, *, margin: int | None = None) -> bool:
     async def get_access_token(self) -> str:
         async with self._lock:
             if self._expired_or_expiring():
+                reason = "initial" if self._token is None else "proactive"
                 await self._refresh_locked()
+                self._record_refresh(reason)
             assert self._token is not None
             return self._token.access_token
 
     async def force_refresh(self) -> str:
         async with self._lock:
             await self._refresh_locked()
+            self._record_refresh("forced_401")
             assert self._token is not None
             return self._token.access_token
 
+    def _record_refresh(self, reason: str) -> None:
+        """Emit metric + log entry for a successful token refresh."""
+        inc_token_refresh(deployment=self._creds.deployment, reason=reason)
+        log_event(
+            _LOG, "token_refresh",
+            deployment=self._creds.deployment,
+            reason=reason,
+            user=self._creds.user,
+        )
+
     async def _refresh_locked(self) -> None:  # pragma: no cover - abstract
         raise NotImplementedError
 

eset_mcp/client_pool.py

@@ -19,6 +19,7 @@
 from .config import Settings
 from .credentials import Credentials
 from .http_client import EsetHttpClient
+from .observability import set_pool_size
 
 # Hard cap on concurrently-cached clients. Realistic multi-tenant MCP
 # deployments fronting an ESET MSP will see well under 100 active credentials
@@ -64,12 +65,13 @@ async def get(self, creds: Credentials) -> EsetHttpClient:
             client = EsetHttpClient(creds)
             self._clients[key] = client
             await self._maybe_evict_locked()
+            set_pool_size(len(self._clients))
             return client
 
     async def _maybe_evict_locked(self) -> None:
         while len(self._clients) > self._max:
             _, evicted = self._clients.popitem(last=False)
-            # Close in background — don't block the get() that triggered eviction.
+            # Close in background - don't block the get() that triggered eviction.
             task = asyncio.create_task(evicted.aclose())
             self._eviction_tasks.add(task)
             task.add_done_callback(self._eviction_tasks.discard)
@@ -79,6 +81,7 @@ async def close(self) -> None:
         async with self._lock:
             clients = list(self._clients.values())
             self._clients.clear()
+        set_pool_size(0)
         # Best-effort close of all clients in parallel.
         await asyncio.gather(*(c.aclose() for c in clients), return_exceptions=True)
 

eset_mcp/config.py

@@ -61,6 +61,18 @@ class Settings:
     onprem_cf_access_client_id: str
     onprem_cf_access_client_secret: str
 
+    # ─── Observability ────────────────────────────────────────────────────
+    # ``text`` (default) - human-readable single line per record, ideal for
+    # local dev. ``json`` - JSON Lines, ideal for log shippers in prod.
+    log_format: str
+    # Opt-in Prometheus /metrics endpoint (HTTP transport only). When
+    # enabled, the metrics route is mounted alongside /mcp on the same
+    # uvicorn server but bypasses the basic-auth middleware (Prometheus
+    # scrapers don't speak Basic auth, and /metrics carries no secrets).
+    # Protect at the network layer instead.
+    metrics_enabled: bool
+    metrics_path: str
+
     @property
     def has_env_credentials(self) -> bool:
         return bool(self.user and self.password)
@@ -142,6 +154,12 @@ def from_env(cls, env_file: Path | None = None) -> Settings:
             ),
             onprem_cf_access_client_id=cf_id,
             onprem_cf_access_client_secret=cf_secret,
+            log_format=_parse_log_format(os.getenv("ESET_MCP_LOG_FORMAT", "text")),
+            metrics_enabled=_parse_bool(
+                os.getenv("ESET_MCP_METRICS_ENABLED", "false"),
+                "ESET_MCP_METRICS_ENABLED",
+            ),
+            metrics_path=os.getenv("ESET_MCP_METRICS_PATH", "/metrics"),
         )
 
 
@@ -191,6 +209,13 @@ def _parse_int(raw: str, var: str) -> int:
     return n
 
 
+def _parse_log_format(raw: str) -> str:
+    val = raw.split("#", 1)[0].strip().lower()
+    if val not in ("text", "json"):
+        raise RuntimeError(f"ESET_MCP_LOG_FORMAT must be 'text' or 'json', got {raw!r}")
+    return val
+
+
 def _parse_bool(raw: str, var: str) -> bool:
     cleaned = raw.split("#", 1)[0].strip().lower()
     if cleaned in ("true", "1", "yes", "on"):

eset_mcp/http_client.py

@@ -28,6 +28,7 @@
 from .auth import TokenManagerProto, make_token_manager
 from .credentials import Credentials
 from .errors import EsetApiError, map_http_error
+from .observability import inc_http_retry, log_event
 from .regions import resolve_base_url
 
 _LOG = logging.getLogger("eset_mcp.http")
@@ -119,27 +120,49 @@ async def request(
         json: Any | None = None,
         _retry_after_401: bool = True,
     ) -> dict[str, Any] | list[Any]:
-        """Execute a request, handle 202/401/429/5xx, return the body as JSON."""
+        """Execute a request, handle 202/401/429/5xx, return the body as JSON.
+
+        ``path`` is the *template* path from the OpenAPI spec (with
+        ``{placeholder}`` segments already substituted by the caller). We
+        intentionally do NOT log the substituted URL - it can reveal device
+        UUIDs and other tenant data. Logs carry only the template-shaped
+        path the caller passed plus the upstream status code.
+        """
         url = resolve_base_url(self._creds, service) + path
         token = await self._token_mgr.get_access_token()
         headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
 
+        import time as _time
+        t0 = _time.monotonic()
         resp = await self._http.request(method, url, params=query, json=json, headers=headers)
 
-        # 401 — token expired between get_access_token() and the request; force-refresh + retry.
+        # 401 - token expired between get_access_token() and the request; force-refresh + retry.
         if resp.status_code == 401 and _retry_after_401:
-            _LOG.info("401 from %s — forcing token refresh and retrying", url)
+            inc_http_retry(deployment=self._creds.deployment, status="401")
+            log_event(
+                _LOG, "http_request_retry",
+                deployment=self._creds.deployment,
+                method=method, service=service, status=401,
+                reason="token_expired",
+            )
             await self._token_mgr.force_refresh()
             return await self.request(
                 method, service, path, query=query, json=json, _retry_after_401=False
             )
 
-        # 429 — rate limit, backoff.
+        # 429 - rate limit, backoff.
         if resp.status_code == 429:
             for attempt in range(1, _RATE_LIMIT_MAX_TRIES + 1):
                 retry_after = _parse_retry_after(resp.headers.get("Retry-After"))
                 wait_s = retry_after or _RATE_LIMIT_BASE_BACKOFF_S * (2 ** (attempt - 1))
-                _LOG.warning("429 from %s — sleeping %.1fs (try %d/%d)", url, wait_s, attempt, _RATE_LIMIT_MAX_TRIES)
+                inc_http_retry(deployment=self._creds.deployment, status="429")
+                log_event(
+                    _LOG, "http_request_retry", level=logging.WARNING,
+                    deployment=self._creds.deployment,
+                    method=method, service=service, status=429,
+                    reason="rate_limited", wait_s=round(wait_s, 2),
+                    attempt=attempt, max_attempts=_RATE_LIMIT_MAX_TRIES,
+                )
                 await asyncio.sleep(wait_s)
                 resp = await self._http.request(method, url, params=query, json=json, headers=headers)
                 if resp.status_code != 429:
@@ -153,13 +176,27 @@ async def request(
             return await self._poll_pending(method, url, response_id, headers, query, json)
 
         if resp.status_code >= 400:
+            log_event(
+                _LOG, "http_request_complete", level=logging.WARNING,
+                deployment=self._creds.deployment,
+                method=method, service=service, status=resp.status_code,
+                duration_ms=int((_time.monotonic() - t0) * 1000),
+                request_id=resp.headers.get("request-id"),
+            )
             raise map_http_error(
                 status=resp.status_code,
                 body=resp.text,
                 endpoint=url,
                 request_id=resp.headers.get("request-id"),
             )
 
+        log_event(
+            _LOG, "http_request_complete",
+            deployment=self._creds.deployment,
+            method=method, service=service, status=resp.status_code,
+            duration_ms=int((_time.monotonic() - t0) * 1000),
+            request_id=resp.headers.get("request-id"),
+        )
         return _parse_json(resp)
 
     async def _poll_pending(

eset_mcp/middleware.py

@@ -51,13 +51,23 @@
 
 
 class BasicAuthCredentialsMiddleware(BaseHTTPMiddleware):
-    """Require Basic auth on every HTTP request and stash creds in a ContextVar."""
+    """Require Basic auth on every HTTP request and stash creds in a ContextVar.
 
-    def __init__(self, app: ASGIApp, *, settings: Settings):
+    ``skip_paths`` lets specific routes bypass auth entirely - used for the
+    Prometheus ``/metrics`` endpoint, which carries no secrets and is
+    expected to be reachable by scrapers that can't speak Basic auth.
+    Protect those routes at the network layer instead.
+    """
+
+    def __init__(self, app: ASGIApp, *, settings: Settings, skip_paths: tuple[str, ...] = ()):
         super().__init__(app)
         self._settings = settings
+        self._skip_paths = tuple(skip_paths)
 
     async def dispatch(self, request: Request, call_next) -> Response:
+        # Routes the operator opted out of auth for (e.g. /metrics).
+        if self._skip_paths and request.url.path in self._skip_paths:
+            return await call_next(request)
         auth_header = request.headers.get("authorization", "")
         region_header = request.headers.get("x-eset-region")
         server_url_header = request.headers.get("x-eset-server-url")