feat(MCP): Emit session and tool call events

Contributes to Flagsmith/flagsmith-private#152.

A middleware emits structured events on the mcp domain:

- session.opened on initialize, with the client name mapped from the
  User-Agent product token against MCP_KNOWN_CLIENTS
- client.unknown when the User-Agent is not recognised, so real-world
  client User-Agents can be discovered and added to the known set
- tool.called for every tools/call, with tool, client, status and
  duration
- tool.failed on the error path, with exc_info

beep boop

Author: khvn26

mcp/pyproject.toml

@@ -24,6 +24,7 @@ dev = [
     "pytest-asyncio>=1.3.0,<2.0.0",               # Run asynchronous tests
     "pytest-cov>=7.0.0,<8.0.0",                   # Measure test coverage
     "pytest-mock>=3.15.1,<4.0.0",                 # Mock via fixtures
+    "pytest-structlog>=1.1,<2.0.0",               # Assert structlog events
     "respx>=0.22,<1.0",                           # Mock HTTP interactions
     "ruff>=0.15.12,<0.16.0",                      # Lint and format
 ]

mcp/src/flagsmith_mcp/constants.py

@@ -1,3 +1,14 @@
 # TODO: consume a version-controlled schema — https://github.com/Flagsmith/flagsmith/issues/7669
 OPENAPI_SPEC_URL = "https://api.flagsmith.com/api/v1/swagger.json"
 OAUTH_SCOPES = ["mcp"]
+
+# Mirrors MCP_KNOWN_CLIENTS in the Core API's app_analytics.
+UNKNOWN_CLIENT = "unknown"
+MCP_KNOWN_CLIENTS = {
+    "claude-code",
+    "claude-desktop",
+    "cursor",
+    "windsurf",
+    "zed",
+    "mcp-remote",
+}

mcp/src/flagsmith_mcp/events.py

@@ -0,0 +1,83 @@
+import time
+
+import mcp.types as mt
+import structlog
+from fastmcp.server.dependencies import get_http_headers
+from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext
+from fastmcp.tools.base import ToolResult
+
+from flagsmith_mcp import constants
+
+logger = structlog.get_logger("mcp")
+
+
+def get_user_agent() -> str:
+    return get_http_headers(include={"user-agent"}).get("user-agent", "")
+
+
+def get_client_name(user_agent: str) -> str:
+    """Map a User-Agent to a known MCP client name.
+
+    The product token (the part before the first slash) identifies
+    the client, e.g. `claude-code/2.1 (cli)` -> `claude-code`.
+    """
+    name = user_agent.partition("/")[0].strip().lower()
+    if name in constants.MCP_KNOWN_CLIENTS:
+        return name
+    return constants.UNKNOWN_CLIENT
+
+
+class EventLoggingMiddleware(Middleware):
+    """Emit structured product events for MCP sessions and tool calls."""
+
+    async def on_initialize(
+        self,
+        context: MiddlewareContext[mt.InitializeRequest],
+        call_next: CallNext[mt.InitializeRequest, mt.InitializeResult | None],
+    ) -> mt.InitializeResult | None:
+        result = await call_next(context)
+        user_agent = get_user_agent()
+        client_name = get_client_name(user_agent)
+        if client_name == constants.UNKNOWN_CLIENT:
+            logger.info("client.unknown", client__user_agent=user_agent)
+        logger.info(
+            "session.opened",
+            client__name=client_name,
+            client__user_agent=user_agent,
+        )
+        return result
+
+    async def on_call_tool(
+        self,
+        context: MiddlewareContext[mt.CallToolRequestParams],
+        call_next: CallNext[mt.CallToolRequestParams, ToolResult],
+    ) -> ToolResult:
+        client_name = get_client_name(get_user_agent())
+        start = time.perf_counter()
+        try:
+            result = await call_next(context)
+        except Exception:
+            duration_ms = (time.perf_counter() - start) * 1000
+            logger.info(
+                "tool.called",
+                tool__name=context.message.name,
+                client__name=client_name,
+                status="error",
+                duration_ms=duration_ms,
+            )
+            logger.exception(
+                "tool.failed",
+                tool__name=context.message.name,
+                client__name=client_name,
+                status="error",
+                duration_ms=duration_ms,
+            )
+            raise
+        logger.info(
+            "tool.called",
+            tool__name=context.message.name,
+            client__name=client_name,
+            status="success",
+            duration_ms=(time.perf_counter() - start) * 1000,
+        )
+        return result

mcp/src/flagsmith_mcp/server.py

@@ -12,6 +12,7 @@
 
 from flagsmith_mcp import config, constants
 from flagsmith_mcp.auth import FlagsmithAuth
+from flagsmith_mcp.events import EventLoggingMiddleware
 from flagsmith_mcp.metrics import PrometheusMiddleware
 from flagsmith_mcp.oauth import FlagsmithResourceAuth
 from flagsmith_mcp.telemetry import setup_telemetry
@@ -70,6 +71,7 @@ def create_server(settings: config.Settings) -> FastMCP[None]:
     )
 
     server.add_middleware(PrometheusMiddleware())
+    server.add_middleware(EventLoggingMiddleware())
 
     @server.custom_route("/health", methods=["GET"])
     async def health(request: Request) -> PlainTextResponse:

mcp/tests/integration/test_events.py

@@ -0,0 +1,70 @@
+import pytest
+from fastmcp import Client
+from fastmcp.client.transports import FastMCPTransport
+from fastmcp.exceptions import ToolError
+from pytest_structlog import StructuredLogCapture
+from respx import MockRouter
+
+
+async def test_events__session_initialised__emits_session_opened(
+    log: StructuredLogCapture,
+    client: Client[FastMCPTransport],
+) -> None:
+    # Given the server started via the client fixture, with no HTTP context
+    # When the session is initialised by the fixture
+    # Then the client is reported as unknown
+    assert log.has("session.opened", client__name="unknown", client__user_agent="")
+    assert log.has("client.unknown", client__user_agent="")
+
+
+async def test_events__successful_tool_call__emits_tool_called(
+    log: StructuredLogCapture,
+    client: Client[FastMCPTransport],
+    respx_mock: MockRouter,
+) -> None:
+    # Given
+    respx_mock.get("https://api.flagsmith.com/environments/").respond(
+        json={"results": []}
+    )
+
+    # When
+    await client.call_tool("list_environments", {})
+
+    # Then
+    [event] = [e for e in log.events if e["event"] == "tool.called"]
+    assert event.pop("duration_ms") >= 0
+    assert event == {
+        "event": "tool.called",
+        "level": "info",
+        "tool__name": "list_environments",
+        "client__name": "unknown",
+        "status": "success",
+    }
+    assert not log.has("tool.failed")
+
+
+async def test_events__failing_tool_call__emits_tool_called_and_tool_failed(
+    log: StructuredLogCapture,
+    client: Client[FastMCPTransport],
+    respx_mock: MockRouter,
+) -> None:
+    # Given
+    respx_mock.get("https://api.flagsmith.com/environments/").respond(status_code=502)
+
+    # When
+    with pytest.raises(ToolError):
+        await client.call_tool("list_environments", {})
+
+    # Then
+    [called] = [e for e in log.events if e["event"] == "tool.called"]
+    assert called["status"] == "error"
+    [failed] = [e for e in log.events if e["event"] == "tool.failed"]
+    assert failed.pop("duration_ms") >= 0
+    assert failed.pop("exc_info") is True
+    assert failed == {
+        "event": "tool.failed",
+        "level": "error",
+        "tool__name": "list_environments",
+        "client__name": "unknown",
+        "status": "error",
+    }

mcp/tests/unit/test_event_logging.py

@@ -0,0 +1,48 @@
+import pytest
+from pytest_mock import MockerFixture
+from pytest_structlog import StructuredLogCapture
+
+from flagsmith_mcp import events
+
+
+@pytest.mark.parametrize(
+    "user_agent, expected",
+    [
+        ("claude-code/2.1 (cli)", "claude-code"),
+        ("Claude-Desktop/1.0", "claude-desktop"),
+        ("Cursor/1.2.3", "cursor"),
+        ("zed", "zed"),
+        ("mcp-remote/0.1.29 node/22.0.0", "mcp-remote"),
+        ("python-httpx/0.28", "unknown"),
+        ("", "unknown"),
+    ],
+)
+def test_get_client_name__user_agent__maps_to_known_client_or_unknown(
+    user_agent: str,
+    expected: str,
+) -> None:
+    # Given the parametrised User-Agent
+    # When / Then
+    assert events.get_client_name(user_agent) == expected
+
+
+async def test_event_logging_middleware__known_client__no_unknown_client_event(
+    mocker: MockerFixture,
+    log: StructuredLogCapture,
+) -> None:
+    # Given
+    user_agent = "claude-code/2.1 (cli)"
+    mocker.patch.object(events, "get_user_agent", return_value=user_agent)
+    middleware = events.EventLoggingMiddleware()
+    call_next = mocker.AsyncMock(return_value=None)
+
+    # When
+    await middleware.on_initialize(mocker.Mock(), call_next)
+
+    # Then
+    assert log.has(
+        "session.opened",
+        client__name="claude-code",
+        client__user_agent=user_agent,
+    )
+    assert not log.has("client.unknown")