fix(mcp): warn when stateless session middleware never attached

.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`.

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.

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.

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)

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)

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."""

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

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)