mcp: add structured analytics logging

docs/mcp.md

@@ -84,3 +84,29 @@ The dashboard has no server-side lab filter, so the `lab` option of the
 list tools is applied to the fetched page after the request. It shrinks
 the response, not the query: `total` counts entries before filtering and
 `matched` after.
+
+## Analytics logging
+
+The server writes newline-delimited JSON analytics to stderr by default,
+without requiring `--debug`. To collect analytics in a separate file:
+
+```sh
+kci-dev mcp --transport http --log-file /var/log/kci-dev-mcp.jsonl
+```
+
+The file is opened in append mode; its parent directory must exist and be
+writable. File rotation and retention are managed externally. Other diagnostic
+messages continue to use stderr, and stdio protocol output stays on stdout.
+
+Each `tool_call` event includes a UTC `timestamp`, a generated unique `call_id`,
+`tool`, configured `instance` (or null), `outcome` (`success`, `error`, or
+`cancelled`), and `duration_ms`. Duration covers validation and tool execution.
+Invalid arguments and unknown tools count as errors. These records support
+tool usage counts, error rates, and latency percentiles. Calls are logged when
+they finish; a process killed abruptly cannot log its unfinished calls.
+
+`server_start` and `server_stop` events include the transport and instance.
+Analytics records omit arguments, results, credentials, and exception messages.
+This does not change the contents of existing diagnostic logs. When embedding
+`create_server()` in Python, configure the `kcidev.mcp.analytics` logger at INFO
+to collect the tool events through your application's logging setup.

kcidev/mcp/__init__.py

@@ -6,6 +6,7 @@
 from kcidev.api import KernelCIClient
 from kcidev.libs.common import kcidev_version
 from kcidev.mcp import tools_dashboard, tools_maestro
+from kcidev.mcp.analytics import instrument_tools
 
 SERVER_INSTRUCTIONS = """KernelCI MCP server (experimental: tools,
 parameters and response formats may change between releases).
@@ -27,4 +28,5 @@ def create_server(cfg=None, instance=None, host="127.0.0.1", port=8000):
     tools_maestro.register_tools(
         server, client, icfg.get("api"), icfg.get("pipeline"), icfg.get("token")
     )
+    instrument_tools(server, client.instance)
     return server

kcidev/mcp/analytics.py

@@ -0,0 +1,74 @@
+"""Payload-free JSON events for MCP usage and latency analytics."""
+
+import json
+import logging
+import time
+import uuid
+from contextlib import contextmanager
+from datetime import datetime, timezone
+
+import anyio
+from mcp.types import CallToolRequest
+
+logger = logging.getLogger("kcidev.mcp.analytics")
+
+
+def log_event(event, **fields):
+    logger.info(
+        json.dumps(
+            {
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "event": event,
+                **fields,
+            }
+        )
+    )
+
+
+@contextmanager
+def analytics_logging(path=None):
+    """Configure only our analytics logger; keep stdout free for MCP."""
+    handler = (
+        logging.FileHandler(path, encoding="utf-8")
+        if path is not None
+        else logging.StreamHandler()
+    )
+    previous = logger.handlers[:], logger.level, logger.propagate
+    logger.handlers = [handler]
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+    try:
+        yield
+    finally:
+        logger.handlers, logger.level, logger.propagate = previous
+        handler.close()
+
+
+def instrument_tools(server, instance):
+    # Wrap the protocol handler so schema errors and unknown tools count too.
+    # FastMCP converts tool exceptions to isError results inside this handler.
+    handlers = server._mcp_server.request_handlers
+    call_tool = handlers[CallToolRequest]
+
+    async def logged_call(request):
+        started = time.perf_counter()
+        call_id = uuid.uuid4().hex
+        outcome = "error"
+        try:
+            result = await call_tool(request)
+            outcome = "error" if result.root.isError else "success"
+            return result
+        except anyio.get_cancelled_exc_class():
+            outcome = "cancelled"
+            raise
+        finally:
+            log_event(
+                "tool_call",
+                call_id=call_id,
+                tool=request.params.name,
+                instance=instance,
+                outcome=outcome,
+                duration_ms=round((time.perf_counter() - started) * 1000, 3),
+            )
+
+    handlers[CallToolRequest] = logged_call

kcidev/subcommands/mcp.py

@@ -2,7 +2,6 @@
 # -*- coding: utf-8 -*-
 
 import contextlib
-import logging
 import sys
 
 import click
@@ -41,10 +40,16 @@
 )
 @click.option("--host", default="127.0.0.1", help="Bind address for http transport")
 @click.option("--port", default=8000, type=int, help="Port for http transport")
+@click.option(
+    "--log-file",
+    type=click.Path(dir_okay=False),
+    help="Append JSON analytics to this file (default: stderr)",
+)
 @click.pass_context
-def mcp(ctx, transport, host, port):
+def mcp(ctx, transport, host, port, log_file):
     try:
         from kcidev.mcp import create_server
+        from kcidev.mcp.analytics import analytics_logging, log_event
     except ImportError:
         kci_err("MCP support is not installed, install with: pip install kci-dev[mcp]")
         raise click.Abort()
@@ -55,17 +60,22 @@ def mcp(ctx, transport, host, port):
         kci_err(f"Instance {instance} not found in config")
         raise click.Abort()
     server = create_server(cfg, instance, host=host, port=port)
-    logging.info(
-        "Starting MCP server %s",
-        "via stdio" if transport == "stdio" else f"on {host}:{port}",
-    )
     import anyio
 
-    if transport == "stdio":
-        anyio.run(_run_stdio, server)
-    else:
-        with contextlib.redirect_stdout(sys.stderr):
-            server.run(transport="streamable-http")
+    with contextlib.ExitStack() as stack:
+        try:
+            stack.enter_context(analytics_logging(log_file))
+        except OSError as exc:
+            raise click.ClickException(f"Cannot open analytics log: {exc}") from exc
+        log_event("server_start", transport=transport, instance=instance)
+        try:
+            if transport == "stdio":
+                anyio.run(_run_stdio, server)
+            else:
+                with contextlib.redirect_stdout(sys.stderr):
+                    server.run(transport="streamable-http")
+        finally:
+            log_event("server_stop", transport=transport, instance=instance)
 
 
 async def _run_stdio(server):

tests/test_mcp_analytics.py

@@ -0,0 +1,128 @@
+import json
+from unittest.mock import Mock
+
+import pytest
+
+pytest.importorskip("mcp")
+
+import anyio
+from click.testing import CliRunner
+from mcp.shared.memory import (
+    create_connected_server_and_client_session as client_session,
+)
+
+from kcidev.mcp import create_server
+from kcidev.mcp.analytics import analytics_logging, log_event
+from kcidev.subcommands import mcp
+
+
+def test_tool_analytics_covers_success_failure_and_validation(tmp_path):
+    server = create_server()
+
+    @server.tool()
+    async def example(value: int):
+        if value == 0:
+            raise ValueError("private exception text")
+        return "private result text"
+
+    path = tmp_path / "analytics.jsonl"
+
+    async def run():
+        async with client_session(server._mcp_server) as session:
+            for name, arguments, error in [
+                ("example", {"value": 1}, False),
+                ("example", {"value": 0}, True),
+                ("example", {"value": "private argument text"}, True),
+                ("missing_tool", {}, True),
+            ]:
+                result = await session.call_tool(name, arguments)
+                assert result.isError is error
+
+    with analytics_logging(path):
+        anyio.run(run)
+
+    text = path.read_text()
+    assert "private" not in text
+    events = [json.loads(line) for line in text.splitlines()]
+    assert len(events) == 4
+    assert [event["outcome"] for event in events] == [
+        "success",
+        "error",
+        "error",
+        "error",
+    ]
+    assert len({event["call_id"] for event in events}) == 4
+    for event in events:
+        assert event["event"] == "tool_call"
+        assert event["duration_ms"] >= 0
+        assert event["timestamp"].endswith("+00:00")
+        assert "instance" in event
+
+
+def test_logging_appends_without_duplicate_handlers(tmp_path):
+    path = tmp_path / "analytics.jsonl"
+    for _ in range(2):
+        with analytics_logging(path):
+            log_event("server_start")
+    assert len(path.read_text().splitlines()) == 2
+
+
+def test_default_logging_uses_stderr(capsys):
+    with analytics_logging():
+        log_event("server_start")
+    output = capsys.readouterr()
+    assert output.out == ""
+    assert json.loads(output.err)["event"] == "server_start"
+
+
+@pytest.mark.parametrize("transport", ["stdio", "http"])
+def test_cli_lifecycle_logging(tmp_path, monkeypatch, transport):
+    server = Mock()
+    monkeypatch.setattr("kcidev.mcp.create_server", Mock(return_value=server))
+
+    async def run_stdio(server):
+        pass
+
+    monkeypatch.setattr(mcp, "_run_stdio", run_stdio)
+    path = tmp_path / "analytics.jsonl"
+    result = CliRunner().invoke(
+        mcp.mcp,
+        ["--transport", transport, "--log-file", str(path)],
+        obj={},
+    )
+    assert result.exit_code == 0, result.output
+    events = [json.loads(line) for line in path.read_text().splitlines()]
+    assert [event["event"] for event in events] == ["server_start", "server_stop"]
+    assert all(event["transport"] == transport for event in events)
+
+
+def test_cli_reports_unwritable_log_path(tmp_path):
+    result = CliRunner().invoke(
+        mcp.mcp, ["--log-file", str(tmp_path / "missing" / "log")], obj={}
+    )
+    assert result.exit_code == 1
+    assert "Cannot open analytics log" in result.output
+
+
+def test_cancelled_tool_is_logged(tmp_path):
+    server = create_server()
+
+    @server.tool()
+    async def waiting():
+        await anyio.sleep_forever()
+
+    path = tmp_path / "analytics.jsonl"
+
+    async def run():
+        from mcp.types import CallToolRequest, CallToolRequestParams
+
+        handler = server._mcp_server.request_handlers[CallToolRequest]
+        with anyio.move_on_after(0.05) as scope:
+            await handler(CallToolRequest(params=CallToolRequestParams(name="waiting")))
+        assert scope.cancel_called
+
+    with analytics_logging(path):
+        anyio.run(run)
+    event = json.loads(path.read_text())
+    assert event["outcome"] == "cancelled"
+    assert event["tool"] == "waiting"