Add privacy-safe configurable OpenTelemetry tracing (#24)

* Add privacy-safe OpenTelemetry tracing

* Make MCP access telemetry configurable

* Fix cryptography security advisory

Author: rodion-m

pyproject.toml

@@ -13,6 +13,7 @@ dependencies = [
     "opentelemetry-sdk==1.43.0",
     "opentelemetry-exporter-otlp-proto-http==1.43.0",
     "opentelemetry-instrumentation-httpx==0.64b0",
+    "opentelemetry-instrumentation-starlette==0.64b0",
     "opentelemetry-semantic-conventions==0.64b0",
 ]
 
@@ -45,6 +46,10 @@ fallback_version = "3.0.2"
 # Older versions silently ignore the whole [tool.uv] section on parse error.
 required-version = "==0.11.28"
 exclude-newer = "7 days"
+# Security exception: cryptography 50.0.0 fixes GHSA-g6cj-pr64-35w5 and was
+# released inside the normal quarantine window. Keep the exception package-
+# scoped so all unrelated dependencies remain subject to the seven-day delay.
+exclude-newer-package = { cryptography = "2026-08-01T00:00:00Z" }
 # Security minimums are older than the global seven-day quarantine. Exact direct
 # pins plus exclude-newer keep every future lock update inside the same policy.
 constraint-dependencies = [

src/codealive_mcp_server.py

@@ -53,6 +53,13 @@ def _package_version() -> str:
         return "unknown"
 
 
+def _environment_flag(name: str, *, default: bool) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    return value.strip().lower() not in {"false", "0", "no", "off"}
+
+
 # Initialize FastMCP server with lifespan and enhanced system instructions
 mcp = FastMCP(
     name="CodeAlive MCP Server",
@@ -301,6 +308,13 @@ def main():
             allowed_origins=allowed_origins or None,
             uvicorn_config={
                 "forwarded_allow_ips": "*",
+                # Access logs scale linearly with unauthenticated traffic. Keep
+                # the upstream default for self-hosted operators, while allowing
+                # hardened deployments to rely on sampled traces and safe events.
+                "access_log": _environment_flag(
+                    "CODEALIVE_MCP_ACCESS_LOG_ENABLED",
+                    default=True,
+                ),
             },
         )
     else:

src/core/logging.py

@@ -6,6 +6,7 @@
 """
 
 import logging
+import os
 import sys
 import uuid
 from typing import Any, Dict, List, Optional, Tuple, Union
@@ -115,13 +116,23 @@ def setup_logging(debug: bool = False) -> None:
     # Intercept stdlib logging
     logging.basicConfig(handlers=[_InterceptHandler()], level=0, force=True)
 
+    # FastMCP validation and exception records can contain rejected argument
+    # values. In production its logger is disabled in favor of the privacy-safe
+    # CodeAlive middleware logs. Stop propagation explicitly because FastMCP's
+    # disabled setting otherwise leaves child loggers attached to the root.
+    fastmcp_logger = logging.getLogger("fastmcp")
+    if os.environ.get("FASTMCP_LOG_ENABLED", "").lower() in {"false", "0", "no"}:
+        fastmcp_logger.handlers.clear()
+        fastmcp_logger.addHandler(logging.NullHandler())
+        fastmcp_logger.propagate = False
+    else:
+        fastmcp_logger.propagate = True
+
     logger.info("Logging initialized at {level} level", level=_current_level)
 
 
 def setup_debug_logging() -> bool:
     """Backward-compatible helper: enable debug logging if ``DEBUG_MODE`` env is set."""
-    import os
-
     if os.environ.get("DEBUG_MODE", "").lower() in ["true", "1", "yes"]:
         setup_logging(debug=True)
         return True

src/core/observability.py

@@ -1,47 +1,150 @@
 """OpenTelemetry setup for CodeAlive MCP server.
 
-Initialises a ``TracerProvider`` with an OTLP/HTTP exporter when the
-``OTEL_EXPORTER_OTLP_ENDPOINT`` env var is set.  Otherwise tracing is
-configured as a no-op so the rest of the code can call ``trace.get_tracer()``
-unconditionally.
+Initialises a ``TracerProvider`` with an OTLP/HTTP exporter when either the
+generic or traces-specific OTLP endpoint is configured. Otherwise tracing is
+configured without an exporter so the rest of the code can call
+``trace.get_tracer()`` unconditionally.
 
-HTTPX client instrumentation is always enabled so outbound HTTP calls
-automatically get ``traceparent`` headers injected.
+Starlette and HTTPX instrumentation connect inbound MCP requests to outbound
+CodeAlive API calls without recording request or response bodies.
 """
 
 import atexit
 import os
+from collections.abc import Sequence
 
 from loguru import logger
 from opentelemetry import trace
 from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
+from opentelemetry.instrumentation.starlette import StarletteInstrumentor
 from opentelemetry.sdk.resources import Resource
-from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace import Event, ReadableSpan, TracerProvider
+from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
+from opentelemetry.trace import Status
 
 _SERVICE_NAME = "codealive-mcp"
 
+_SENSITIVE_ATTRIBUTE_PREFIXES = (
+    "enduser.",
+    "http.request.header.",
+    "http.response.header.",
+)
+_SENSITIVE_ATTRIBUTES = {
+    "client.address",
+    "http.url",
+    "mcp.session.id",
+    "mcp.resource.uri",
+    "network.peer.address",
+    "url.full",
+    "url.query",
+    "user_agent.original",
+}
+
+
+class _SanitizingSpanExporter(SpanExporter):
+    """Remove client data added by framework auto-instrumentation before export."""
+
+    def __init__(self, delegate: SpanExporter) -> None:
+        self._delegate = delegate
+
+    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
+        return self._delegate.export(tuple(self._sanitize(span) for span in spans))
+
+    def shutdown(self) -> None:
+        self._delegate.shutdown()
+
+    def force_flush(self, timeout_millis: int = 30000) -> bool:
+        return self._delegate.force_flush(timeout_millis)
+
+    @staticmethod
+    def _sanitize(span: ReadableSpan) -> ReadableSpan:
+        attributes = {
+            key: value
+            for key, value in span.attributes.items()
+            if key not in _SENSITIVE_ATTRIBUTES
+            and not key.startswith(_SENSITIVE_ATTRIBUTE_PREFIXES)
+        }
+        events = tuple(
+            Event(
+                event.name,
+                {"exception.type": event.attributes["exception.type"]}
+                if event.name == "exception"
+                and event.attributes
+                and "exception.type" in event.attributes
+                else {},
+                event.timestamp,
+            )
+            for event in span.events
+        )
+        return ReadableSpan(
+            name=span.name,
+            context=span.context,
+            parent=span.parent,
+            resource=span.resource,
+            attributes=attributes,
+            events=events,
+            links=span.links,
+            kind=span.kind,
+            status=Status(span.status.status_code),
+            start_time=span.start_time,
+            end_time=span.end_time,
+            instrumentation_scope=span.instrumentation_scope,
+        )
+
+
+def _resource_attributes() -> dict[str, str]:
+    """Build low-cardinality resource identity from deployment metadata only."""
+    attributes = {
+        "service.name": os.environ.get("OTEL_SERVICE_NAME", _SERVICE_NAME),
+        "k8s.container.name": "mcp-server",
+    }
+    optional_attributes = {
+        "service.version": os.environ.get("CODEALIVE_MCP_VERSION"),
+        "service.instance.id": os.environ.get("POD_NAME")
+        or os.environ.get("HOSTNAME"),
+        "deployment.environment.name": os.environ.get("DEPLOYMENT_ENVIRONMENT")
+        or os.environ.get("ENVIRONMENT"),
+        "k8s.namespace.name": os.environ.get("POD_NAMESPACE"),
+        "k8s.pod.name": os.environ.get("POD_NAME"),
+        "k8s.node.name": os.environ.get("NODE_NAME"),
+    }
+    attributes.update(
+        {key: value for key, value in optional_attributes.items() if value}
+    )
+    return attributes
+
 
 def init_tracing() -> None:
     """Bootstrap OpenTelemetry tracing.
 
-    * If ``OTEL_EXPORTER_OTLP_ENDPOINT`` is set, traces are exported via
-      OTLP/HTTP (protobuf) to that endpoint.
-    * Otherwise a no-op provider is configured (zero overhead).
+    * If a generic or traces-specific OTLP endpoint is set, traces are exported
+      via OTLP/HTTP (protobuf). The exporter reads the standard OTel env vars so
+      it can apply the correct ``/v1/traces`` path semantics.
+    * Otherwise a provider without an exporter is configured (no network I/O).
     * HTTPX client instrumentation is always enabled so that ``traceparent``
       propagates to the CodeAlive backend regardless of whether traces are
       exported.
     """
-    otlp_endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT")
+    otlp_endpoint = os.environ.get(
+        "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"
+    ) or os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT")
 
-    resource = Resource.create({"service.name": _SERVICE_NAME})
+    resource = Resource.create(_resource_attributes())
 
     if otlp_endpoint:
-        from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+        from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
+            OTLPSpanExporter,
+        )
         from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
-        exporter = OTLPSpanExporter(endpoint=otlp_endpoint)
+        # Do not pass the endpoint explicitly. The exporter distinguishes the
+        # signal-specific URL from the generic base URL and appends /v1/traces
+        # only where the OTel environment-variable contract requires it.
+        exporter = OTLPSpanExporter()
         provider = TracerProvider(resource=resource)
-        provider.add_span_processor(BatchSpanProcessor(exporter))
+        provider.add_span_processor(
+            BatchSpanProcessor(_SanitizingSpanExporter(exporter))
+        )
         trace.set_tracer_provider(provider)
 
         logger.info(
@@ -59,5 +162,8 @@ def init_tracing() -> None:
     # Flush pending spans on process exit
     atexit.register(provider.shutdown)
 
-    # Auto-instrument httpx so outbound requests carry traceparent
+    # Instrument before FastMCP creates its Starlette app. Neither integration
+    # captures bodies by default; health endpoints are excluded through the
+    # standard OTEL_PYTHON_STARLETTE_EXCLUDED_URLS deployment setting.
+    StarletteInstrumentor().instrument()
     HTTPXClientInstrumentor().instrument()

src/middleware/observability_middleware.py

@@ -4,7 +4,7 @@
   - ``gen_ai.operation.name``  = ``"execute_tool"``
   - ``gen_ai.tool.name``       = tool name
   - ``mcp.tool.name``          = tool name (MCP-specific alias)
-  - ``mcp.method``             = ``"tools/call"``
+  - ``mcp.method.name``        = ``"tools/call"``
 
 The middleware also injects ``trace_id`` into loguru context via
 ``logger.contextualize`` so that every log emitted during the tool
@@ -60,7 +60,27 @@ def _extract_tool_arguments(context: "MiddlewareContext") -> dict[str, Any]:
 
 
 class ObservabilityMiddleware(Middleware):
-    """Wrap each ``tools/call`` in an OTel span and log its outcome."""
+    """Trace MCP requests and nested tool execution without recording payloads."""
+
+    async def on_request(self, context: "MiddlewareContext", call_next: "CallNext"):
+        method = context.method or "unknown"
+        with _tracer.start_as_current_span(
+            f"mcp {method}",
+            record_exception=False,
+            set_status_on_exception=False,
+            attributes={"mcp.method.name": method},
+        ) as span:
+            try:
+                result = await call_next(context)
+            except Exception as exc:
+                error_type = type(exc).__name__
+                span.set_attribute("error.type", error_type)
+                span.set_status(StatusCode.ERROR, error_type)
+                span.add_event("exception", {"exception.type": error_type})
+                raise
+
+            span.set_status(StatusCode.OK)
+            return result
 
     async def on_call_tool(self, context: "MiddlewareContext", call_next: "CallNext"):
         tool_name = getattr(context.message, "name", "unknown")
@@ -75,7 +95,7 @@ async def on_call_tool(self, context: "MiddlewareContext", call_next: "CallNext"
                 "gen_ai.operation.name": "execute_tool",
                 "gen_ai.tool.name": tool_name,
                 "mcp.tool.name": tool_name,
-                "mcp.method": "tools/call",
+                "mcp.method.name": "tools/call",
             },
         ) as span:
             # Inject trace_id into loguru so every log inside the tool carries it

src/tests/test_http_transport_security.py

@@ -97,6 +97,20 @@ def test_http_main_enables_guard_and_reads_environment_allowlists(monkeypatch):
     assert options["host_origin_protection"] is True
     assert options["allowed_hosts"] == ["mcp.codealive.ai", "codealive-mcp-server"]
     assert options["allowed_origins"] == ["https://mcp.codealive.ai"]
+    assert options["uvicorn_config"]["access_log"] is True
+
+
+def test_http_main_can_disable_per_request_access_logs(monkeypatch):
+    run = MagicMock()
+    monkeypatch.setattr(server.mcp, "run", run)
+    monkeypatch.setattr(server, "setup_logging", MagicMock())
+    monkeypatch.setattr(server, "init_tracing", MagicMock())
+    monkeypatch.setenv("CODEALIVE_MCP_ACCESS_LOG_ENABLED", "false")
+    monkeypatch.setattr(sys, "argv", ["codealive-mcp", "--transport", "http"])
+
+    server.main()
+
+    assert run.call_args.kwargs["uvicorn_config"]["access_log"] is False
 
 
 def test_http_main_fails_closed_when_oauth_exchange_secret_is_missing(monkeypatch):

src/tests/test_logging.py

@@ -2,6 +2,7 @@
 
 import io
 import json
+import logging
 import sys
 from unittest.mock import MagicMock
 
@@ -106,6 +107,35 @@ def test_setup_logging_sets_level(self):
 
         logger.remove(handler_id)
 
+    def test_fastmcp_logs_do_not_propagate_when_disabled(self, monkeypatch, capsys):
+        monkeypatch.setenv("FASTMCP_LOG_ENABLED", "false")
+        sink = io.StringIO()
+        setup_logging()
+        logger.remove()
+        handler_id = logger.add(sink, level="DEBUG", serialize=True)
+
+        logging.getLogger("fastmcp.server.server").warning(
+            "Invalid arguments: secret query text"
+        )
+
+        assert "secret query text" not in sink.getvalue()
+        assert "secret query text" not in capsys.readouterr().err
+        logger.remove(handler_id)
+
+    def test_fastmcp_logs_propagate_by_default_for_self_hosted(self, monkeypatch):
+        monkeypatch.delenv("FASTMCP_LOG_ENABLED", raising=False)
+        sink = io.StringIO()
+        setup_logging()
+        logger.remove()
+        handler_id = logger.add(sink, level="DEBUG", serialize=True)
+
+        logging.getLogger("fastmcp.server.server").warning(
+            "Self-hosted framework diagnostic"
+        )
+
+        assert "Self-hosted framework diagnostic" in sink.getvalue()
+        logger.remove(handler_id)
+
     def test_setup_debug_logging_env_var(self, monkeypatch):
         monkeypatch.setenv("DEBUG_MODE", "true")
         assert setup_debug_logging() is True