diff --git a/.sampo/changesets/mcp-warn-when-stateless-session-not-wired.md b/.sampo/changesets/mcp-warn-when-stateless-session-not-wired.md new file mode 100644 index 000000000..601c5d907 --- /dev/null +++ b/.sampo/changesets/mcp-warn-when-stateless-session-not-wired.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: patch +--- + +MCP analytics now surfaces the previously-silent case where the stateless session mint middleware (`PostHogMcpStatelessSessionMiddleware`) never attached — the trap where an ASGI app is built or mounted before `instrument()` runs, so autowiring can't retrofit it and every session falls back to a fragmented per-process id. `instrument()` warns when `streamable_http_app()` was already called before it ran, and a one-time warning fires the first time a tool call arrives over streamable HTTP and the session still has to come from process memory. Both go to the `posthog.mcp` standard-library logger as well as the `MCPAnalyticsOptions(logger=...)` sink, so they are visible without opting in — silence them with `logging.getLogger("posthog.mcp").setLevel(logging.ERROR)`. Neither fires for stdio, a correctly-wired server, a conversation-anchored session, or the SSE transport (which the mint cannot fix). Documented in the new `posthog/mcp/README.md`. diff --git a/examples/mcp_stateless.py b/examples/mcp_stateless.py index 943c44c13..5d31a4187 100644 --- a/examples/mcp_stateless.py +++ b/examples/mcp_stateless.py @@ -37,10 +37,19 @@ def greet(name: str) -> str: server.run(transport="streamable-http") -# No FastMCP server to wire (a custom dispatcher)? Add the middleware to your own -# ASGI app and read the recovered session per request: +# Building the ASGI app yourself (e.g. mounting into FastAPI) or wiring a custom +# dispatcher? Autowiring only affects an app built AFTER instrument() runs, so an app +# built or mounted earlier gets no middleware. Add it to your own app explicitly, and +# read the recovered session per request: # # from posthog.mcp import PostHogMcpStatelessSessionMiddleware, get_mcp_session # # app.add_middleware(PostHogMcpStatelessSessionMiddleware) # sess = get_mcp_session(request) # sess.session_id, sess.client_name, ... +# +# Get that wrong and the SDK now says so, on the `posthog.mcp` logger: once at +# instrument() time, and once on the first request that resolves without a session. +# MCPAnalyticsOptions(enable_conversation_id=True) sidesteps the whole ordering +# question -- it anchors the session with no middleware at all. +# +# See posthog/mcp/README.md (stateless / multi-pod servers) for the full rundown. diff --git a/posthog/mcp/README.md b/posthog/mcp/README.md new file mode 100644 index 000000000..b358360ce --- /dev/null +++ b/posthog/mcp/README.md @@ -0,0 +1,95 @@ +# PostHog MCP analytics + +Product analytics for Model Context Protocol servers. Wrap a Python MCP server so +every tool call, agent intent, and failure is captured to PostHog as a `$mcp_*` event. + +```python +from posthog import Posthog +from posthog.mcp import instrument +from mcp.server.fastmcp import FastMCP + +posthog = Posthog("phc_...", host="https://us.i.posthog.com") +server = FastMCP("my-server") +analytics = instrument(server, posthog) +``` + +Install is just `pip install posthog`. `instrument()` needs the MCP SDK at runtime, +but anyone wrapping a server already has it. + +## Stateless / multi-pod servers + +A stateless MCP server issues no session id, so `$session_id` fragments across pods +and the client identity (sent only at `initialize`) is lost. PostHog fixes this with +a small ASGI middleware — `PostHogMcpStatelessSessionMiddleware` — that mints a +self-encoded token onto the `Mcp-Session-Id` response header at `initialize`; the +client replays it on every request, so any pod recovers the session and harness from +the header alone. + +### Zero-config path (recommended) + +`instrument()` wraps the FastMCP server's app factories (`streamable_http_app()` / +`sse_app()`), so an app you build **after** calling `instrument()` already carries the +middleware — including `mcp.run(transport="streamable-http")`, which calls those +factories internally. Nothing extra to add, as long as `instrument()` runs first: + +```python +server = FastMCP("my-server", stateless_http=True) +instrument(server, posthog) +server.run(transport="streamable-http") # already wired +``` + +### Manual path — required when you build the app yourself + +Autowiring only affects an app built **after** `instrument()` runs. If you build or +mount the ASGI app before `instrument()`, or in a different module — the common +FastAPI case — the running app gets **no** middleware and every session falls back to +a fragmented per-process id. Add the middleware to your app explicitly: + +```python +from posthog.mcp import PostHogMcpStatelessSessionMiddleware, get_mcp_session + +app = mcp.streamable_http_app() +app.add_middleware(PostHogMcpStatelessSessionMiddleware) +``` + +This is also the path for a custom `PostHogMCP` dispatcher (you own the ASGI app), +where you then read the recovered session per request: + +```python +sess = get_mcp_session(request) # sess.session_id, sess.client_name, ... +``` + +### Or skip the middleware entirely: conversation ids + +`MCPAnalyticsOptions(enable_conversation_id=True)` derives `$session_id` from the +agent's conversation handle, deterministically and identically on every pod. That +needs no middleware and no ordering discipline, and it is the only thing that +correlates a session under the 2026-07-28 revision's per-request server instances. +Prefer it if you're on a recent client. + +### How the SDK tells you it's misconfigured + +The failure used to be silent. It now surfaces two ways: + +- **At `instrument()`** — if `streamable_http_app()` was already called before + `instrument()` ran, so the live app has no middleware. +- **At runtime, once** — the first time a tool call arrives over streamable HTTP and the + session still has to come from this process's memory. + +Both go to the logger you pass via `MCPAnalyticsOptions(logger=...)` **and** to the +`posthog.mcp` standard-library logger, so you see them without opting in. Silence them +like any other logger: + +```python +logging.getLogger("posthog.mcp").setLevel(logging.ERROR) +``` + +Neither fires for stdio, for a correctly-wired server, or for a conversation-anchored +session. The instrument-time check can't see whether you added the middleware yourself +(the app is already built by then), so ignore it if you did. + +Two gaps worth knowing: jlowin's `fastmcp` 2.x/3.x doesn't expose the attribute the +instrument-time check reads, so those servers get the runtime warning only. And the +deprecated SSE transport is excluded — it keys sessions off a query parameter, and the +mint sets a response header an SSE client never replays, so the middleware wouldn't +help it. diff --git a/posthog/mcp/_instrumentation.py b/posthog/mcp/_instrumentation.py index 43f985711..be274c6cc 100644 --- a/posthog/mcp/_instrumentation.py +++ b/posthog/mcp/_instrumentation.py @@ -21,10 +21,11 @@ from ._exceptions import capture_exception from ._intent import resolve_tool_call_intent, set_event_intent from ._internal import MCPAnalyticsData, handle_identify, resolve_event_properties -from .logger import log +from .logger import log, warn +from .request_headers import get_request from ._sanitization import build_captured_mcp_parameters from ._transport_identity import stamp_transport_identity -from .session import resolve_session_id +from .session import resolve_session_id, resolve_session_id_with_source from .session_token import SessionTokenPayload, decode_session_id # Keep strong refs to in-flight capture tasks/futures and their lifecycle owners so @@ -268,6 +269,46 @@ async def prime_session( await resolve_session_id(data, mcp_session_id, token=token) +def _is_sse_request(extra: Optional[Dict[str, Any]]) -> bool: + """True for the deprecated SSE transport, which carries its session as a + ``session_id`` query parameter rather than a header. + + Such a request resolves to a ``generated`` session for a reason the stateless + mint cannot fix -- the mint sets a response header an SSE client never replays -- + so :func:`_warn_stateless_session_not_wired` would be recommending a remedy that + does not apply.""" + try: + params = getattr(get_request(extra), "query_params", None) + return bool(params is not None and params.get("session_id")) + except Exception: # noqa: BLE001 - a transport probe must never break a tool call + return False + + +def _warn_stateless_session_not_wired(data: MCPAnalyticsData) -> None: + """Warn once per server when a tool call/listing arrives over HTTP but the + session still had to come from this process's memory. + + That is the fingerprint of a stateless/multi-pod server whose mint middleware + never attached — most often because the ASGI app was built (or mounted from + another module) *before* ``instrument()`` ran, so wrapping the app factories + couldn't retrofit the already-built app. The result is a silently fragmented + ``$session_id``; this makes that failure loud instead of dark-in-prod.""" + if data.warned_no_stateless_session: + return + data.warned_no_stateless_session = True + warn( + "Warning: an MCP tool request arrived over streamable HTTP with no session id, so " + "PostHog generated a per-process $session_id that will fragment across requests " + "and pods. This usually means PostHogMcpStatelessSessionMiddleware never attached " + "— e.g. the ASGI app was built or mounted before instrument() ran. If you build " + "the app yourself, add the middleware explicitly: " + "app.add_middleware(PostHogMcpStatelessSessionMiddleware). " + "Enabling conversation ids (MCPAnalyticsOptions(enable_conversation_id=True)) also " + "anchors the session without any middleware. " + "See posthog/mcp/README.md (stateless / multi-pod servers)." + ) + + async def prepare_request( data: MCPAnalyticsData, *, @@ -305,10 +346,20 @@ async def prepare_request( when ``capture_event`` builds the initialize event — otherwise the first ``$mcp_initialize`` is anonymous even when identify resolves on the same request. (Still not byte-parity with the TS SDK, which wraps the real initialize handler; - the Python SDK handles initialize in the session layer, not ``request_handlers``.)""" - session_id = await resolve_session_id( + the Python SDK handles initialize in the session layer, not ``request_handlers``.) + + A request that reached us over HTTP yet still resolved to this process's memory + has nothing correlating it across pods, which on a stateless server means the + mint middleware never attached — warn once rather than fragment silently.""" + session_id, session_source = await resolve_session_id_with_source( data, mcp_session_id, token=token, conversation_id=conversation_id ) + if ( + session_source == "generated" + and get_request(extra) is not None + and not _is_sse_request(extra) + ): + _warn_stateless_session_not_wired(data) identify_event = await handle_identify(data, session_id, request, extra) if identify_event: fire_and_forget(capture_event(data, identify_event), data) diff --git a/posthog/mcp/_internal.py b/posthog/mcp/_internal.py index 1a29dbd47..a91a6935f 100644 --- a/posthog/mcp/_internal.py +++ b/posthog/mcp/_internal.py @@ -62,6 +62,10 @@ class MCPAnalyticsData: session_id: str = "" session_source: str = "generated" # "generated" | "mcp" | "token" last_mcp_session_id: Optional[str] = None + # Set once we've warned that an HTTP request resolved with no session id — the + # signature of a stateless server whose mint middleware never attached. Warned + # a single time per server so the log isn't flooded on every request. + warned_no_stateless_session: bool = False last_activity: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) identified_sessions: IdentityCache = field(default_factory=IdentityCache) tool_categories: Dict[str, str] = field(default_factory=dict) diff --git a/posthog/mcp/asgi.py b/posthog/mcp/asgi.py index 7139751ab..4e742a143 100644 --- a/posthog/mcp/asgi.py +++ b/posthog/mcp/asgi.py @@ -40,7 +40,7 @@ import json from typing import Any, Optional -from .logger import log +from .logger import log, warn from .session import new_session_id from .session_token import ( MCP_SESSION_HEADER, @@ -245,6 +245,7 @@ def autowire_stateless_mint(server: Any) -> None: On fastmcp 2.x, ``streamable_http_app`` / ``sse_app`` can be thin wrappers over ``http_app``; wrapping all three could add the middleware twice to one app, so the factory guards against a double-add (see ``_app_already_wrapped``).""" + _warn_if_app_built_before_instrument(server) for attr in ("streamable_http_app", "sse_app", "http_app"): original = getattr(server, attr, None) if not callable(original) or getattr(original, _AUTOWIRED, False): @@ -255,6 +256,51 @@ def autowire_stateless_mint(server: Any) -> None: log(f"PostHog MCP: could not auto-wire stateless mint on {attr} - {error}") +def _app_was_already_built(server: Any) -> bool: + """Whether the streamable-HTTP app already exists, so wrapping the factories + now cannot retrofit it. + + The tell is ``_session_manager``, created lazily on the first + ``streamable_http_app()`` call and non-``None`` forever after. It sits on the + server itself on the official SDK's ``FastMCP`` (1.x) and on the low-level + server it delegates to (2.x's ``MCPServer`` renamed that attribute + ``_lowlevel_server``; older/other wrappers may still use ``_mcp_server``), so + check both names. + + Deliberately partial: jlowin's ``fastmcp`` 2.x/3.x keeps its session manager as + a local inside ``http_app()`` and never stores it, so there is nothing to probe + and those servers get no instrument-time warning. The runtime warning in + ``_instrumentation`` still covers them.""" + low_level = getattr(server, "_mcp_server", None) or getattr( + server, "_lowlevel_server", None + ) + for candidate in (server, low_level): + try: + if getattr(candidate, "_session_manager", None) is not None: + return True + except Exception: # noqa: BLE001 - never let a probe break instrument() + continue + return False + + +def _warn_if_app_built_before_instrument(server: Any) -> None: + """Catch the ordering trap that silently disables stateless capture: the + streamable-HTTP app was built (and likely already mounted) *before* ``instrument()`` + ran, so wrapping the factories now can't retrofit that already-built app.""" + if not _app_was_already_built(server): + return + warn( + "Warning: streamable_http_app() was called before instrument(), so the ASGI app " + "already in use has no PostHog MCP middleware and stateless sessions will not be " + "captured (autowiring only affects apps built after instrument() runs). Call " + "instrument(server) before building or mounting the app, or add the middleware " + "manually: app.add_middleware(PostHogMcpStatelessSessionMiddleware). " + "You can ignore this if you already added the middleware yourself — the app is " + "built by then, so there is no way for us to tell from here. " + "See posthog/mcp/README.md (stateless / multi-pod servers)." + ) + + def _app_already_wrapped(app: Any) -> bool: """True if ``app`` already carries our middleware -- so wrapping a factory that delegates to another wrapped factory (fastmcp 2.x aliases) doesn't add it twice.""" diff --git a/posthog/mcp/logger.py b/posthog/mcp/logger.py index 8e6b5e009..33a41e1ea 100644 --- a/posthog/mcp/logger.py +++ b/posthog/mcp/logger.py @@ -8,10 +8,13 @@ protocol messages, so the SDK must never ``print``. We accept a ``logger`` option on the public API; when omitted, log calls are silently dropped. Plug in any callable (e.g. a file logger, or ``print`` for non-STDIO transports). + +:func:`warn` is the exception to "silently dropped" -- see its docstring. """ from __future__ import annotations +import logging from typing import Callable, Optional __all__ = ["set_logger"] @@ -20,6 +23,8 @@ _active_logger: Optional[LoggerFn] = None +_stdlib_logger = logging.getLogger("posthog.mcp") + def set_logger(logger: Optional[LoggerFn]) -> None: global _active_logger @@ -33,3 +38,21 @@ def log(message: str) -> None: except Exception: # never let logging blow up the tracking pipeline pass + + +def warn(message: str) -> None: + """A misconfiguration the host almost certainly wants to know about, sent to + the ``logger`` option *and* to the ``posthog.mcp`` standard-library logger. + + Reserved for warnings that can only fire on an HTTP transport, where the + STDIO constraint above does not apply. A default-configured host still sees + these on stderr (logging's lastResort handler), which is the whole point: + the misconfigurations this is used for are invisible in the data, so a + warning nobody has opted in to receive is a warning nobody reads. Hosts that + do configure logging can route or silence them by name like any other + logger.""" + log(message) + try: + _stdlib_logger.warning(message) + except Exception: + pass diff --git a/posthog/mcp/request_headers.py b/posthog/mcp/request_headers.py index 6b31023d1..abb478722 100644 --- a/posthog/mcp/request_headers.py +++ b/posthog/mcp/request_headers.py @@ -38,22 +38,35 @@ def identify(request, extra): RequestHeaderBag = Dict[str, str] -def get_request_headers(extra: Any) -> Optional[RequestHeaderBag]: - """The request's HTTP headers as a plain dict with lowercase keys, or ``None``. +def get_request(extra: Any) -> Optional[Any]: + """The transport's per-request object (Starlette ``Request`` or equivalent) + underneath ``extra``, or ``None`` on stdio / in-memory transports. Accepts the ``extra`` dict handed to a callback, or the raw per-request - context itself, so it works whichever one a host happens to hold. + context itself, so it works whichever one a host happens to hold. Both SDK + majors reach it the same way from their own context object + (``ServerRequestContext`` on 2.x, ``RequestContext`` on 1.x). + + Shared by anything that needs to read the request beyond just its headers + (e.g. query params) -- one place that knows how to unwrap ``extra``/``ctx`` + down to the request, instead of each caller re-deriving it. """ ctx = extra if isinstance(extra, dict): ctx = extra.get("ctx") if ctx is None: return None + return getattr(ctx, "request", None) + - # Both majors reach the transport's request the same way from their own - # context object (`ServerRequestContext` on 2.x, `RequestContext` on 1.x); +def get_request_headers(extra: Any) -> Optional[RequestHeaderBag]: + """The request's HTTP headers as a plain dict with lowercase keys, or ``None``. + + Accepts the ``extra`` dict handed to a callback, or the raw per-request + context itself, so it works whichever one a host happens to hold. + """ # `request` is None on stdio. - source = getattr(getattr(ctx, "request", None), "headers", None) + source = getattr(get_request(extra), "headers", None) if source is None: return None return _to_header_bag(source) diff --git a/posthog/mcp/session.py b/posthog/mcp/session.py index a7edbc8d4..dc59f51fb 100644 --- a/posthog/mcp/session.py +++ b/posthog/mcp/session.py @@ -51,9 +51,32 @@ async def resolve_session_id( token: Optional[SessionTokenPayload] = None, conversation_id: Optional[str] = None, ) -> str: + """The session id for this request. See :func:`resolve_session_id_with_source`, + which this wraps -- callers that need to know *where* the session came from + should use that instead of reading ``data.session_source`` afterwards.""" + session_id, _ = await resolve_session_id_with_source( + data, mcp_session_id, token=token, conversation_id=conversation_id + ) + return session_id + + +async def resolve_session_id_with_source( + data: MCPAnalyticsData, + mcp_session_id: Optional[str], + *, + token: Optional[SessionTokenPayload] = None, + conversation_id: Optional[str] = None, +) -> tuple[str, str]: """Resolve the session id for a request. Mutates per-server state under a lock so concurrent async requests can't race on session rotation. + Returns ``(session_id, source)`` where source is one of ``"conversation"``, + ``"token"``, ``"mcp"`` or ``"generated"`` -- describing *this* request. Callers + must not infer it from ``data.session_source`` instead: that field is shared + mutable state which the conversation branch below deliberately never writes, so + reading it after the fact reports whatever some earlier request happened to + leave there. + Priority mirrors posthog-js ``getSessionId``: the agent's ``conversation_id`` handle first (the only id that survives the 2026-07-28 revision's per-request server instances), then our self-encoded session token, then the @@ -75,7 +98,7 @@ async def resolve_session_id( a genuine token session never needs the fallback. """ if conversation_id: - return derive_session_id_from_conversation(conversation_id) + return derive_session_id_from_conversation(conversation_id), "conversation" async with data.session_lock: now = datetime.now(timezone.utc) @@ -87,20 +110,20 @@ async def resolve_session_id( data.session_id = token.session_id data.session_source = "token" data.last_activity = now - return data.session_id + return data.session_id, "token" if mcp_session_id: data.session_id = derive_session_id_from_mcp_session(mcp_session_id) data.last_mcp_session_id = mcp_session_id data.session_source = "mcp" data.last_activity = now - return data.session_id + return data.session_id, "mcp" # Once a session is MCP-derived, keep it even if a later request arrives # without the MCP session id, so the session doesn't fragment. if data.session_source == "mcp" and data.last_mcp_session_id: data.last_activity = now - return data.session_id + return data.session_id, "mcp" # Memory fallback (single-owner transports like stdio). A leftover token # session must NOT leak to a credential-less request, so anything that @@ -112,4 +135,4 @@ async def resolve_session_id( data.session_id = new_session_id() data.session_source = "generated" data.last_activity = now - return data.session_id + return data.session_id, "generated" diff --git a/posthog/test/mcp/test_session_token.py b/posthog/test/mcp/test_session_token.py index 82ac19e8c..10d79cbac 100644 --- a/posthog/test/mcp/test_session_token.py +++ b/posthog/test/mcp/test_session_token.py @@ -4,14 +4,21 @@ from __future__ import annotations import json +from contextlib import contextmanager +from types import SimpleNamespace import pytest from posthog.mcp._internal import MCPAnalyticsData from posthog.mcp.asgi import ( PostHogMcpStatelessSessionMiddleware, + _app_was_already_built, get_mcp_session, ) +from posthog.test.mcp._helpers import MCP_MAJOR +from posthog.mcp import logger as logger_module +from posthog.mcp._instrumentation import prepare_request +from posthog.mcp.logger import set_logger from posthog.mcp.session import new_session_id, resolve_session_id from posthog.mcp.session_token import ( MCP_SESSION_HEADER, @@ -496,10 +503,6 @@ def test_instrument_autowires_stateless_mint_no_manual_middleware(): from posthog.mcp import instrument - class _Sink: - def capture(self, *_: object, **__: object) -> None: - pass - srv = FastMCP( "posthog-autowire-test", stateless_http=True, @@ -542,3 +545,316 @@ def ping() -> str: assert payload is not None, "instrument() did not auto-wire the mint" assert payload.client_name == "Cursor" assert payload.client_version == "0.42" + + +# --- loud diagnostics for the silent "middleware never attached" failure ----- +# +# The failure these guard: on a stateless server whose mint middleware never +# attached, every request falls back to a per-process session and `$session_id` +# fragments across pods with nothing in the SDK saying so. Two signals cover it -- +# one at instrument() time, one on the first affected request. + +# Substring unique to the runtime warning (the instrument-time one has its own). +_NO_SESSION = "no session id" +_WRONG_ORDER = "streamable_http_app() was called before instrument()" + + +class _Sink: + def capture(self, *_: object, **__: object) -> None: + pass + + +@contextmanager +def _captured_logs(): + """Route the SDK logger into a list for the duration of the block, then put + back whatever sink was installed before -- `set_logger` is global process + state, so restoring `None` unconditionally would silence a concurrent test.""" + logs: list[str] = [] + previous = logger_module._active_logger + set_logger(logs.append) + try: + yield logs + finally: + set_logger(previous) + + +def _http_ctx(headers=None): + """The per-request context shape callbacks receive as `extra["ctx"]` on an + HTTP transport. `request=None` is how every SDK major represents stdio.""" + return SimpleNamespace(request=SimpleNamespace(headers=headers or {})) + + +def _stateless_server(name: str): + """A real stateless streamable-HTTP FastMCP with one tool.""" + from mcp.server.fastmcp import FastMCP + from mcp.server.transport_security import TransportSecuritySettings + + srv = FastMCP( + name, + stateless_http=True, + json_response=True, + # TestClient sends Host: testserver; allow it past DNS-rebinding protection. + transport_security=TransportSecuritySettings( + enable_dns_rebinding_protection=False + ), + ) + + @srv.tool() + def ping() -> str: + return "pong" + + return srv + + +def _rpc(method, params=None, id=1, extra_headers=None): + headers = { + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + **(extra_headers or {}), + } + body = {"jsonrpc": "2.0", "id": id, "method": method} + if params is not None: + body["params"] = params + return headers, json.dumps(body) + + +def _call_ping(client, extra_headers=None, id=1): + headers, body = _rpc( + "tools/call", + {"name": "ping", "arguments": {}}, + id=id, + extra_headers=extra_headers, + ) + return client.post("/mcp", headers=headers, content=body) + + +# --- runtime signal, end to end ---------------------------------------------- + + +def test_runtime_warns_when_app_was_built_before_instrument(): + """The customer's ordering trap, reproduced against a real transport: the ASGI + app is built before instrument(), so autowiring can't retrofit it and a real + tools/call arrives with no session of any kind. Both signals must fire.""" + pytest.importorskip("starlette.testclient") + pytest.importorskip("mcp.server.fastmcp") # v1-only server; skipped under mcp>=2 + from starlette.testclient import TestClient + + from posthog.mcp import instrument + + srv = _stateless_server("posthog-ordering-trap") + app = srv.streamable_http_app() # BEFORE instrument() -- the trap + + with _captured_logs() as logs: + instrument(srv, _Sink()) + with TestClient(app) as client: + resp = _call_ping(client) + assert resp.status_code == 200, resp.text + + assert [m for m in logs if _WRONG_ORDER in m], "no instrument-time warning" + assert [m for m in logs if _NO_SESSION in m], "no runtime warning" + + +def test_runtime_silent_when_correctly_wired(): + """The regression that matters most: a correctly-ordered stateless server whose + client replays the minted token must stay completely quiet. A diagnostic that + cries wolf on healthy servers is worse than no diagnostic.""" + pytest.importorskip("starlette.testclient") + pytest.importorskip("mcp.server.fastmcp") + from starlette.testclient import TestClient + + from posthog.mcp import instrument + + srv = _stateless_server("posthog-correctly-wired") + + with _captured_logs() as logs: + instrument(srv, _Sink()) # BEFORE building the app -- autowiring works + app = srv.streamable_http_app() + + with TestClient(app) as client: + headers, body = _rpc( + "initialize", + { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "Claude Code", "version": "9.9.9"}, + }, + ) + resp = client.post("/mcp", headers=headers, content=body) + assert resp.status_code == 200, resp.text + token = resp.headers.get(MCP_SESSION_HEADER) + assert decode_session_id(token) is not None, "mint did not attach" + + # A compliant client replays the token on every subsequent request. + resp2 = _call_ping( + client, + extra_headers={ + MCP_SESSION_HEADER: token, + "mcp-protocol-version": "2025-06-18", + }, + id=2, + ) + assert resp2.status_code == 200, resp2.text + + assert not [m for m in logs if _WRONG_ORDER in m] + assert not [m for m in logs if _NO_SESSION in m] + + +def test_warnings_are_visible_without_configuring_a_logger(caplog): + """The point of the whole change. `log()` is a no-op unless the host passes + MCPAnalyticsOptions(logger=...), so routing these warnings through it alone + would leave the failure exactly as dark as it was -- the customer who lost + weeks of sessions had no logger configured. They go to the `posthog.mcp` + stdlib logger too, which a default-configured host actually sees.""" + pytest.importorskip("starlette.testclient") + pytest.importorskip("mcp.server.fastmcp") + from starlette.testclient import TestClient + + from posthog.mcp import instrument + + srv = _stateless_server("posthog-no-logger-configured") + app = srv.streamable_http_app() + + set_logger(None) # explicitly no `logger` option anywhere + with caplog.at_level("WARNING", logger="posthog.mcp"): + instrument(srv, _Sink()) + with TestClient(app) as client: + _call_ping(client) + + messages = [r.getMessage() for r in caplog.records if r.name == "posthog.mcp"] + assert [m for m in messages if _NO_SESSION in m], "runtime warning not visible" + assert [m for m in messages if _WRONG_ORDER in m], "ordering warning not visible" + + +# --- runtime signal, predicate edges ----------------------------------------- + + +async def test_no_warning_for_stdio(): + """stdio carries no request at all, so a per-process session is correct there. + Warning would be pure noise on the most common local-dev path.""" + with _captured_logs() as logs: + data = _data() + await prepare_request( + data, + mcp_session_id=None, + client_name=None, + client_version=None, + request={"method": "tools/call", "params": {}}, + extra={"ctx": SimpleNamespace(request=None)}, + ) + + assert not [m for m in logs if _NO_SESSION in m] + assert data.warned_no_stateless_session is False + + +async def test_no_warning_when_session_is_anchored_by_conversation_id(): + """A conversation-anchored session is derived deterministically and agrees + across pods, so it does not fragment -- no warning, even though the request is + HTTP and carries no session header. + + Regression test for a real trap: `resolve_session_id` returns early on this + path and never writes `data.session_source`, so a check that read that shared + field afterwards would see a stale "generated" and warn about a session that + is perfectly healthy.""" + with _captured_logs() as logs: + data = _data() + session_id = await prepare_request( + data, + mcp_session_id=None, + client_name=None, + client_version=None, + request={"method": "tools/call", "params": {}}, + extra={"ctx": _http_ctx()}, + conversation_id="0199e0a1-0000-7000-8000-000000000000", + ) + + assert session_id.startswith("ses_") + assert not [m for m in logs if _NO_SESSION in m] + + +async def test_no_warning_for_sse_transport(): + """The deprecated SSE transport carries its session as a query param, and the + mint sets a response header an SSE client never replays -- so the middleware we + would be recommending cannot help. Stay quiet rather than give wrong advice.""" + ctx = _http_ctx() + ctx.request.query_params = {"session_id": "sse-session-1"} + + with _captured_logs() as logs: + await prepare_request( + _data(), + mcp_session_id=None, + client_name=None, + client_version=None, + request={"method": "tools/call", "params": {}}, + extra={"ctx": ctx}, + ) + + assert not [m for m in logs if _NO_SESSION in m] + + +async def test_runtime_warning_fires_once_per_server(): + """Warn-once: a busy server must not write this line on every request.""" + with _captured_logs() as logs: + data = _data() + for _ in range(3): + await prepare_request( + data, + mcp_session_id=None, + client_name=None, + client_version=None, + request={"method": "tools/call", "params": {}}, + extra={"ctx": _http_ctx()}, + ) + + warnings = [m for m in logs if _NO_SESSION in m] + assert len(warnings) == 1 + assert "add_middleware(PostHogMcpStatelessSessionMiddleware)" in warnings[0] + assert data.warned_no_stateless_session is True + + +# --- instrument-time signal --------------------------------------------------- + + +def test_no_instrument_warning_when_app_not_yet_built(): + """The correctly-ordered path: instrument() on a server whose app has never + been built has nothing to complain about.""" + pytest.importorskip("mcp.server.fastmcp") + + from posthog.mcp import instrument + + srv = _stateless_server("posthog-app-not-built") + + logs: list[str] = [] + instrument(srv, _Sink(), MCPAnalyticsOptions(logger=logs.append)) + set_logger(None) # instrument() installs the option globally; undo it + + assert not [m for m in logs if _WRONG_ORDER in m] + + +def _server_for_installed_major(name: str): + """A streamable-HTTP server built with whichever MCP SDK major is installed.""" + if MCP_MAJOR < 2: + from mcp.server.fastmcp import FastMCP + + return FastMCP(name, stateless_http=True) + from mcp.server.mcpserver import MCPServer + + return MCPServer(name) + + +def test_app_built_probe_fires_on_the_installed_sdk_major(): + """``_app_was_already_built`` must flip once the app exists -- on *both* MCP + majors, which is why this runs unguarded on whichever one is installed. + + The attribute holding the low-level server was renamed across the major + boundary (``_mcp_server`` on 1.x's FastMCP, ``_lowlevel_server`` on 2.x's + MCPServer). A probe that knows only one name still passes every 1.x test + while silently never firing on 2.x, so the instrument-time warning goes dark + on exactly one half of the matrix with nothing failing to say so.""" + srv = _server_for_installed_major("posthog-probe-major") + + assert _app_was_already_built(srv) is False, "probe fired before the app existed" + srv.streamable_http_app() + assert _app_was_already_built(srv) is True, ( + f"probe blind to a built app on MCP {MCP_MAJOR}.x -- the ordering warning " + "cannot fire for these servers" + ) diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index 06009428a..54ca98d0d 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -349,6 +349,7 @@ alias posthog.mcp.asgi.encode_session_id -> posthog.mcp.session_token.encode_ses alias posthog.mcp.asgi.log -> posthog.mcp.logger.log alias posthog.mcp.asgi.new_session_id -> posthog.mcp.session.new_session_id alias posthog.mcp.asgi.read_mcp_session_header -> posthog.mcp.session_token.read_mcp_session_header +alias posthog.mcp.asgi.warn -> posthog.mcp.logger.warn alias posthog.mcp.decode_session_id -> posthog.mcp.session_token.decode_session_id alias posthog.mcp.derive_session_id_from_conversation -> posthog.mcp.session.derive_session_id_from_conversation alias posthog.mcp.derive_session_id_from_mcp_session -> posthog.mcp.session.derive_session_id_from_mcp_session