From c82e8200c7cfbda16d1de100eeb918af2c960871 Mon Sep 17 00:00:00 2001 From: Georgis Andonis Date: Fri, 21 Aug 2026 13:04:46 +0300 Subject: [PATCH 1/3] fix(mcp): warn when stateless session middleware never attached The stateless-session mint (PostHogMcpStatelessSessionMiddleware) is zero-config only when the ASGI app is built after instrument() runs. An app built or mounted earlier (the common FastAPI case) silently gets no middleware, so every session falls back to a fragmented per-process id with nothing in the SDK saying so. Make the failure loud with two signals: - instrument() warns when streamable_http_app() was already called before it ran (a cached _session_manager is the tell, on the FastMCP server or the low-level server it delegates to). - A one-time runtime warning fires when a tool call arrives over streamable HTTP and the session still has to come from this process's memory. Both go to the posthog.mcp stdlib logger as well as the logger option, so they are visible without opting in -- routing them only through the opt-in sink would have left the failure as dark as it was. Detection reads the session source returned for *this* request rather than data.session_source, which is shared mutable state the conversation_id branch deliberately never writes; reading it after the fact would warn about conversation-anchored sessions that are perfectly healthy. The HTTP probe reuses get_request_headers, so all three adapters (v1 FastMCP, low-level, v2) are covered with no plumbing. Silent for stdio, correctly-wired servers, conversation-anchored sessions, and the deprecated SSE transport, whose session lives in a query param the mint cannot help with. Generated-By: PostHog Desktop Task-Id: 145ef960-7152-4c88-bed9-3214c268b1d0 --- ...p-warn-when-stateless-session-not-wired.md | 5 + examples/mcp_stateless.py | 13 +- posthog/mcp/README.md | 95 ++++++ posthog/mcp/_instrumentation.py | 63 +++- posthog/mcp/_internal.py | 4 + posthog/mcp/asgi.py | 43 ++- posthog/mcp/logger.py | 23 ++ posthog/mcp/session.py | 33 +- posthog/test/mcp/test_session_token.py | 288 ++++++++++++++++++ references/public_api_snapshot.txt | 1 + 10 files changed, 556 insertions(+), 12 deletions(-) create mode 100644 .sampo/changesets/mcp-warn-when-stateless-session-not-wired.md create mode 100644 posthog/mcp/README.md 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..bd9f2e4bb 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_headers 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,50 @@ 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.""" + if not isinstance(extra, dict): + return False + try: + params = getattr( + getattr(extra.get("ctx"), "request", None), "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 +350,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_headers(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..15e25188e 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,46 @@ 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 moved it there), so check both. + + 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.""" + for candidate in (server, getattr(server, "_mcp_server", None)): + 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/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..94c769545 100644 --- a/posthog/test/mcp/test_session_token.py +++ b/posthog/test/mcp/test_session_token.py @@ -4,6 +4,8 @@ from __future__ import annotations import json +from contextlib import contextmanager +from types import SimpleNamespace import pytest @@ -12,6 +14,9 @@ PostHogMcpStatelessSessionMiddleware, get_mcp_session, ) +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, @@ -542,3 +547,286 @@ 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] 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 From a36f7ef877aae91cfc7c2aa8f00857d33a37db76 Mon Sep 17 00:00:00 2001 From: Georgis Andonis Date: Fri, 21 Aug 2026 15:23:09 +0300 Subject: [PATCH 2/3] fix(mcp): probe the right attribute for a built app on MCP 2.x MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The instrument-time "app built before instrument()" check looked for the low-level server at `_mcp_server`. That is the 1.x FastMCP name; 2.x's MCPServer calls it `_lowlevel_server`, so the probe never saw a built app on 2.x and the warning could not fire there — the half of the matrix the check claimed to cover. Probe both names, and cover it with a test that runs on whichever major is installed rather than one guarded to 1.x. That asymmetry is what hid the bug: the probe had no 2.x coverage at all, so every leg stayed green. Reverting the fix now fails the 2.x leg and passes 1.x. Also drop a `_Sink` test double that was redefined inside one test while an identical one sits at module scope. Generated-By: PostHog Desktop Task-Id: 145ef960-7152-4c88-bed9-3214c268b1d0 --- posthog/mcp/asgi.py | 9 +++++-- posthog/test/mcp/test_session_token.py | 36 +++++++++++++++++++++++--- 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/posthog/mcp/asgi.py b/posthog/mcp/asgi.py index 15e25188e..4e742a143 100644 --- a/posthog/mcp/asgi.py +++ b/posthog/mcp/asgi.py @@ -263,13 +263,18 @@ def _app_was_already_built(server: Any) -> bool: 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 moved it there), so check both. + 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.""" - for candidate in (server, getattr(server, "_mcp_server", None)): + 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 diff --git a/posthog/test/mcp/test_session_token.py b/posthog/test/mcp/test_session_token.py index 94c769545..10d79cbac 100644 --- a/posthog/test/mcp/test_session_token.py +++ b/posthog/test/mcp/test_session_token.py @@ -12,8 +12,10 @@ 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 @@ -501,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, @@ -830,3 +828,33 @@ def test_no_instrument_warning_when_app_not_yet_built(): 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" + ) From 5cf9629c8ac36d559c6041c15a08167f7a5e2b1e Mon Sep 17 00:00:00 2001 From: Georgis Andonis Date: Fri, 21 Aug 2026 16:28:15 +0300 Subject: [PATCH 3/3] refactor(mcp): share the request unwrap with get_request_headers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_is_sse_request` needs the request object itself (for query params), not a header bag, so it hand-rolled the extra -> ctx -> request traversal that `request_headers` already owns. Lift that step into `get_request()` and have both call it, so only one place knows the shape — the same reason `_instrument_v2` already routes its header read through this module. The HTTP-ness probe in `prepare_request` now uses `get_request` too. It only ever asked "is there a request", so riding on the header-bag contract was indirect as well as wasteful: it built and iterated a dict per request to answer a question two getattrs settle. Generated-By: PostHog Desktop Task-Id: 145ef960-7152-4c88-bed9-3214c268b1d0 --- posthog/mcp/_instrumentation.py | 10 +++------- posthog/mcp/request_headers.py | 25 +++++++++++++++++++------ 2 files changed, 22 insertions(+), 13 deletions(-) diff --git a/posthog/mcp/_instrumentation.py b/posthog/mcp/_instrumentation.py index bd9f2e4bb..be274c6cc 100644 --- a/posthog/mcp/_instrumentation.py +++ b/posthog/mcp/_instrumentation.py @@ -22,7 +22,7 @@ from ._intent import resolve_tool_call_intent, set_event_intent from ._internal import MCPAnalyticsData, handle_identify, resolve_event_properties from .logger import log, warn -from .request_headers import get_request_headers +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, resolve_session_id_with_source @@ -277,12 +277,8 @@ def _is_sse_request(extra: Optional[Dict[str, Any]]) -> bool: 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.""" - if not isinstance(extra, dict): - return False try: - params = getattr( - getattr(extra.get("ctx"), "request", None), "query_params", None - ) + 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 @@ -360,7 +356,7 @@ async def prepare_request( ) if ( session_source == "generated" - and get_request_headers(extra) is not None + and get_request(extra) is not None and not _is_sse_request(extra) ): _warn_stateless_session_not_wired(data) 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)