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 00000000..83c7ff53 --- /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()` now warns when `streamable_http_app()` was already called before it ran, and a one-time runtime warning fires the first time a tool call arrives over HTTP with no session id. Both point to the manual fix (`app.add_middleware(PostHogMcpStatelessSessionMiddleware)`), which is now documented in `posthog/mcp/README.md`. No behavior change on correctly-wired servers or stdio. diff --git a/examples/mcp_stateless.py b/examples/mcp_stateless.py index 943c44c1..d0db826e 100644 --- a/examples/mcp_stateless.py +++ b/examples/mcp_stateless.py @@ -37,10 +37,14 @@ 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 and sessions fragment silently. Add the +# middleware 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, ... +# +# 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 00000000..beea7564 --- /dev/null +++ b/posthog/mcp/README.md @@ -0,0 +1,72 @@ +# 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: + +```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, ... +``` + +### 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), a warning is logged. +- **At runtime** — the first time a tool call arrives over HTTP with no session id and + PostHog falls back to a per-process `generated` session, a one-time warning is logged. + +Both point back to `app.add_middleware(PostHogMcpStatelessSessionMiddleware)`. Warnings +go through the logger you pass via `MCPAnalyticsOptions(logger=...)`. diff --git a/posthog/mcp/_instrument_fastmcp.py b/posthog/mcp/_instrument_fastmcp.py index 6fd871f2..b52fb94c 100644 --- a/posthog/mcp/_instrument_fastmcp.py +++ b/posthog/mcp/_instrument_fastmcp.py @@ -110,6 +110,7 @@ async def wrapped( request=request, extra=extra, token=token, + http_request=_has_http_request(context), ) missing_name = resolve_missing_capability_tool_name(data.options) @@ -242,6 +243,7 @@ async def list_handler(req: Any) -> Any: request=request, extra=extra, token=token, + http_request=_low_level_has_http_request(server), ) start = time.monotonic() @@ -445,3 +447,20 @@ def _mcp_session_id(context: Any) -> Optional[str]: except Exception: # noqa: BLE001 pass return None + + +def _has_http_request(context: Any) -> bool: + """True when this call arrived over an HTTP transport (a request object is on the + request context). stdio has none, so this distinguishes a legitimate per-process + session from an HTTP server whose stateless mint middleware never attached.""" + try: + return getattr(context.request_context, "request", None) is not None + except Exception: # noqa: BLE001 + return False + + +def _low_level_has_http_request(server: Any) -> bool: + """HTTP-transport check for the ``tools/list`` seam, which runs on the underlying + low-level server rather than a FastMCP ``Context`` (see ``_low_level_session_id``).""" + ctx = _low_level_request_context(server) + return getattr(ctx, "request", None) is not None diff --git a/posthog/mcp/_instrument_lowlevel.py b/posthog/mcp/_instrument_lowlevel.py index cbe6c500..6812b501 100644 --- a/posthog/mcp/_instrument_lowlevel.py +++ b/posthog/mcp/_instrument_lowlevel.py @@ -113,6 +113,7 @@ async def handler(req: Any) -> Any: request=request, extra=extra, token=token, + http_request=_has_http_request(server), ) missing_name = resolve_missing_capability_tool_name(data.options) @@ -254,6 +255,7 @@ async def handler(req: Any) -> Any: request=request, extra=extra, token=token, + http_request=_has_http_request(server), ) start = time.monotonic() @@ -402,3 +404,11 @@ def _mcp_session_id(server: Any) -> Optional[str]: except Exception: # noqa: BLE001 pass return None + + +def _has_http_request(server: Any) -> bool: + """True when the current request arrived over an HTTP transport (a request object + is on the request context). stdio has none — this tells a legitimate per-process + session apart from an HTTP server whose stateless mint middleware never attached.""" + ctx = _request_context(server) + return getattr(ctx, "request", None) is not None diff --git a/posthog/mcp/_instrumentation.py b/posthog/mcp/_instrumentation.py index ff5380ab..eb78cd33 100644 --- a/posthog/mcp/_instrumentation.py +++ b/posthog/mcp/_instrumentation.py @@ -240,6 +240,29 @@ def resolve_session_and_client( return token, client_name, client_version, protocol_version +def _warn_stateless_session_not_wired(data: MCPAnalyticsData) -> None: + """Warn once per server when a tool call/listing arrives over HTTP with no + session id at all, so PostHog fell back to a per-process ``generated`` session. + + 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 + log( + "Warning: an MCP tool request arrived over 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. Fix it by adding the " + "middleware to your app explicitly: " + "app.add_middleware(PostHogMcpStatelessSessionMiddleware). " + "See posthog/mcp/README.md (stateless / multi-pod servers)." + ) + + async def prepare_request( data: MCPAnalyticsData, *, @@ -250,6 +273,7 @@ async def prepare_request( extra: Optional[Dict[str, Any]], token: Optional[SessionTokenPayload] = None, protocol_version: Optional[str] = None, + http_request: bool = False, ) -> str: """Resolve the session id, run identify, then lazily emit initialize. Returns the session id to stamp on the event for this request. @@ -262,8 +286,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``.)""" + the Python SDK handles initialize in the session layer, not ``request_handlers``.) + + ``http_request`` marks requests that arrived over an HTTP transport (vs stdio). + When such a request carries no token and no ``mcp_session_id`` and resolves to a + per-process ``generated`` session, the stateless mint middleware isn't attached — + we warn once so the otherwise-silent misconfiguration surfaces.""" session_id = await resolve_session_id(data, mcp_session_id, token=token) + if ( + http_request + and token is None + and not mcp_session_id + and data.session_source == "generated" + ): + _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 5c529541..e8ffbb80 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 7139751a..0e8250bb 100644 --- a/posthog/mcp/asgi.py +++ b/posthog/mcp/asgi.py @@ -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,30 @@ def autowire_stateless_mint(server: Any) -> None: log(f"PostHog MCP: could not auto-wire stateless mint on {attr} - {error}") +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. + + FastMCP lazily creates ``_session_manager`` the first time ``streamable_http_app()`` + is called, so a non-``None`` value here means the app already exists without our + middleware. Best-effort and guarded — an SDK that doesn't expose this attribute + just yields no warning.""" + try: + if getattr(server, "_session_manager", None) is None: + return + except Exception: # noqa: BLE001 - never let a probe break instrument() + return + log( + "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). " + "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/test/mcp/test_session_token.py b/posthog/test/mcp/test_session_token.py index cf536e80..411adbed 100644 --- a/posthog/test/mcp/test_session_token.py +++ b/posthog/test/mcp/test_session_token.py @@ -12,6 +12,8 @@ PostHogMcpStatelessSessionMiddleware, get_mcp_session, ) +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, @@ -540,3 +542,115 @@ 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 ----- + + +def _capture_logs(): + """Route the SDK logger into a list, restoring the previous sink after.""" + logs: list[str] = [] + set_logger(logs.append) + return logs + + +async def test_prepare_request_warns_once_on_sessionless_http_request(): + """An HTTP request that resolves to a per-process `generated` session (no token, + no Mcp-Session-Id) is the fingerprint of a stateless server whose mint middleware + never attached. That used to be silent; it must now warn -- but only once, so the + log isn't flooded on every subsequent request.""" + logs = _capture_logs() + try: + 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={}, + token=None, + http_request=True, + ) + finally: + set_logger(None) + + warnings = [m for m in logs if "no session id" in m] + assert len(warnings) == 1 + assert "add_middleware(PostHogMcpStatelessSessionMiddleware)" in warnings[0] + assert data.warned_no_stateless_session is True + + +async def test_prepare_request_does_not_warn_for_stdio(): + """stdio has no HTTP request, so a generated per-process session is correct -- + never warn there (that would be noise on the common local-dev path).""" + logs = _capture_logs() + try: + data = _data() + await prepare_request( + data, + mcp_session_id=None, + client_name=None, + client_version=None, + request={"method": "tools/call", "params": {}}, + extra={}, + token=None, + http_request=False, + ) + finally: + set_logger(None) + + assert not [m for m in logs if "no session id" in m] + assert data.warned_no_stateless_session is False + + +async def test_prepare_request_does_not_warn_when_token_present(): + """A correctly-wired stateless server replays our token, so the session resolves + from it -- no warning even though the request came over HTTP.""" + logs = _capture_logs() + try: + data = _data() + token = decode_session_id( + encode_session_id(SessionTokenPayload(session_id="ses_ok")) + ) + await prepare_request( + data, + mcp_session_id=None, + client_name=None, + client_version=None, + request={"method": "tools/call", "params": {}}, + extra={}, + token=token, + http_request=True, + ) + finally: + set_logger(None) + + assert not [m for m in logs if "no session id" in m] + + +def test_autowire_warns_when_app_built_before_instrument(): + """The ordering trap: building streamable_http_app() before instrument() leaves the + live app without our middleware, and wrapping the factory afterward can't retrofit + it. instrument() must warn instead of failing silently.""" + pytest.importorskip("starlette.testclient") + from mcp.server.fastmcp import FastMCP + + from posthog.mcp import instrument + + class _Sink: + def capture(self, *_: object, **__: object) -> None: + pass + + srv = FastMCP("posthog-ordering-trap", stateless_http=True, json_response=True) + + # Build the app BEFORE instrument() -- the customer's failure mode. + srv.streamable_http_app() + + logs: list[str] = [] + instrument(srv, _Sink(), MCPAnalyticsOptions(logger=logs.append)) + + assert any( + "streamable_http_app() was called before instrument()" in m for m in logs + )