diff --git a/.sampo/changesets/mcp-client-conversation-session.md b/.sampo/changesets/mcp-client-conversation-session.md new file mode 100644 index 000000000..342c19abf --- /dev/null +++ b/.sampo/changesets/mcp-client-conversation-session.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: minor +--- + +MCP analytics now groups tool calls into sessions by the client's own conversation id in `_meta`: ChatGPT's `openai/session`, Codex's `sessionId`, and VS Code's `vscode.conversationId`, each read only from the client its `User-Agent` or client name identifies. The client's id wins over a conversation handle and a transport session. ChatGPT gets no `conversation_id` argument and no handle, so its tool schemas and results stay unchanged. Events from `instrument()` now carry `$mcp_session_source` (`client_metadata`, `conversation`, `transport` or `generated`), so session fragmentation can be measured. `PostHogMCP.prepare_tool_list` and `prepare_tool_call` take the request's `headers` (and `client_info`), and `capture_tool_call` takes `request_meta`. diff --git a/posthog/mcp/README.md b/posthog/mcp/README.md index bd179597d..a1dd60f83 100644 --- a/posthog/mcp/README.md +++ b/posthog/mcp/README.md @@ -40,6 +40,16 @@ Use a client dedicated to MCP analytics if the application also captures unrelat A valid echoed conversation handle takes precedence over transport sessions. A request that already carries a transport session or a PostHog session token does not mint a new handle or append a prompt-back block. Requests carrying neither use the conversation fallback. +Some MCP clients name the conversation in each tool call's `params._meta`: ChatGPT's +`openai/session`, Codex's `sessionId`, and VS Code's `vscode.conversationId`. The SDK reads each key +only from a request whose `User-Agent` or client name identifies that client, and derives +`$session_id` from it ahead of any conversation handle or transport session. ChatGPT sends its id on +every call, so its `tools/list` has no `conversation_id` argument and its results get no handle, even +with `enable_conversation_id` on. Each event records which source grouped it in +`$mcp_session_source`: `client_metadata`, `conversation`, `transport`, or `generated`. Custom events +from `McpAnalytics.capture()` carry the instance's own source. The rules are the sdk-specs "Session +identity" requirement, and posthog-js `docs/adr/0014` records the reasoning. + Intent, model capture, conversation correlation, MCP exception capture, and `$mcp_initialize` are on by default. Missing-capability reporting and feedback collection remain off. @@ -233,9 +243,12 @@ Model and conversation injection copy tool objects instead of changing their ori Always advertise the returned list and pass the original application tool to `prepare_tool_call()`. Repeatedly preparing the original list preserves ownership. -Pass an existing transport or request session as `session_id`. A valid echoed -`conversation_id` takes precedence. Otherwise the existing session stays, and no new handle is -minted. `prepare_tool_result()` appends a new handle to the result's `content` and mirrors it +Pass `request_meta` whenever the request supplies `_meta`, and `headers` (plus `client_info` from a +handshake bound to the request, if you hold one) so the SDK can recognize the client; this enables +recognized client metadata and client conversation ids without changing the arguments sent to the +tool. Pass the same `headers` to `prepare_tool_list()`. Pass an existing transport or request +session as `session_id`. A client conversation id or a valid echoed `conversation_id` takes +precedence. Otherwise the existing session stays, and no new handle is minted. `prepare_tool_result()` appends a new handle to the result's `content` and mirrors it into `structuredContent` when the tool declares an output schema. If the result has no channel that can carry a new handle, `conversation_id` is `None` and the derived `session_id` is kept. Set `enable_conversation_id=False` on `PostHogMCP` to keep the previous custom-dispatcher @@ -447,6 +460,9 @@ Prefer it if you're on a recent client. The `get_more_tools` and `send_feedback` virtual tools also use the conversation handle when this option is enabled. +A recognized client's own conversation id in `params._meta` (such as ChatGPT's `openai/session`) +also groups calls on per-request servers, with or without this option, and wins over the handle. + ### How the SDK tells you it's misconfigured The failure used to be silent. It now surfaces two ways: diff --git a/posthog/mcp/__init__.py b/posthog/mcp/__init__.py index 98a73bede..820777f7a 100644 --- a/posthog/mcp/__init__.py +++ b/posthog/mcp/__init__.py @@ -165,6 +165,7 @@ async def capture(self, event: str, properties: Optional[dict] = None) -> None: data, { "session_id": data.session_id, + "session_source": data.session_source, "event_type": MCPAnalyticsEventType.CUSTOM, "event_name": event, "timestamp": datetime.now(timezone.utc), diff --git a/posthog/mcp/_capture.py b/posthog/mcp/_capture.py index ef48ef4a4..76326941f 100644 --- a/posthog/mcp/_capture.py +++ b/posthog/mcp/_capture.py @@ -39,6 +39,7 @@ def capture_event( full_event: Dict[str, Any] = { "id": event_input.get("id") or "", "session_id": session_id, + "session_source": event_input.get("session_source"), "event_type": event_input.get("event_type") or MCPAnalyticsEventType.CUSTOM, "event_name": event_input.get("event_name"), "timestamp": timestamp, diff --git a/posthog/mcp/_client_conversation.py b/posthog/mcp/_client_conversation.py new file mode 100644 index 000000000..d922f6fca --- /dev/null +++ b/posthog/mcp/_client_conversation.py @@ -0,0 +1,173 @@ +"""Client conversation ids: a value an MCP client puts in a ``tools/call`` +request's ``_meta`` to name the conversation the call belongs to. + +``CLIENT_CONVERSATION_ROWS`` is the table of the spec's "Session identity" +requirement, row for row. A row's key is read only from a request its recognition +rule matches, so a generic key such as Codex's ``sessionId`` is never read from a +client that means something else by it. posthog-js keeps the same table. +""" + +from __future__ import annotations + +import re +from typing import Any, Callable, NamedTuple, Optional, Sequence + +from ._transport_identity import CLIENT_USER_AGENT_HEADER +from .request_headers import _to_header_bag, get_request_headers +from .session_token import decode_session_id, read_mcp_session_header + +__all__ = [ + "CLIENT_CONVERSATION_ROWS", + "ClientConversation", + "ClientConversationRow", + "meta_client_name", + "read_client_conversation", + "recognize_client", + "recognize_manual_client", + "request_user_agent", +] + + +class ClientConversation(NamedTuple): + """A recognized client's conversation id, and the ``_meta`` key it came from.""" + + key: str + id: str + + +class ClientConversationRow(NamedTuple): + key: str + read_value: Callable[[Any], Optional[str]] + recognizes: Callable[[Optional[str], Optional[str]], bool] + # The client sends its id on every call, so the SDK advertises no + # ``conversation_id`` to it and mints it no handle. Rows whose clients have + # versions or kinds of conversation that send no id keep the handle. + handle_free: bool + + +_PRINTABLE_ASCII_UP_TO_128 = re.compile(r"[\x20-\x7e]{1,128}") +_UUID = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}") +_NIL_UUID = "00000000-0000-0000-0000-000000000000" +_CHATGPT_USER_AGENT = re.compile(r"openai-mcp/[^\s(]*\s*\(\s*chatgpt\s*[;,)]", re.I) +_VSCODE_PRODUCT_NAMES = ("Visual Studio Code", "Visual Studio Code - Insiders") + + +def _read_printable_ascii(value: Any) -> Optional[str]: + if isinstance(value, str) and _PRINTABLE_ASCII_UP_TO_128.fullmatch(value): + return value + return None + + +def _read_uuid(value: Any) -> Optional[str]: + if not isinstance(value, str): + return None + lowered = value.lower() + return lowered if _UUID.fullmatch(lowered) and lowered != _NIL_UUID else None + + +def _user_agent_starts_with(user_agent: Optional[str], prefixes: Sequence[str]) -> bool: + lowered = (user_agent or "").lower() + return bool(lowered) and any(lowered.startswith(p.lower()) for p in prefixes) + + +def _client_name_is(client_name: Optional[str], names: Sequence[str]) -> bool: + lowered = (client_name or "").lower() + return bool(lowered) and any(lowered == name.lower() for name in names) + + +# Checked in this order; the first row that recognizes a request is its client's row. +CLIENT_CONVERSATION_ROWS = ( + ClientConversationRow( + key="openai/session", + read_value=_read_printable_ascii, + recognizes=lambda user_agent, _: bool( + user_agent and _CHATGPT_USER_AGENT.match(user_agent) + ), + handle_free=True, + ), + ClientConversationRow( + key="sessionId", + read_value=_read_uuid, + recognizes=lambda user_agent, client_name: ( + _user_agent_starts_with(user_agent, ["codex-mcp-client/"]) + or _client_name_is(client_name, ["codex-mcp-client"]) + ), + handle_free=False, + ), + ClientConversationRow( + key="vscode.conversationId", + read_value=_read_uuid, + recognizes=lambda user_agent, client_name: ( + _user_agent_starts_with( + user_agent, [f"{name}/" for name in _VSCODE_PRODUCT_NAMES] + ) + or _client_name_is(client_name, _VSCODE_PRODUCT_NAMES) + ), + handle_free=False, + ), +) + + +def recognize_client( + user_agent: Optional[str], client_name: Optional[str] +) -> Optional[ClientConversationRow]: + """The row of the client a request's ``User-Agent`` or client name identifies. + + The client name is ``$mcp_client_name`` as the adapters resolve it for the + request: its own ``_meta``, then a replayed session token, then a handshake.""" + user_agent = user_agent if isinstance(user_agent, str) else None + client_name = client_name if isinstance(client_name, str) else None + for row in CLIENT_CONVERSATION_ROWS: + if row.recognizes(user_agent, client_name): + return row + return None + + +def read_client_conversation( + row: Optional[ClientConversationRow], meta: Any +) -> Optional[ClientConversation]: + """The recognized client's conversation id in ``meta``, or ``None`` when there + is none or it is not the row's shape.""" + if row is None or not isinstance(meta, dict): + return None + value = row.read_value(meta.get(row.key)) + return ClientConversation(row.key, value) if value is not None else None + + +def request_user_agent(extra: Any) -> Optional[str]: + """The request's ``User-Agent``, or ``None`` off HTTP. Never raises.""" + try: + headers = get_request_headers(extra) + except Exception: # noqa: BLE001 - a header read must not break a request + return None + value = (headers or {}).get(CLIENT_USER_AGENT_HEADER) + return value if isinstance(value, str) else None + + +def recognize_manual_client( + headers: Any, request_meta: Any, handshake_client_name: Optional[str] +) -> Optional[ClientConversationRow]: + """:func:`recognize_client` for a custom dispatcher, from what it passes in: + the request's headers, its ``_meta``, and a handshake ``clientInfo`` name + bound to the request's transport session. The client name follows the + request's ``_meta``, then a replayed session token, then the handshake.""" + bag = _to_header_bag(headers) if headers is not None else None + user_agent = (bag or {}).get(CLIENT_USER_AGENT_HEADER) + token = decode_session_id(read_mcp_session_header(bag)) if bag else None + return recognize_client( + user_agent if isinstance(user_agent, str) else None, + meta_client_name(request_meta) + or (token.client_name if token is not None else None) + or handshake_client_name, + ) + + +def meta_client_name(meta: Any) -> Optional[str]: + """The client name a 2026-07-28 request carries in its own ``_meta``.""" + info = ( + meta.get("io.modelcontextprotocol/clientInfo") + if isinstance(meta, dict) + else None + ) + name = info.get("name") if isinstance(info, dict) else None + return name if isinstance(name, str) and name else None diff --git a/posthog/mcp/_instrument_fastmcp.py b/posthog/mcp/_instrument_fastmcp.py index 78bd1324f..38f06623b 100644 --- a/posthog/mcp/_instrument_fastmcp.py +++ b/posthog/mcp/_instrument_fastmcp.py @@ -179,7 +179,7 @@ async def wrapped( # block on the minting response only. delivered = False if lifecycle.conversation_id: - if data.tool_output_instructions.get(name): + if lifecycle.mirrors_output_instructions: result, delivered = mirror_instructions_into_structured_content( result, lifecycle.conversation_id ) @@ -203,7 +203,13 @@ async def wrapped( # --- tools/list seam --------------------------------------------------------- -def _inject_tool_schemas(server: Any, data: MCPAnalyticsData, tools: list) -> None: +def _inject_tool_schemas( + server: Any, + data: MCPAnalyticsData, + tools: list, + *, + advertise_conversation_id: bool = True, +) -> None: """Advertise the analytics parameters on a listing's tools, in place. Runs on both the client-facing listing and the SDK's internal cache- @@ -218,6 +224,7 @@ def _inject_tool_schemas(server: Any, data: MCPAnalyticsData, tools: list) -> No owns_context=_tool_owns_context(server, tool.name), context_required=True, is_sdk_virtual_tool=False, + advertise_conversation_id=advertise_conversation_id, ) @@ -286,10 +293,20 @@ async def list_handler(req: Any) -> Any: is_first_page=is_first_listing_page(getattr(req, "params", None)), ) - _inject_tool_schemas(server, data, tools) + _inject_tool_schemas( + server, + data, + tools, + advertise_conversation_id=lifecycle.advertises_conversation_id, + ) result = apply_virtual_tool_injection( - result, injection, names, data, schema_field="inputSchema" + result, + injection, + names, + data, + schema_field="inputSchema", + advertise_conversation_id=lifecycle.advertises_conversation_id, ) await lifecycle.record_result( diff --git a/posthog/mcp/_instrument_lowlevel.py b/posthog/mcp/_instrument_lowlevel.py index 00bd4ef4d..d3ec0579e 100644 --- a/posthog/mcp/_instrument_lowlevel.py +++ b/posthog/mcp/_instrument_lowlevel.py @@ -132,7 +132,7 @@ async def handler(req: Any) -> Any: request = request_to_dict(req) extra = {"session_id": mcp_session_id, "ctx": _request_context(server)} try: - session_id = await prepare_request( + session = await prepare_request( data, mcp_session_id=mcp_session_id, client_name=client_name, @@ -152,7 +152,7 @@ async def handler(req: Any) -> Any: except Exception as error: await record_resource_request( data, - session_id, + session, event_type=event_type, request=request, error=error, @@ -166,7 +166,7 @@ async def handler(req: Any) -> Any: await record_resource_request( data, - session_id, + session, event_type=event_type, request=request, response=resource_listing_response(event_type, result), @@ -305,7 +305,7 @@ async def handler(req: Any) -> Any: # when the agent needs the handle, or the retry starts a fresh conversation. delivered = False if lifecycle.conversation_id: - if data.tool_output_instructions.get(name): + if lifecycle.mirrors_output_instructions: call_result, delivered = mirror_instructions_into_structured_content( call_result, lifecycle.conversation_id ) @@ -351,6 +351,7 @@ def _inject_tool_schemas( *, context_required: bool, high_level: Any = None, + advertise_conversation_id: bool = True, ) -> None: """Advertise the analytics parameters on a listing's tools, in place. @@ -373,6 +374,7 @@ def _inject_tool_schemas( context_required=context_required, is_sdk_virtual_tool=False, inject_model=inject_model, + advertise_conversation_id=advertise_conversation_id, ) verdict = data.tool_model_parameter_injected.get(tool.name) if verdict is None: @@ -513,10 +515,16 @@ async def handler(req: Any) -> Any: tools, context_required=context_required, high_level=high_level, + advertise_conversation_id=lifecycle.advertises_conversation_id, ) result = apply_virtual_tool_injection( - result, injection, names, data, schema_field="inputSchema" + result, + injection, + names, + data, + schema_field="inputSchema", + advertise_conversation_id=lifecycle.advertises_conversation_id, ) await lifecycle.record_result( diff --git a/posthog/mcp/_instrument_v2.py b/posthog/mcp/_instrument_v2.py index 58d45f778..df5cc0c57 100644 --- a/posthog/mcp/_instrument_v2.py +++ b/posthog/mcp/_instrument_v2.py @@ -77,7 +77,7 @@ from .asgi import get_mcp_session from .logger import log from .request_headers import get_request_headers -from .session import handshake_session_id +from .session import handshake_session from .session_token import SessionTokenPayload, read_mcp_session_header from .tools import ( get_more_tools_result_text, @@ -204,10 +204,11 @@ async def _record_v2_initialize( ) -> None: try: mcp_session_id, token = _initialize_session(ctx) + params = dict(ctx.params) if isinstance(ctx.params, Mapping) else {} await record_initialize( data, - session_id=await handshake_session_id(data, mcp_session_id, token), - params=dict(ctx.params) if isinstance(ctx.params, Mapping) else {}, + session=await handshake_session(data, mcp_session_id, token), + params=params, result=_to_jsonable(result), extra={"session_id": mcp_session_id, "ctx": ctx}, timestamp=started, @@ -303,6 +304,22 @@ def _ctx_mcp_session_id(ctx: Any) -> Optional[str]: return read_mcp_session_header(get_request_headers(ctx)) +def _is_internal_dispatch(ctx: Any) -> bool: + """True when the SDK invoked this handler itself, not for a client request. + + On the 2026-07-28 streamable-HTTP path the SDK runs the ``tools/list`` handler + while serving a ``tools/call`` (to validate ``Mcp-Param-*`` headers against the + tool's input schema). That call shares the inbound HTTP request, whose required + ``Mcp-Method`` header still names the outer method, so it gives the call away.""" + try: + declared_method = ctx.request.headers.get("mcp-method") + except Exception: # noqa: BLE001 - no HTTP request (stdio, legacy era) means a client call + return False + return declared_method is not None and declared_method != getattr( + ctx, "method", declared_method + ) + + def _resolve_ctx( ctx: Any, ) -> Tuple[Optional[Any], Optional[str], Optional[str], Optional[str], Optional[str]]: @@ -459,11 +476,10 @@ async def wrapped( delivered = False if lifecycle.conversation_id: result, delivered = _deliver_conversation_id( - data, result, - name, lifecycle.conversation_id, lifecycle.minted_conversation_id, + mirror=lifecycle.mirrors_output_instructions, ) await lifecycle.record_result( @@ -506,7 +522,7 @@ def _append_prompt_back(result: Any, conversation_id: str) -> Tuple[Any, bool]: def _deliver_conversation_id( - data: MCPAnalyticsData, result: Any, name: str, conversation_id: str, minted: bool + result: Any, conversation_id: str, minted: bool, *, mirror: bool ) -> Tuple[Any, bool]: """Hand the conversation handle back over both channels a result has: mirrored into ``structuredContent`` on every response (for tools whose @@ -514,7 +530,7 @@ def _deliver_conversation_id( the minting response only. Returns ``(result, delivered)`` — a minted handle the agent never received must not be stamped on the event.""" delivered = False - if data.tool_output_instructions.get(name): + if mirror: result, delivered = mirror_instructions_into_structured_content( result, conversation_id ) @@ -678,11 +694,10 @@ async def handler(ctx: Any, params: Any) -> Any: delivered = False if lifecycle.conversation_id: result, delivered = _deliver_conversation_id( - data, result, - name, lifecycle.conversation_id, lifecycle.minted_conversation_id, + mirror=lifecycle.mirrors_output_instructions, ) await lifecycle.record_result( @@ -709,7 +724,7 @@ async def handler(ctx: Any, params: Any) -> Any: request = params_to_request_dict(method, params, by_alias=True) extra: Dict[str, Any] = {"session_id": mcp_session_id, "ctx": ctx} try: - session_id = await prepare_request( + session = await prepare_request( data, mcp_session_id=mcp_session_id, client_name=client_name, @@ -729,7 +744,7 @@ async def handler(ctx: Any, params: Any) -> Any: except Exception as error: await record_resource_request( data, - session_id, + session, event_type=event_type, request=request, error=error, @@ -743,7 +758,7 @@ async def handler(ctx: Any, params: Any) -> Any: await record_resource_request( data, - session_id, + session, event_type=event_type, request=request, response=resource_listing_response(event_type, result), @@ -783,6 +798,8 @@ async def probe_raw_tool_names(ctx: Any = None) -> Optional[Set[str]]: data.raw_tool_names_probe = probe_raw_tool_names async def handler(ctx: Any, params: Any) -> Any: + if _is_internal_dispatch(ctx): + return await original(ctx, params) token, client_name, client_version, protocol_version, mcp_session_id = ( _resolve_ctx(ctx) ) @@ -831,9 +848,15 @@ async def handler(ctx: Any, params: Any) -> Any: owns_context=owns_context, context_required=context_required, is_sdk_virtual_tool=False, + advertise_conversation_id=lifecycle.advertises_conversation_id, ) result = apply_virtual_tool_injection( - result, injection, names, data, schema_field="input_schema" + result, + injection, + names, + data, + schema_field="input_schema", + advertise_conversation_id=lifecycle.advertises_conversation_id, ) await lifecycle.record_result( diff --git a/posthog/mcp/_instrumentation.py b/posthog/mcp/_instrumentation.py index cd87dd4a2..74af393b4 100644 --- a/posthog/mcp/_instrumentation.py +++ b/posthog/mcp/_instrumentation.py @@ -54,7 +54,20 @@ 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 +from ._client_conversation import ( + ClientConversation, + ClientConversationRow, + meta_client_name, + read_client_conversation, + recognize_client, + request_user_agent, +) +from .session import ( + ResolvedSession, + resolve_session_id, + resolve_session_id_with_source, + touch_connection_session, +) from .session_token import SessionTokenPayload, decode_session_id from .tools import resolve_missing_capability_tool_name from .types import CollectFeedbackOptions, FeedbackReport, ToolInputOptions @@ -243,7 +256,7 @@ def _wrap_response(result: Any) -> Any: async def record_initialize( data: MCPAnalyticsData, *, - session_id: str, + session: ResolvedSession, params: Dict[str, Any], result: Any, extra: Optional[Dict[str, Any]], @@ -261,7 +274,7 @@ async def record_initialize( in the background, as for every other event.""" request = {"method": "initialize", "params": params} - await handle_identify(data, session_id, request, extra) + await handle_identify(data, session.session_id, request, extra) # The option stops only the event. Identify still runs, so later events of # the session keep an identity that only the handshake can resolve. if not data.options.send_initialize_event: @@ -273,10 +286,11 @@ def text(value: Any) -> Optional[str]: client_info = params.get("clientInfo") client_info = client_info if isinstance(client_info, dict) else {} agreed = result.get("protocolVersion") if isinstance(result, dict) else None - data.mark_session_initialized(session_id) + data.mark_session_initialized(session.session_id) event: Dict[str, Any] = { "event_type": MCPAnalyticsEventType.MCP_INITIALIZE, - "session_id": session_id, + "session_id": session.session_id, + "session_source": session.source, "client_name": text(client_info.get("name")), "client_version": text(client_info.get("version")), "protocol_version": text(agreed) or text(params.get("protocolVersion")), @@ -305,10 +319,10 @@ def _first_request_stands_in_for_initialize( conversation-anchored session: it belongs to a chat, not to a handshake, and synthesizing one there is how a single connection used to report two. So a client whose first request already echoes a conversation reports no initialize - here.""" + here. The same goes for a session anchored by the client's own conversation id.""" if data.handshake_observed_in_process: return False - if session_source == "conversation": + if session_source in ("conversation", "client_metadata"): return False scope = getattr(get_request(extra), "scope", None) return not (isinstance(scope, dict) and scope.get(HANDSHAKE_OBSERVER_KEY) is data) @@ -316,8 +330,7 @@ def _first_request_stands_in_for_initialize( async def _maybe_emit_initialize( data: MCPAnalyticsData, - session_id: str, - session_source: str, + session: ResolvedSession, client_name: Optional[str], client_version: Optional[str], extra: Optional[Dict[str, Any]], @@ -328,14 +341,15 @@ async def _maybe_emit_initialize( :func:`_first_request_stands_in_for_initialize`).""" if not data.options.send_initialize_event: return - if not _first_request_stands_in_for_initialize(data, session_source, extra): + if not _first_request_stands_in_for_initialize(data, session.source, extra): return - if session_id in data.initialized_sessions: + if session.session_id in data.initialized_sessions: return - data.mark_session_initialized(session_id) + data.mark_session_initialized(session.session_id) event: Dict[str, Any] = { "event_type": MCPAnalyticsEventType.MCP_INITIALIZE, - "session_id": session_id, + "session_id": session.session_id, + "session_source": session.source, "client_name": client_name, "client_version": client_version, "protocol_version": protocol_version, @@ -387,6 +401,7 @@ async def prime_session( *, mcp_session_id: Optional[str], token: Optional[SessionTokenPayload] = None, + client_conversation: Optional[ClientConversation] = None, ) -> None: """Point the shared per-server session at *this* request before the tool body runs. @@ -397,7 +412,9 @@ async def prime_session( and attribute a custom event to the wrong caller. Emits nothing — it only settles the transport/memory session an in-tool event should belong to. """ - await resolve_session_id(data, mcp_session_id, token=token) + await resolve_session_id( + data, mcp_session_id, token=token, client_conversation=client_conversation + ) def _is_sse_request(extra: Optional[Dict[str, Any]]) -> bool: @@ -451,10 +468,11 @@ async def prepare_request( token: Optional[SessionTokenPayload] = None, protocol_version: Optional[str] = None, conversation_id: Optional[str] = None, -) -> str: - """Resolve the session id, run identify, then emit initialize if this request - has to stand in for an unobserved handshake. Returns the session id to stamp on - the event for this request. + client_conversation: Optional[ClientConversation] = None, +) -> ResolvedSession: + """Resolve the session, run identify, then emit initialize if this request + has to stand in for an unobserved handshake. Returns the session to stamp on + the event for this request: its id, and the source that grouped it. ``conversation_id`` is the agent's handle for this request, and when present it anchors the session (ADR-0004) so every event of the request — initialize @@ -483,26 +501,24 @@ async def prepare_request( 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 + session = await resolve_session_id_with_source( + data, + mcp_session_id, + token=token, + conversation_id=conversation_id, + client_conversation=client_conversation, ) if ( - session_source == "generated" + session.source == "generated" and get_request(extra) is not None and not _is_sse_request(extra) ): _warn_stateless_session_not_wired(data) - await handle_identify(data, session_id, request, extra) + await handle_identify(data, session.session_id, request, extra) await _maybe_emit_initialize( - data, - session_id, - session_source, - client_name, - client_version, - extra, - protocol_version, + data, session, client_name, client_version, extra, protocol_version ) - return session_id + return session @dataclass(frozen=True) @@ -518,6 +534,8 @@ class ToolCallLifecycle: name: str arguments: Optional[Dict[str, Any]] request_meta: Optional[Dict[str, Any]] + client: Optional[ClientConversationRow] + client_conversation: Optional[ClientConversation] allow_self_reported_model: bool analytics_owned_parameters: Optional[FrozenSet[str]] request: Dict[str, Any] @@ -544,7 +562,16 @@ def is_missing_capability(self) -> bool: def is_feedback(self) -> bool: return self.feedback_name is not None and self.name == self.feedback_name - async def prepare_session(self, conversation_id: Optional[str]) -> str: + @property + def mirrors_output_instructions(self) -> bool: + """Whether a handle goes into ``structuredContent`` too. A handle-free + client's listing never declared ``_mcp_instructions``.""" + handle_free = self.client is not None and self.client.handle_free + return not handle_free and bool( + self.data.tool_output_instructions.get(self.name) + ) + + async def prepare_session(self, conversation_id: Optional[str]) -> ResolvedSession: return await prepare_request( self.data, mcp_session_id=self.mcp_session_id, @@ -555,12 +582,23 @@ async def prepare_session(self, conversation_id: Optional[str]) -> str: extra=self.extra, token=self.token, conversation_id=conversation_id, + client_conversation=self.client_conversation, ) async def prime_session(self) -> None: await prime_session( - self.data, mcp_session_id=self.mcp_session_id, token=self.token + self.data, + mcp_session_id=self.mcp_session_id, + token=self.token, + client_conversation=self.client_conversation, ) + on_one_client_connection = ( + get_request(self.extra) is None + and self.token is None + and not self.mcp_session_id + ) + if self.client_conversation is not None and on_one_client_connection: + await touch_connection_session(self.data) def virtual_result_texts(self, primary_text: str) -> List[str]: """Build the text payload for an SDK virtual-tool result.""" @@ -577,10 +615,10 @@ async def record_missing_capability( self, *, conversation_id_delivered: bool = False ) -> None: conversation_id = self._anchored_conversation_id(conversation_id_delivered) - session_id = await self.prepare_session(conversation_id) + session = await self.prepare_session(conversation_id) await record_missing_capability( self.data, - session_id, + session, conversation_id=conversation_id, tool_name=self.missing_name or self.name, context=(self.arguments or {}).get("context"), @@ -599,10 +637,10 @@ async def record_feedback(self, *, conversation_id_delivered: bool = False) -> s whether or not the handler raises.""" report = parse_feedback_report(self.arguments, self.feedback_options) conversation_id = self._anchored_conversation_id(conversation_id_delivered) - session_id = await self.prepare_session(conversation_id) + session = await self.prepare_session(conversation_id) await record_feedback( self.data, - session_id, + session, conversation_id=conversation_id, report=report, tool_name=self.feedback_name or self.name, @@ -619,10 +657,10 @@ async def record_error(self, error: Any, duration_ms: float) -> None: # A freshly minted handle cannot anchor or be captured when dispatch # raised: no adapter had an opportunity to deliver it to the agent. conversation_id = self._anchored_conversation_id(False) - session_id = await self.prepare_session(conversation_id) + session = await self.prepare_session(conversation_id) await record_tool_call( self.data, - session_id, + session, name=self.name, arguments=self.arguments, request_meta=self.request_meta, @@ -642,10 +680,10 @@ async def record_result( self, result: Any, duration_ms: float, *, conversation_id_delivered: bool ) -> None: conversation_id = self._anchored_conversation_id(conversation_id_delivered) - session_id = await self.prepare_session(conversation_id) + session = await self.prepare_session(conversation_id) await record_tool_call( self.data, - session_id, + session, name=self.name, arguments=self.arguments, request_meta=self.request_meta, @@ -693,15 +731,29 @@ def start_tool_call_lifecycle( ), arguments, ) - # A carried session stays stable until the agent supplies its own handle. - has_carried_session = token is not None or bool(mcp_session_id) - if minted and has_carried_session: + client = recognize_client( + request_user_agent(extra), meta_client_name(request_meta) or client_name + ) + client_conversation = read_client_conversation(client, request_meta) + # The client's own id wins, so a handle the agent echoes alongside it is + # ignored. A carried session stays stable until the agent supplies its own + # handle, and a handle-free client names its conversation itself, so neither + # is minted one. + if client_conversation is not None: + conversation_id, minted = None, False + elif minted and ( + token is not None + or bool(mcp_session_id) + or (client is not None and client.handle_free) + ): conversation_id, minted = None, False return ToolCallLifecycle( data=data, name=name, arguments=arguments, request_meta=request_meta, + client=client, + client_conversation=client_conversation, allow_self_reported_model=allow_self_reported_model, analytics_owned_parameters=analytics_owned_parameters, request=build_tool_call_request(name, arguments), @@ -722,7 +774,7 @@ def start_tool_call_lifecycle( async def record_tool_call( data: MCPAnalyticsData, - session_id: str, + session: ResolvedSession, *, name: str, arguments: Optional[Dict[str, Any]], @@ -745,7 +797,8 @@ async def record_tool_call( request = build_tool_call_request(name, arguments) event: Dict[str, Any] = { "event_type": MCPAnalyticsEventType.MCP_TOOLS_CALL, - "session_id": session_id, + "session_id": session.session_id, + "session_source": session.source, "resource_name": name, "tool_description": data.tool_descriptions.get(name), "tool_category": data.tool_categories.get(name), @@ -887,7 +940,13 @@ def virtual_tool_descriptor( def append_virtual_tool_by_kind( - result: Any, kind: str, name: str, data: MCPAnalyticsData, *, schema_field: str + result: Any, + kind: str, + name: str, + data: MCPAnalyticsData, + *, + schema_field: str, + advertise_conversation_id: bool = True, ) -> Any: """Add a virtual tool to a ``tools/list`` result and return the result to serve. ``schema_field`` is the SDK major's spelling of the input schema @@ -915,6 +974,7 @@ def append_virtual_tool_by_kind( owns_context=True, context_required=True, is_sdk_virtual_tool=True, + advertise_conversation_id=advertise_conversation_id, ) return append_virtual_tool(result, tool) @@ -926,6 +986,7 @@ def apply_virtual_tool_injection( data: MCPAnalyticsData, *, schema_field: str, + advertise_conversation_id: bool = True, ) -> Any: """Append every virtual tool this page won, recording each name for the ``$mcp_tools_list`` event. Missing-capability first: every call path tests @@ -936,7 +997,12 @@ def apply_virtual_tool_injection( if name is None: continue result = append_virtual_tool_by_kind( - result, kind, name, data, schema_field=schema_field + result, + kind, + name, + data, + schema_field=schema_field, + advertise_conversation_id=advertise_conversation_id, ) names.append(name) return result @@ -1159,9 +1225,13 @@ def mutate_tool_schema( context_required: bool, is_sdk_virtual_tool: bool, inject_model: bool = True, + advertise_conversation_id: bool = True, ) -> None: """Apply the common analytics schema pipeline and write it back in place. + ``advertise_conversation_id`` is False for a handle-free client's listing + (see :func:`lists_conversation_id`). + The adapter explicitly supplies its SDK model's schema attribute and its own ownership decision. Those are the parts that differ across MCP generations; context/conversation mutation and output-channel bookkeeping do not. @@ -1187,21 +1257,30 @@ def mutate_tool_schema( # Discard any earlier registered-tool verdict: middleware may now # advertise or dispatch an application-owned field with the same name. data.tool_model_parameter_injected[tool.name] = False - if data.options.enable_conversation_id and not schema_has_param( - schema, "conversation_id" - ): + advertises_conversation_id = ( + advertise_conversation_id and data.options.enable_conversation_id + ) + if advertises_conversation_id and not schema_has_param(schema, "conversation_id"): schema = add_conversation_id_to_schema(schema, tool.name) if schema is not original_schema: try: setattr(tool, schema_attribute, schema) except Exception: # noqa: BLE001 - some schema attrs may be read-only log(f"WARN: could not set {schema_attribute} on tool {tool.name}") - if data.options.enable_conversation_id: + if advertises_conversation_id: data.tool_output_instructions[tool.name] = add_instructions_to_output_schema( tool ) +def lists_conversation_id(extra: Any, client_name: Optional[str]) -> bool: + """Whether a listing advertises ``conversation_id``. A handle-free client + names the conversation in ``_meta``, so the argument would only add a second, + weaker grouping key.""" + client = recognize_client(request_user_agent(extra), client_name) + return client is None or not client.handle_free + + def _mutate_model_schema( data: MCPAnalyticsData, name: str, schema: Any, *, required: bool ) -> Any: @@ -1251,17 +1330,21 @@ class ToolsListLifecycle: """Common capture lifecycle for one client-facing tools/list dispatch.""" data: MCPAnalyticsData - session_id: str + session: ResolvedSession request: Dict[str, Any] extra: Dict[str, Any] client_name: Optional[str] client_version: Optional[str] protocol_version: Optional[str] + @property + def advertises_conversation_id(self) -> bool: + return lists_conversation_id(self.extra, self.client_name) + async def record_error(self, error: Any, duration_ms: float) -> None: await record_tools_list( self.data, - self.session_id, + self.session, names=[], request=self.request, duration_ms=duration_ms, @@ -1283,7 +1366,7 @@ async def record_result( ) -> None: await record_tools_list( self.data, - self.session_id, + self.session, names=names, request=self.request, result=result, @@ -1308,7 +1391,7 @@ async def start_tools_list_lifecycle( client_version: Optional[str], protocol_version: Optional[str], ) -> ToolsListLifecycle: - session_id = await prepare_request( + session = await prepare_request( data, mcp_session_id=mcp_session_id, client_name=client_name, @@ -1320,7 +1403,7 @@ async def start_tools_list_lifecycle( ) return ToolsListLifecycle( data=data, - session_id=session_id, + session=session, request=request, extra=extra, client_name=client_name, @@ -1331,7 +1414,7 @@ async def start_tools_list_lifecycle( async def record_missing_capability( data: MCPAnalyticsData, - session_id: str, + session: ResolvedSession, *, conversation_id: Optional[str] = None, tool_name: str, @@ -1350,7 +1433,8 @@ async def record_missing_capability( request = build_tool_call_request(tool_name, arguments) event: Dict[str, Any] = { "event_type": MCPAnalyticsEventType.MCP_MISSING_CAPABILITY, - "session_id": session_id, + "session_id": session.session_id, + "session_source": session.source, "conversation_id": conversation_id, "resource_name": tool_name, "parameters": build_captured_mcp_parameters( @@ -1381,7 +1465,7 @@ async def record_missing_capability( async def record_feedback( data: MCPAnalyticsData, - session_id: str, + session: ResolvedSession, *, conversation_id: Optional[str] = None, report: FeedbackReport, @@ -1403,7 +1487,8 @@ async def record_feedback( request = build_tool_call_request(tool_name, arguments) event: Dict[str, Any] = { "event_type": MCPAnalyticsEventType.MCP_FEEDBACK, - "session_id": session_id, + "session_id": session.session_id, + "session_source": session.source, "conversation_id": conversation_id, "resource_name": tool_name, "client_name": client_name, @@ -1436,7 +1521,7 @@ async def record_feedback( async def record_tools_list( data: MCPAnalyticsData, - session_id: str, + session: ResolvedSession, *, names: List[str], request: Dict[str, Any], @@ -1454,7 +1539,8 @@ async def record_tools_list( try: event: Dict[str, Any] = { "event_type": MCPAnalyticsEventType.MCP_TOOLS_LIST, - "session_id": session_id, + "session_id": session.session_id, + "session_source": session.source, "listed_tool_names": names, "parameters": build_captured_mcp_parameters(request), "response": tools_list_envelope(result), @@ -1486,7 +1572,7 @@ def resource_listing_response(event_type: str, result: Any) -> Any: async def record_resource_request( data: MCPAnalyticsData, - session_id: str, + session: ResolvedSession, *, event_type: str, request: Dict[str, Any], @@ -1504,7 +1590,8 @@ async def record_resource_request( uri = params.get("uri") if isinstance(params, dict) else None event: Dict[str, Any] = { "event_type": event_type, - "session_id": session_id, + "session_id": session.session_id, + "session_source": session.source, "resource_name": uri if event_type == MCPAnalyticsEventType.MCP_RESOURCES_READ else None, diff --git a/posthog/mcp/_posthog_events.py b/posthog/mcp/_posthog_events.py index 51fc5ff75..ef706b39c 100644 --- a/posthog/mcp/_posthog_events.py +++ b/posthog/mcp/_posthog_events.py @@ -33,6 +33,11 @@ _P = PostHogMCPAnalyticsProperty +# `$mcp_session_source` is `conversation`, `transport`, `client_metadata` or +# `generated`. A session token and a transport MCP session id are both transport +# carried, so they share one value. +_WIRE_SESSION_SOURCE = {"token": "transport", "mcp": "transport"} + def _get_distinct_id(event: Event) -> str: return ( @@ -85,6 +90,11 @@ def _add_session_id(event: Event, properties: Dict[str, Any]) -> None: session_id = event.get("session_id") if isinstance(session_id, str) and len(session_id) > 0: properties[_P.SESSION_ID] = session_id + session_source = event.get("session_source") + if session_source: + properties[_P.SESSION_SOURCE] = _WIRE_SESSION_SOURCE.get( + session_source, session_source + ) def _add_conversation_id(event: Event, properties: Dict[str, Any]) -> None: diff --git a/posthog/mcp/asgi.py b/posthog/mcp/asgi.py index a66b41113..6870702bf 100644 --- a/posthog/mcp/asgi.py +++ b/posthog/mcp/asgi.py @@ -51,7 +51,7 @@ from ._instrumentation import HANDSHAKE_OBSERVER_KEY, record_initialize from ._internal import MCPAnalyticsData, get_server_tracking_data, tracking_key from .logger import log, warn -from .session import handshake_session_id, new_session_id +from .session import handshake_session, new_session_id from .session_token import ( MCP_SESSION_HEADER, SessionTokenPayload, @@ -345,12 +345,11 @@ async def _record_initialize( from starlette.requests import Request header = session_header + params = _initialize_params(message)[0] await record_initialize( data, - session_id=await handshake_session_id( - data, header, decode_session_id(header) - ), - params=_initialize_params(message)[0], + session=await handshake_session(data, header, decode_session_id(header)), + params=params, result=None, extra={"session_id": header, "ctx": _InitializeContext(Request(scope))}, timestamp=started, diff --git a/posthog/mcp/constants.py b/posthog/mcp/constants.py index de359e693..54887a767 100644 --- a/posthog/mcp/constants.py +++ b/posthog/mcp/constants.py @@ -101,6 +101,7 @@ class PostHogMCPAnalyticsProperty: SERVER_BUILD = "$mcp_server_build" SERVER_VERSION = "$mcp_server_version" SESSION_ID = "$session_id" + SESSION_SOURCE = "$mcp_session_source" SOURCE = "$mcp_source" TOOL_CATEGORY = "$mcp_tool_category" TOOL_DESCRIPTION = "$mcp_tool_description" diff --git a/posthog/mcp/posthog_mcp.py b/posthog/mcp/posthog_mcp.py index 69951eade..5d483708c 100644 --- a/posthog/mcp/posthog_mcp.py +++ b/posthog/mcp/posthog_mcp.py @@ -14,7 +14,7 @@ import copy from datetime import datetime, timezone from dataclasses import dataclass -from typing import Any, Dict, List, Optional, Set, Tuple, Union +from typing import Any, Dict, List, Mapping, Optional, Set, Tuple, Union from typing_extensions import Unpack @@ -70,7 +70,16 @@ resolve_collect_feedback_options, resolve_send_feedback_tool_name, ) -from .session import derive_session_id_from_conversation +from ._client_conversation import ( + meta_client_name, + read_client_conversation, + recognize_client, + recognize_manual_client, +) +from .session import ( + derive_session_id_from_client_conversation, + derive_session_id_from_conversation, +) from .tools import build_report_missing_descriptor from .types import ( CollectFeedbackOptions, @@ -213,8 +222,14 @@ def capture_tool_call( groups: Optional[Dict[str, str]] = None, properties: Optional[JsonRecord] = None, timestamp: Optional[datetime] = None, + request_meta: Optional[JsonRecord] = None, ) -> None: - """Capture a tool invocation. Emits ``$mcp_tool_call`` (+ ``$exception`` on error).""" + """Capture a tool invocation. Emits ``$mcp_tool_call`` (+ ``$exception`` on error). + + ``request_meta`` is the ``tools/call`` request's ``_meta``. With the + ``client_user_agent`` or ``_meta`` client name that recognizes the client, + its conversation id (such as ChatGPT's ``openai/session``) sets + ``$session_id`` in place of ``session_id``.""" event = self._base_event( MCPAnalyticsEventType.MCP_TOOLS_CALL, distinct_id, @@ -227,6 +242,16 @@ def capture_tool_call( client_user_agent, vendor_client, ) + client_conversation = read_client_conversation( + recognize_client(client_user_agent, meta_client_name(request_meta)), + request_meta, + ) + if client_conversation is not None: + event["session_id"] = derive_session_id_from_client_conversation( + *client_conversation + ) + event["session_source"] = "client_metadata" + event["conversation_id"] = None event["resource_name"] = tool_name event["tool_description"] = tool_description event["tool_category"] = category @@ -428,6 +453,7 @@ def prepare_tool_list( context: Union[bool, MCPAnalyticsContextOptions] = True, report_missing: bool = False, collect_feedback: bool = False, + headers: Any = None, ) -> List[Any]: """Inject the ``context`` argument into every tool so agents state their intent (captured as ``$mcp_intent``), and optionally append the @@ -500,8 +526,11 @@ def prepare_tool_list( # Read ownership before any analytics field lands on the schemas. conversation_ownership = _collect_conversation_ownership(prepared) prepared = self._inject_models(prepared) - prepared = self._inject_conversation(prepared, conversation_ownership) - return prepared + # A handle-free client (ChatGPT, by its ``User-Agent`` in ``headers``) + # names the conversation in ``_meta`` instead. + client = recognize_manual_client(headers, None, None) + advertise = client is None or not client.handle_free + return self._inject_conversation(prepared, conversation_ownership, advertise) def _warn_virtual_tool_collision( self, @@ -530,6 +559,8 @@ def prepare_tool_call( request_meta: Optional[JsonRecord] = None, original_tool: Any = None, session_id: Optional[str] = None, + headers: Any = None, + client_info: Optional[Mapping[str, Any]] = None, ) -> PreparedToolCall: """Pull the agent's intent off the injected ``context`` argument, strip ``context`` from the arguments, and flag the ``get_more_tools`` and @@ -564,7 +595,14 @@ def prepare_tool_call( ``session_id`` is a session carried by the request or transport. A valid echoed ``conversation_id`` takes precedence. Otherwise the carried session - stays, and no new handle is minted.""" + stays, and no new handle is minted. + + ``request_meta`` is the request's ``_meta``. With the request's + ``headers`` (its ``User-Agent``) or ``client_info`` (the ``initialize`` + ``clientInfo`` bound to its transport session) recognizing the client, a + client conversation id in it (such as ChatGPT's ``openai/session``) sets + the session ahead of an echoed handle and ``session_id``, and mints no + handle.""" raw_context = (args or {}).get("context") intent = ( raw_context.strip() @@ -597,14 +635,30 @@ def prepare_tool_call( can_read_conversation = self._enable_conversation_id and ( ownership is None or ownership.conversation_id ) - resolved_conversation_id, minted = resolve_conversation_id( - can_read_conversation, args + client = recognize_manual_client( + headers, request_meta, (client_info or {}).get("name") + ) + handle_free = client is not None and client.handle_free + client_conversation = read_client_conversation(client, request_meta) + # Same order as instrument(): the client's id, then the handle, then the + # carried session. A carried session or a handle-free client is minted + # no handle. + resolved_conversation_id, minted = ( + (None, False) + if client_conversation is not None + else resolve_conversation_id(can_read_conversation, args) ) minted_conversation_id = ( - resolved_conversation_id if minted and not session_id else None + resolved_conversation_id + if minted and not session_id and not handle_free + else None ) conversation_id = None if minted else resolved_conversation_id - if conversation_id: + if client_conversation is not None: + session_id = derive_session_id_from_client_conversation( + *client_conversation + ) + elif conversation_id: session_id = derive_session_id_from_conversation(conversation_id) if ( self._enable_conversation_id @@ -650,6 +704,7 @@ def prepare_tool_call( _conversation_state=PreparedConversationState( minted_conversation_id=minted_conversation_id, output_instructions=self._enable_conversation_id + and not handle_free and ownership is not None and ownership.output_instructions, ), @@ -826,8 +881,15 @@ def _inject_model(self, tool: Any, ownership: Dict[str, bool]) -> Any: return tool def _inject_conversation( - self, tools: List[Any], ownership: Dict[str, _ConversationOwnership] + self, + tools: List[Any], + ownership: Dict[str, _ConversationOwnership], + advertise: bool = True, ) -> List[Any]: + """Add the conversation fields to ``tools`` and record which ones the SDK + owns. A listing that does not ``advertise`` them still records + ``conversation_id`` ownership, so a later call strips an echoed handle, but + declares no ``_mcp_instructions`` and so records none.""" if not self._enable_conversation_id: self._conversation_ownership = {} return tools @@ -838,6 +900,14 @@ def _inject_conversation( previous.conversation_id and current.conversation_id, previous.output_instructions and current.output_instructions, ) + if not advertise: + self._conversation_ownership.update( + { + name: _ConversationOwnership(owned.conversation_id, False) + for name, owned in ownership.items() + } + ) + return tools prepared = [] for tool in tools: name = _tool_name(tool) diff --git a/posthog/mcp/session.py b/posthog/mcp/session.py index 4da6a6273..1abfa942b 100644 --- a/posthog/mcp/session.py +++ b/posthog/mcp/session.py @@ -9,14 +9,27 @@ from __future__ import annotations from datetime import datetime, timezone -from typing import Optional +from typing import NamedTuple, Optional from .constants import INACTIVITY_TIMEOUT_IN_MINUTES +from ._client_conversation import ClientConversation from ._ids import deterministic_prefixed_id, new_prefixed_id from ._internal import MCPAnalyticsData from .session_token import SessionTokenPayload -__all__ = ["derive_session_id_from_mcp_session", "derive_session_id_from_conversation"] +__all__ = [ + "derive_session_id_from_mcp_session", + "derive_session_id_from_conversation", + "derive_session_id_from_client_conversation", +] + + +class ResolvedSession(NamedTuple): + """A request's session id and where it came from: ``"conversation"``, + ``"token"``, ``"mcp"``, ``"client_metadata"`` or ``"generated"``.""" + + session_id: str + source: str def new_session_id() -> str: @@ -44,27 +57,43 @@ def derive_session_id_from_conversation(conversation_id: str) -> str: return deterministic_prefixed_id("ses", conversation_id) +def derive_session_id_from_client_conversation(key: str, id: str) -> str: + """Derive the SDK session id from a client conversation id. + + Keyed so two clients' ids, or a client id and a conversation handle, never + share a hash input. Part of the cross-SDK contract, like + :func:`derive_session_id_from_conversation`: posthog-js produces the same value + byte for byte (``openai/session`` + ``v1/conv-example-0001`` is + ``ses_2f5d35b7108d17c807766f2f769a1e8c``).""" + return deterministic_prefixed_id("ses", f"{key}:{id}") + + async def resolve_session_id( data: MCPAnalyticsData, mcp_session_id: Optional[str], *, token: Optional[SessionTokenPayload] = None, conversation_id: Optional[str] = None, + client_conversation: Optional[ClientConversation] = 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 + data, + mcp_session_id, + token=token, + conversation_id=conversation_id, + client_conversation=client_conversation, ) return session_id -async def handshake_session_id( +async def handshake_session( data: MCPAnalyticsData, mcp_session_id: Optional[str], token: Optional[SessionTokenPayload], -) -> str: +) -> ResolvedSession: """The session an ``initialize`` belongs to: the one the client's later requests resolve to through :func:`resolve_session_id_with_source`. @@ -73,10 +102,12 @@ async def handshake_session_id( move the session that ``McpAnalytics.capture()`` reads. Without either, the session is this process's own, which only the resolver can hand out.""" if token is not None: - return token.session_id + return ResolvedSession(token.session_id, "token") if mcp_session_id: - return derive_session_id_from_mcp_session(mcp_session_id) - return await resolve_session_id(data, None) + return ResolvedSession( + derive_session_id_from_mcp_session(mcp_session_id), "mcp" + ) + return await resolve_session_id_with_source(data, None) async def resolve_session_id_with_source( @@ -85,25 +116,28 @@ async def resolve_session_id_with_source( *, token: Optional[SessionTokenPayload] = None, conversation_id: Optional[str] = None, -) -> tuple[str, str]: + client_conversation: Optional[ClientConversation] = None, +) -> ResolvedSession: """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 + ``"token"``, ``"mcp"``, ``"client_metadata"`` 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 - transport's MCP session id, then this instance's own memory. + Priority mirrors posthog-js ``resolveSession`` and the spec's "Session + identity": a recognized client's own conversation id first (it sends it on + every call), then the agent's ``conversation_id`` handle (the only id of ours + that survives the 2026-07-28 revision's per-request server instances), then our + self-encoded session token, then the transport's MCP session id, then this + instance's own memory. - ``conversation_id`` is resolved *per request and never stored*: the handle - belongs to one chat, and persisting it on shared ``data`` (or advancing - ``last_activity``) would leak one chat's session onto a concurrent chat's + ``conversation_id`` and ``client_conversation`` are resolved *per request and + never stored*: each belongs to one chat, and persisting it on shared ``data`` (or + advancing ``last_activity``) would leak one chat's session onto a concurrent chat's request. ``token`` is our self-encoded session token (see :mod:`.session_token`), @@ -116,8 +150,16 @@ async def resolve_session_id_with_source( one ``$session_id``. A compliant client replays the header on every request, so a genuine token session never needs the fallback. """ + if client_conversation is not None: + return ResolvedSession( + derive_session_id_from_client_conversation(*client_conversation), + "client_metadata", + ) + if conversation_id: - return derive_session_id_from_conversation(conversation_id), "conversation" + return ResolvedSession( + derive_session_id_from_conversation(conversation_id), "conversation" + ) async with data.session_lock: now = datetime.now(timezone.utc) @@ -129,29 +171,49 @@ async def resolve_session_id_with_source( data.session_id = token.session_id data.session_source = "token" data.last_activity = now - return data.session_id, "token" + return ResolvedSession(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, "mcp" + return ResolvedSession(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, "mcp" + return ResolvedSession(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 # isn't already a generated session starts fresh; generated sessions # persist and roll over on inactivity. - timeout_seconds = INACTIVITY_TIMEOUT_IN_MINUTES * 60 - is_stale = (now - data.last_activity).total_seconds() > timeout_seconds - if data.session_source != "generated" or is_stale: + if data.session_source != "generated" or _is_stale(data, now): data.session_id = new_session_id() data.session_source = "generated" data.last_activity = now - return data.session_id, "generated" + return ResolvedSession(data.session_id, "generated") + + +def _is_stale(data: MCPAnalyticsData, now: datetime) -> bool: + timeout_seconds = INACTIVITY_TIMEOUT_IN_MINUTES * 60 + return (now - data.last_activity).total_seconds() > timeout_seconds + + +async def touch_connection_session(data: MCPAnalyticsData) -> None: + """A one-client connection's id rotates on the first tool call that arrives 30 + minutes or more after the previous one, whichever source grouped that call. So + a call grouped by the client still rotates a stale id and resets the clock, or + a later call without one would rotate early. + + Only for a one-client connection (stdio): on a shared HTTP instance, keeping + the instance's own session alive would group unrelated clients' requests.""" + async with data.session_lock: + if data.session_source != "generated": + return + now = datetime.now(timezone.utc) + if _is_stale(data, now): + data.session_id = new_session_id() + data.last_activity = now diff --git a/posthog/mcp/types.py b/posthog/mcp/types.py index 04d1256c8..f00977260 100644 --- a/posthog/mcp/types.py +++ b/posthog/mcp/types.py @@ -65,9 +65,9 @@ # ``posthog_events``. Known keys: client_name, client_version, conversation_id, # duration, error, error_type, event_name, event_type, groups, id, identify_actor_data, # identify_actor_given_id, is_error, listed_tool_names, parameters, properties, -# resource_name, response, server_name, server_version, session_id, timestamp, -# tool_category, tool_description, user_intent, user_intent_source, llm_model, -# llm_model_source. +# resource_name, response, server_name, server_version, session_id, session_source, +# timestamp, tool_category, tool_description, user_intent, user_intent_source, +# llm_model, llm_model_source. Event = Dict[str, Any] McpEvent = Dict[str, Any] diff --git a/posthog/test/mcp/test_client_conversation_session.py b/posthog/test/mcp/test_client_conversation_session.py new file mode 100644 index 000000000..2bd2b56ef --- /dev/null +++ b/posthog/test/mcp/test_client_conversation_session.py @@ -0,0 +1,582 @@ +"""Client conversation ids, per the spec's "Session identity" requirement: a +recognized client's own id in a tool call's ``_meta`` sets the session, ahead of +any handle or transport session. Every call lands on fresh server state, as on a +per-request stateless server. Session ids below are the spec's vectors, which +posthog-js reproduces too. Runs under both MCP majors.""" + +import json +from datetime import timedelta +from types import SimpleNamespace + +import pytest + +from posthog.mcp import McpAnalytics, PostHogMCP +from posthog.mcp._client_conversation import ( + CLIENT_CONVERSATION_ROWS, + ClientConversation, + read_client_conversation, + recognize_client, +) +from posthog.mcp._instrumentation import prepare_request, start_tool_call_lifecycle +from posthog.mcp._internal import MCPAnalyticsData, set_server_tracking_data +from posthog.mcp._posthog_events import build_posthog_capture_events +from posthog.mcp._sink import McpEventSink +from posthog.mcp.session import ( + derive_session_id_from_client_conversation, + derive_session_id_from_mcp_session, + new_session_id, + resolve_session_id_with_source, +) +from posthog.mcp.session_token import SessionTokenPayload +from posthog.mcp.types import MCPAnalyticsOptions +from posthog.test.mcp._helpers import ( + FakeClient, + events_named as _events, + flush_background as _flush, +) +from posthog.test.mcp.test_initialize_handshake import BASE_HEADERS, KINDS, fleet, rpc + +CHATGPT = "openai-mcp/1.0.0 (ChatGPT)" +CODEX = "codex-mcp-client/0.50.0" +OTHER = "example-client/1.0" +CHATGPT_ID = "v1/conv-example-0001" +CHATGPT_META = {"openai/session": CHATGPT_ID} +CHATGPT_SESSION = "ses_2f5d35b7108d17c807766f2f769a1e8c" +CODEX_ID = "019A4F2E-1C3D-7B8A-9E0F-1A2B3C4D5E6F" +CODEX_SESSION = "ses_0fdf126b8e2c1f9689ca12f3dd3383e6" +VSCODE_ID = "3f1c2b7e-8d4a-4c6b-9a1e-5f2d7c8b9a0e" +VSCODE_SESSION = "ses_0571acb894e6753d33e37e5c2b327c59" +HANDLE = "0190f0e8-7a6b-7c3d-9e4f-5a6b7c8d9e0f" +HANDLE_SESSION = "ses_6df45f0102a182bcd5e8dd5dad6c65a0" +VSCODE_CLIENT_INFO = { + "io.modelcontextprotocol/clientInfo": {"name": "Visual Studio Code"}, + "io.modelcontextprotocol/protocolVersion": "2026-07-28", +} + + +class FakeServer: + """A weak-referenceable stand-in for the server key tracking state hangs off.""" + + +def http_extra(user_agent=None, **headers): + """An HTTP request's ``extra``, or a stdio one (no request) without a user agent.""" + if not user_agent and not headers: + return {} + if user_agent: + headers["user-agent"] = user_agent + return {"ctx": SimpleNamespace(request=SimpleNamespace(headers=headers))} + + +def fresh_data(client=None, enable_conversation_id=True): + return MCPAnalyticsData( + options=MCPAnalyticsOptions(enable_conversation_id=enable_conversation_id), + sink=McpEventSink(client or FakeClient()), + session_id=new_session_id(), + ) + + +def start_call( + data, + *, + user_agent=None, + meta=None, + arguments=None, + client_name=None, + mcp_session_id=None, + token=None, +): + return start_tool_call_lifecycle( + data, + name="search_docs", + arguments=arguments or {}, + request_meta=meta, + allow_self_reported_model=True, + mcp_session_id=mcp_session_id, + token=token, + client_name=client_name, + client_version=None, + protocol_version=None, + extra=http_extra(user_agent), + ) + + +async def call_on_fresh_pod(enable_conversation_id=True, **call_input): + client = FakeClient() + call = start_call(fresh_data(client, enable_conversation_id), **call_input) + await call.record_result( + {"content": []}, + 1.0, + conversation_id_delivered=bool(call.minted_conversation_id), + ) + await _flush() + (event,) = _events(client, "$mcp_tool_call") + return call, event["properties"] + + +@pytest.mark.parametrize( + ("user_agent", "client_name", "key"), + [ + ("openai-mcp/1.0.0 (ChatGPT)", None, "openai/session"), + ("openai-mcp/1.0.0 ( chatgpt; web)", None, "openai/session"), + ("openai-mcp/1.0.0 (Codex)", None, None), + ("openai-mcp/1.0.0 (ChatGPT-Agent)", None, None), + ("Mozilla/5.0 openai-mcp/1.0.0 (ChatGPT)", None, None), + ("CODEX-MCP-CLIENT/0.50.0", None, "sessionId"), + (None, "codex-mcp-client", "sessionId"), + ("Visual Studio Code/1.105.0", None, "vscode.conversationId"), + (None, "visual studio code - insiders", "vscode.conversationId"), + (CHATGPT, "codex-mcp-client", "openai/session"), + (OTHER, None, None), + ], +) +def test_recognize_client(user_agent, client_name, key): + row = recognize_client(user_agent, client_name) + assert (row.key if row else None) == key + + +@pytest.mark.parametrize( + ("row", "meta", "expected"), + [ + (0, {"openai/session": "x" * 128}, "x" * 128), + (0, {"openai/session": " ~"}, " ~"), + (0, {"openai/session": "x" * 129}, None), + (0, {"openai/session": ""}, None), + (0, {"openai/session": "chat-é"}, None), + (0, {"openai/session": "tab\there"}, None), + (0, {"openai/session": 42}, None), + (0, "openai/session", None), + (1, {"sessionId": CODEX_ID}, CODEX_ID.lower()), + (1, {"sessionId": "00000000-0000-0000-0000-000000000000"}, None), + (1, {"sessionId": "not-a-uuid"}, None), + (2, {"vscode.conversationId": "chat-1"}, None), + ], +) +def test_read_client_conversation(row, meta, expected): + conversation = read_client_conversation(CLIENT_CONVERSATION_ROWS[row], meta) + assert (conversation.id if conversation else None) == expected + + +@pytest.mark.parametrize( + ("key", "value", "session_id"), + [ + ("openai/session", CHATGPT_ID, CHATGPT_SESSION), + ("sessionId", CODEX_ID.lower(), CODEX_SESSION), + ("vscode.conversationId", VSCODE_ID, VSCODE_SESSION), + ], +) +def test_derivation_matches_the_spec_vectors(key, value, session_id): + assert derive_session_id_from_client_conversation(key, value) == session_id + + +@pytest.mark.parametrize("enable_conversation_id", [True, False]) +@pytest.mark.parametrize( + ("call_input", "session_id"), + [ + ({"user_agent": CHATGPT, "meta": CHATGPT_META}, CHATGPT_SESSION), + ({"user_agent": CODEX, "meta": {"sessionId": CODEX_ID}}, CODEX_SESSION), + ( + {"client_name": "codex-mcp-client", "meta": {"sessionId": CODEX_ID}}, + CODEX_SESSION, + ), + ( + {"meta": {**VSCODE_CLIENT_INFO, "vscode.conversationId": VSCODE_ID}}, + VSCODE_SESSION, + ), + ( + { + "user_agent": "Visual Studio Code - Insiders/1.106.0", + "meta": {"vscode.conversationId": VSCODE_ID}, + }, + VSCODE_SESSION, + ), + ], + ids=["chatgpt", "codex ua", "codex name", "vscode meta name", "insiders ua"], +) +async def test_a_client_conversation_id_sets_the_session_without_a_handle( + call_input, session_id, enable_conversation_id +): + call, props = await call_on_fresh_pod(enable_conversation_id, **call_input) + + assert not call.minted_conversation_id + assert call.virtual_result_texts("ok") == ["ok"] + assert props["$session_id"] == session_id + assert props["$mcp_session_source"] == "client_metadata" + assert "$mcp_conversation_id" not in props + + +@pytest.mark.parametrize( + "carried", + [ + {"arguments": {"conversation_id": HANDLE}}, + {"mcp_session_id": "transport"}, + {"token": SessionTokenPayload(session_id="ses_token")}, + ], + ids=["conversation handle", "transport session id", "session token"], +) +async def test_a_client_conversation_id_wins_over(carried): + call, props = await call_on_fresh_pod( + user_agent=CHATGPT, meta=CHATGPT_META, **carried + ) + + assert call.conversation_id is None + assert props["$session_id"] == CHATGPT_SESSION + assert props["$mcp_session_source"] == "client_metadata" + assert "$mcp_conversation_id" not in props + + +@pytest.mark.parametrize( + "meta", + [ + {"sessionId": "not-a-uuid"}, + {"sessionId": "00000000-0000-0000-0000-000000000000"}, + None, + ], + ids=["not a uuid", "nil uuid", "absent"], +) +async def test_an_invalid_codex_id_falls_back_to_a_handle(meta): + call, props = await call_on_fresh_pod(user_agent=CODEX, meta=meta) + + assert call.minted_conversation_id + assert props["$mcp_session_source"] == "conversation" + + +@pytest.mark.parametrize( + "meta", + [{"openai/session": ""}, {"openai/session": "x" * 129}, None], + ids=["empty", "too long", "absent"], +) +async def test_a_handle_free_client_is_minted_no_handle(meta): + first, first_props = await call_on_fresh_pod(user_agent=CHATGPT, meta=meta) + second, second_props = await call_on_fresh_pod(user_agent=CHATGPT, meta=meta) + + assert not first.minted_conversation_id and not second.minted_conversation_id + assert first.virtual_result_texts("ok") == ["ok"] + assert first_props["$session_id"] != second_props["$session_id"] + assert first_props["$mcp_session_source"] == "generated" + + +async def test_a_handle_free_client_still_counts_an_echoed_handle(): + call, props = await call_on_fresh_pod( + user_agent=CHATGPT, arguments={"conversation_id": HANDLE} + ) + + assert props["$session_id"] == HANDLE_SESSION + assert not call.mirrors_output_instructions + + +async def test_a_key_is_read_only_from_the_client_that_defines_it(): + call, props = await call_on_fresh_pod( + user_agent=OTHER, meta={"sessionId": CODEX_ID, **CHATGPT_META} + ) + + assert call.minted_conversation_id + assert props["$session_id"] not in (CODEX_SESSION, CHATGPT_SESSION) + + +async def test_a_client_conversation_never_touches_shared_state(): + data = MCPAnalyticsData(options=MCPAnalyticsOptions(), session_id="ses_memory") + before = (data.session_id, data.session_source, data.last_activity) + + resolved = await resolve_session_id_with_source( + data, + "transport", + client_conversation=ClientConversation("openai/session", CHATGPT_ID), + ) + + assert resolved == (CHATGPT_SESSION, "client_metadata") + assert (data.session_id, data.session_source, data.last_activity) == before + + +@pytest.mark.parametrize( + ("gaps", "rotated"), [([20, 20], False), ([40, 1], True)], ids=["keeps", "rotates"] +) +async def test_calls_grouped_by_a_client_id_count_as_connection_activity(gaps, rotated): + client = FakeClient() + data = fresh_data(client, enable_conversation_id=False) + + async def call(meta=None): + lifecycle = start_call(data, client_name="codex-mcp-client", meta=meta) + await lifecycle.prime_session() + await lifecycle.record_result( + {"content": []}, 1.0, conversation_id_delivered=False + ) + + await call() + data.last_activity -= timedelta(minutes=gaps[0]) + await call({"sessionId": CODEX_ID}) + data.last_activity -= timedelta(minutes=gaps[1]) + await call() + await _flush() + + first, grouped, last = (e["properties"] for e in _events(client, "$mcp_tool_call")) + assert grouped["$session_id"] == CODEX_SESSION + assert last["$mcp_session_source"] == "generated" + assert (last["$session_id"] == first["$session_id"]) is not rotated + + +async def test_chatgpt_calls_on_a_shared_http_pod_keep_no_session_alive(): + client = FakeClient() + data = fresh_data(client, enable_conversation_id=False) + + async def call(**call_input): + lifecycle = start_call(data, **call_input) + await lifecycle.prime_session() + await lifecycle.record_result( + {"content": []}, 1.0, conversation_id_delivered=False + ) + + await call(user_agent=OTHER) + for _ in range(3): + data.last_activity -= timedelta(minutes=20) + await call(user_agent=CHATGPT, meta=CHATGPT_META) + await call(user_agent="another-client/2.0") + await _flush() + + calls = [e["properties"] for e in _events(client, "$mcp_tool_call")] + assert calls[0]["$session_id"] != calls[-1]["$session_id"] + + +async def test_a_client_conversation_session_is_not_a_synthesized_handshake(): + client = FakeClient() + + await prepare_request( + fresh_data(client), + mcp_session_id=None, + client_name=None, + client_version=None, + request={"method": "tools/call", "params": {}}, + extra={}, + client_conversation=ClientConversation("openai/session", CHATGPT_ID), + ) + await _flush() + + assert _events(client, "$mcp_initialize") == [] + + +@pytest.mark.parametrize( + ("internal", "wire"), + [ + ("conversation", "conversation"), + ("token", "transport"), + ("mcp", "transport"), + ("client_metadata", "client_metadata"), + ("generated", "generated"), + (None, None), + ], +) +def test_session_source_property_names_the_grouping_source(internal, wire): + (capture,) = build_posthog_capture_events( + { + "event_type": "mcp:tools/call", + "session_id": "ses_x", + "session_source": internal, + } + ) + + assert capture["properties"].get("$mcp_session_source") == wire + + +@pytest.mark.parametrize( + ("session_source", "wire"), + [("generated", "generated"), ("mcp", "transport"), ("token", "transport")], +) +async def test_custom_events_carry_the_instance_session_source(session_source, wire): + client = FakeClient() + key = FakeServer() + set_server_tracking_data( + key, + MCPAnalyticsData( + options=MCPAnalyticsOptions(), + sink=McpEventSink(client), + session_id="ses_instance", + session_source=session_source, + ), + ) + + await McpAnalytics(key).capture("custom_thing") + + (event,) = _events(client, "custom_thing") + assert event["properties"]["$session_id"] == "ses_instance" + assert event["properties"]["$mcp_session_source"] == wire + + +# --- PostHogMCP -------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("kwargs", "session_id", "minted"), + [ + ( + { + "headers": {"User-Agent": CHATGPT}, + "request_meta": CHATGPT_META, + "session_id": "ses_x", + }, + CHATGPT_SESSION, + False, + ), + ( + { + "client_info": {"name": "codex-mcp-client"}, + "request_meta": {"sessionId": CODEX_ID}, + }, + CODEX_SESSION, + False, + ), + ({"headers": {"user-agent": CHATGPT}}, None, False), + ({"request_meta": CHATGPT_META}, None, True), + ( + {"headers": {"user-agent": CHATGPT}, "session_id": "ses_carried"}, + "ses_carried", + False, + ), + ], + ids=[ + "chatgpt id over a carried session", + "codex id from the bound handshake", + "handle-free client without its id", + "chatgpt key without its user-agent", + "carried session", + ], +) +def test_prepare_tool_call(kwargs, session_id, minted): + client = PostHogMCP("phc_test", host="https://us.i.posthog.com") + + call = client.prepare_tool_call("search_docs", {}, **kwargs) + prepared = client.prepare_tool_result({"content": []}, call) + + assert ("conversation_id" in json.dumps(prepared.result)) is minted + if not minted: + assert prepared.session_id == session_id + assert prepared.conversation_id is None + + +@pytest.mark.parametrize( + ("user_agent", "session_id"), + [(CHATGPT, CHATGPT_SESSION), (None, "ses_caller")], + ids=["recognized", "unrecognized"], +) +async def test_capture_tool_call_applies_a_client_conversation_id( + user_agent, session_id +): + fake = FakeClient() + client = PostHogMCP("phc_test", host="https://us.i.posthog.com") + client.capture = fake.capture + + client.capture_tool_call( + "search_docs", + session_id="ses_caller", + conversation_id=HANDLE, + client_user_agent=user_agent, + request_meta=CHATGPT_META, + ) + await _flush() + + (event,) = _events(fake, "$mcp_tool_call") + assert event["properties"]["$session_id"] == session_id + assert ("$mcp_conversation_id" in event["properties"]) is (user_agent is None) + + +@pytest.mark.parametrize( + ("user_agent", "advertised"), [(CHATGPT, False), (OTHER, True), (None, True)] +) +def test_prepare_tool_list_advertises_conversation_id(user_agent, advertised): + client = PostHogMCP("phc_test", host="https://us.i.posthog.com") + tool = {"name": "search_docs", "inputSchema": {"type": "object", "properties": {}}} + + (prepared,) = client.prepare_tool_list( + [tool], headers={"user-agent": user_agent} if user_agent else None + ) + + assert ("conversation_id" in prepared["inputSchema"]["properties"]) is advertised + + +def test_a_chatgpt_listing_still_lets_prepare_tool_call_strip_the_handle(): + client = PostHogMCP("phc_test", host="https://us.i.posthog.com") + tool = {"name": "search_docs", "inputSchema": {"type": "object", "properties": {}}} + client.prepare_tool_list([tool], headers={"user-agent": CHATGPT}) + + call = client.prepare_tool_call( + "search_docs", + {"query": "flags", "conversation_id": HANDLE}, + headers={"user-agent": CHATGPT}, + request_meta=CHATGPT_META, + ) + + assert call.args == {"query": "flags"} + assert call.session_id == CHATGPT_SESSION + + +# --- over HTTP, on every server kind ------------------------------------------- + + +def _headers(user_agent): + return {**BASE_HEADERS, "user-agent": user_agent} + + +@pytest.mark.parametrize("kind", KINDS) +@pytest.mark.parametrize( + ("user_agent", "advertised"), [(CHATGPT, False), (OTHER, True)] +) +async def test_only_a_handle_free_clients_listing_has_no_conversation_id( + kind, user_agent, advertised +): + client = FakeClient() + options = MCPAnalyticsOptions(enable_conversation_id=True) + async with fleet(client, pods=1, kind=kind, options=options) as lb: + response = await lb.post(rpc("tools/list", {}), _headers(user_agent)) + assert response.status_code == 200, response.text + + (tool,) = [t for t in response.json()["result"]["tools"] if t["name"] == "add"] + assert ("context" in tool["inputSchema"]["properties"]) is True + assert ("conversation_id" in tool["inputSchema"]["properties"]) is advertised + + +@pytest.mark.parametrize("kind", KINDS) +async def test_chatgpt_calls_share_a_session_across_pods(kind): + client = FakeClient() + options = MCPAnalyticsOptions(enable_conversation_id=True) + async with fleet(client, pods=2, kind=kind, options=options) as lb: + responses = [ + await lb.post( + rpc( + "tools/call", + { + "name": "add", + "arguments": {"a": 1, "b": 2, "conversation_id": HANDLE}, + "_meta": CHATGPT_META, + }, + request_id, + ), + _headers(CHATGPT), + ) + for request_id in (1, 2) + ] + await _flush() + + for response in responses: + assert response.status_code == 200, response.text + assert not response.json()["result"].get("isError"), response.text + assert "conversation_id" not in json.dumps(response.json()["result"]["content"]) + calls = _events(client, "$mcp_tool_call") + assert [c["properties"]["$session_id"] for c in calls] == [CHATGPT_SESSION] * 2 + assert all("$mcp_conversation_id" not in c["properties"] for c in calls) + assert all( + "conversation_id" not in json.dumps(c["properties"].get("$mcp_parameters")) + for c in calls + ) + + +@pytest.mark.parametrize("kind", KINDS) +async def test_a_chatgpt_listing_reads_no_client_conversation(kind): + """The spec reads client conversation ids from tool calls only.""" + client = FakeClient() + async with fleet(client, pods=1, kind=kind) as lb: + await lb.post(rpc("tools/list", {"_meta": CHATGPT_META}), _headers(CHATGPT)) + await _flush() + + (listing,) = _events(client, "$mcp_tools_list") + assert listing["properties"]["$session_id"] != CHATGPT_SESSION + assert listing["properties"]["$session_id"] != derive_session_id_from_mcp_session( + CHATGPT_ID + ) diff --git a/posthog/test/mcp/test_defaults.py b/posthog/test/mcp/test_defaults.py index 3f974980a..7c123bf8f 100644 --- a/posthog/test/mcp/test_defaults.py +++ b/posthog/test/mcp/test_defaults.py @@ -126,4 +126,5 @@ async def test_carried_session_is_preserved_until_agent_echoes(source, echoed): assert not call.minted_conversation_id assert call.conversation_id == (handle if echoed else None) assert call.virtual_result_texts("ok") == ["ok"] - assert await call.prepare_session(call.conversation_id) == expected + session = await call.prepare_session(call.conversation_id) + assert session.session_id == expected diff --git a/posthog/test/mcp/test_fastmcp.py b/posthog/test/mcp/test_fastmcp.py index 53cdfea34..5e2ed8e69 100644 --- a/posthog/test/mcp/test_fastmcp.py +++ b/posthog/test/mcp/test_fastmcp.py @@ -172,6 +172,44 @@ async def test_tool_call_captures_client_model_without_prior_listing(): assert "llm_model" not in props["$mcp_parameters"]["request"]["params"]["arguments"] +@pytest.mark.parametrize("enable_conversation_id", [True, False]) +async def test_client_conversation_groups_calls_on_fresh_servers( + enable_conversation_id, +): + client = FakeClient() + context = SimpleNamespace( + request_context=SimpleNamespace( + meta={ + "sessionId": "019a4f2e-1c3d-7b8a-9e0f-1a2b3c4d5e6f", + "io.modelcontextprotocol/clientInfo": {"name": "codex-mcp-client"}, + } + ) + ) + results = [] + for _ in range(2): + server = make_server() + instrument( + server, + client, + MCPAnalyticsOptions(enable_conversation_id=enable_conversation_id), + ) + results.append( + await server._tool_manager.call_tool( + "add", {"a": 2, "b": 3}, context=context + ) + ) + await _flush() + + calls = _events(client, "$mcp_tool_call") + assert results == [5, 5] + assert [c["properties"]["$session_id"] for c in calls] == [ + "ses_0fdf126b8e2c1f9689ca12f3dd3383e6" + ] * 2 + assert {c["properties"]["$mcp_session_source"] for c in calls} == { + "client_metadata" + } + + async def test_analytics_flush_drains_its_own_captures(): async def slow_before_send(event): await asyncio.sleep(0.05) diff --git a/posthog/test/mcp/test_session_token.py b/posthog/test/mcp/test_session_token.py index 9db5aca7e..6e6f9995f 100644 --- a/posthog/test/mcp/test_session_token.py +++ b/posthog/test/mcp/test_session_token.py @@ -787,7 +787,7 @@ async def test_no_warning_when_session_is_anchored_by_conversation_id(): is perfectly healthy.""" with _captured_logs() as logs: data = _data() - session_id = await prepare_request( + session = await prepare_request( data, mcp_session_id=None, client_name=None, @@ -797,7 +797,8 @@ async def test_no_warning_when_session_is_anchored_by_conversation_id(): conversation_id="0199e0a1-0000-7000-8000-000000000000", ) - assert session_id.startswith("ses_") + assert session.session_id.startswith("ses_") + assert session.source == "conversation" assert not [m for m in logs if _NO_SESSION in m] diff --git a/posthog/test/mcp/test_v2_wire_dual_era.py b/posthog/test/mcp/test_v2_wire_dual_era.py index 58fbf4e9a..f09154d9d 100644 --- a/posthog/test/mcp/test_v2_wire_dual_era.py +++ b/posthog/test/mcp/test_v2_wire_dual_era.py @@ -20,6 +20,8 @@ import httpx import pytest +import mcp.types as mcp_types +from mcp.server.lowlevel import Server from mcp.server.mcpserver import MCPServer from posthog.mcp import ( @@ -72,15 +74,22 @@ def rpc(method, params, request_id=1): return {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params} -async def modern_call(http, name, arguments, request_id=1, model=None): +async def modern_call( + http, name, arguments, request_id=1, model=None, meta=None, user_agent=None +): body = rpc( "tools/call", - {"name": name, "arguments": arguments, "_meta": modern_meta(model=model)}, + { + "name": name, + "arguments": arguments, + "_meta": {**modern_meta(model=model), **(meta or {})}, + }, request_id, ) - return await http.post( - "/mcp", json=body, headers=modern_headers("tools/call", name) - ) + headers = modern_headers("tools/call", name) + if user_agent: + headers = {**headers, "user-agent": user_agent} + return await http.post("/mcp", json=body, headers=headers) # --- modern era (2026-07-28) --------------------------------------------------- @@ -220,6 +229,41 @@ async def test_modern_conversation_anchors_session_across_instances(): assert [c["properties"]["$mcp_conversation_id"] for c in calls] == [minted, minted] +@pytest.mark.parametrize("enable_conversation_id", [True, False]) +async def test_modern_client_conversation_groups_calls_across_instances( + enable_conversation_id, +): + """ChatGPT's own conversation id groups fresh per-request instances with no + handle and no result change: ``openai/session`` survives the SDK lifting the + reserved envelope keys out of ``_meta``.""" + client = FakeClient() + options = MCPAnalyticsOptions(enable_conversation_id=enable_conversation_id) + responses = [] + for _ in range(2): + server = make_server() + instrument(server, client, options) + async with wire(server) as http: + responses.append( + await modern_call( + http, + "add", + {"a": 1, "b": 1}, + meta={"openai/session": "v1/conv-example-0001"}, + user_agent="openai-mcp/1.0.0 (ChatGPT)", + ) + ) + await _flush() + + calls = _events(client, "$mcp_tool_call") + assert [c["properties"]["$session_id"] for c in calls] == [ + "ses_2f5d35b7108d17c807766f2f769a1e8c" + ] * 2 + assert {c["properties"]["$mcp_session_source"] for c in calls} == { + "client_metadata" + } + assert all("conversation_id" not in r.text for r in responses) + + async def test_modern_result_shape_survives_instrumentation(): """Alive check: the instrumented result matches the bare server's, minus analytics-owned additions.""" @@ -444,3 +488,66 @@ async def test_real_request_headers_reach_the_event(): props = _events(client, "$mcp_tool_call")[0]["properties"] assert props[P.CLIENT_USER_AGENT] == user_agent assert props[P.VENDOR_CLIENT] == "cli" + + +# --- internal tools/list (Mcp-Param-* header validation) ----------------------- + + +def make_lowlevel_server(): + async def on_list_tools(ctx, params): + return mcp_types.ListToolsResult( + tools=[ + mcp_types.Tool( + name="add", + input_schema={ + "type": "object", + "properties": {"a": {"type": "integer"}}, + }, + ) + ] + ) + + async def on_call_tool(ctx, params): + return mcp_types.CallToolResult( + content=[mcp_types.TextContent(type="text", text="ok")] + ) + + return Server("low-wire", on_call_tool=on_call_tool, on_list_tools=on_list_tools) + + +@pytest.mark.parametrize( + "make", [make_server, make_lowlevel_server], ids=["mcpserver", "lowlevel"] +) +async def test_modern_tool_call_emits_no_tools_list_from_header_validation(make): + """The SDK's own tools/call path may run the tools/list handler to find the + tool's input schema. That is not a request the client sent, so it is no event.""" + server = make() + client = FakeClient() + instrument(server, client, MCPAnalyticsOptions()) + + async with wire(server) as http: + response = await modern_call(http, "add", {"a": 1, "b": 2}) + await _flush() + + assert response.status_code == 200 + assert [event["event"] for event in client.events] == ["$mcp_tool_call"] + + +@pytest.mark.parametrize( + "make", [make_server, make_lowlevel_server], ids=["mcpserver", "lowlevel"] +) +async def test_modern_client_tools_list_still_emits(make): + server = make() + client = FakeClient() + instrument(server, client, MCPAnalyticsOptions()) + + async with wire(server) as http: + response = await http.post( + "/mcp", + json=rpc("tools/list", {"_meta": modern_meta()}), + headers=modern_headers("tools/list"), + ) + await _flush() + + assert response.status_code == 200 + assert len(_events(client, "$mcp_tools_list")) == 1 diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index bfffd5d64..bea45894b 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -374,7 +374,7 @@ alias posthog.mcp.asgi.MCP_SESSION_HEADER -> posthog.mcp.session_token.MCP_SESSI alias posthog.mcp.asgi.SessionTokenPayload -> posthog.mcp.session_token.SessionTokenPayload alias posthog.mcp.asgi.decode_session_id -> posthog.mcp.session_token.decode_session_id alias posthog.mcp.asgi.encode_session_id -> posthog.mcp.session_token.encode_session_id -alias posthog.mcp.asgi.handshake_session_id -> posthog.mcp.session.handshake_session_id +alias posthog.mcp.asgi.handshake_session -> posthog.mcp.session.handshake_session 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 @@ -1055,6 +1055,7 @@ attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.SERVER_BUILD = '$mcp attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.SERVER_NAME = '$mcp_server_name' attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.SERVER_VERSION = '$mcp_server_version' attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.SESSION_ID = '$session_id' +attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.SESSION_SOURCE = '$mcp_session_source' attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.SOURCE = '$mcp_source' attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.TOOL_CATEGORY = '$mcp_tool_category' attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.TOOL_DESCRIPTION = '$mcp_tool_description' @@ -1582,6 +1583,7 @@ function posthog.mcp.get_tool_input_properties(input_value: Any, input_schema: A function posthog.mcp.instrument(server: Any, posthog_client: Optional[Client] = None, options: Optional[MCPAnalyticsOptions] = None) -> McpAnalytics function posthog.mcp.logger.set_logger(logger: Optional[LoggerFn]) -> None function posthog.mcp.request_headers.get_request_headers(extra: Any) -> Optional[RequestHeaderBag] +function posthog.mcp.session.derive_session_id_from_client_conversation(key: str, id: str) -> str function posthog.mcp.session.derive_session_id_from_conversation(conversation_id: str) -> str function posthog.mcp.session.derive_session_id_from_mcp_session(mcp_session_id: str) -> str function posthog.mcp.session_token.decode_session_id(value: Any) -> Optional[SessionTokenPayload] @@ -1891,11 +1893,11 @@ method posthog.mcp.posthog_mcp.PostHogMCP.capture(event: str, **kwargs: Unpack[O method posthog.mcp.posthog_mcp.PostHogMCP.capture_feedback(*, report: FeedbackReport, llm_model: Optional[str] = None, llm_model_source: Optional[MCPAnalyticsModelSource] = None, protocol_version: Optional[str] = None, distinct_id: Optional[str] = None, session_id: Optional[str] = None, conversation_id: Optional[str] = None, client_user_agent: Optional[str] = None, vendor_client: Optional[str] = None, set_properties: Optional[JsonRecord] = None, groups: Optional[Dict[str, str]] = None, properties: Optional[JsonRecord] = None, timestamp: Optional[datetime] = None) -> None method posthog.mcp.posthog_mcp.PostHogMCP.capture_initialize(*, client_name: Optional[str] = None, client_version: Optional[str] = None, protocol_version: Optional[str] = None, parameters: Any = None, response: Any = None, duration_ms: Optional[float] = None, distinct_id: Optional[str] = None, session_id: Optional[str] = None, client_user_agent: Optional[str] = None, vendor_client: Optional[str] = None, set_properties: Optional[JsonRecord] = None, groups: Optional[Dict[str, str]] = None, properties: Optional[JsonRecord] = None, timestamp: Optional[datetime] = None) -> None method posthog.mcp.posthog_mcp.PostHogMCP.capture_missing_capability(*, context: Optional[str] = None, llm_model: Optional[str] = None, llm_model_source: Optional[MCPAnalyticsModelSource] = None, parameters: Any = None, protocol_version: Optional[str] = None, distinct_id: Optional[str] = None, session_id: Optional[str] = None, conversation_id: Optional[str] = None, client_user_agent: Optional[str] = None, vendor_client: Optional[str] = None, set_properties: Optional[JsonRecord] = None, groups: Optional[Dict[str, str]] = None, properties: Optional[JsonRecord] = None, timestamp: Optional[datetime] = None) -> None -method posthog.mcp.posthog_mcp.PostHogMCP.capture_tool_call(tool_name: str, *, intent: Optional[str] = None, intent_source: Optional[str] = None, parameters: Any = None, response: Any = None, duration_ms: Optional[float] = None, is_error: bool = False, error: Any = None, error_type: Optional[str] = None, category: Optional[str] = None, tool_description: Optional[str] = None, llm_model: Optional[str] = None, llm_model_source: Optional[MCPAnalyticsModelSource] = None, protocol_version: Optional[str] = None, distinct_id: Optional[str] = None, session_id: Optional[str] = None, conversation_id: Optional[str] = None, client_user_agent: Optional[str] = None, vendor_client: Optional[str] = None, set_properties: Optional[JsonRecord] = None, groups: Optional[Dict[str, str]] = None, properties: Optional[JsonRecord] = None, timestamp: Optional[datetime] = None) -> None +method posthog.mcp.posthog_mcp.PostHogMCP.capture_tool_call(tool_name: str, *, intent: Optional[str] = None, intent_source: Optional[str] = None, parameters: Any = None, response: Any = None, duration_ms: Optional[float] = None, is_error: bool = False, error: Any = None, error_type: Optional[str] = None, category: Optional[str] = None, tool_description: Optional[str] = None, llm_model: Optional[str] = None, llm_model_source: Optional[MCPAnalyticsModelSource] = None, protocol_version: Optional[str] = None, distinct_id: Optional[str] = None, session_id: Optional[str] = None, conversation_id: Optional[str] = None, client_user_agent: Optional[str] = None, vendor_client: Optional[str] = None, set_properties: Optional[JsonRecord] = None, groups: Optional[Dict[str, str]] = None, properties: Optional[JsonRecord] = None, timestamp: Optional[datetime] = None, request_meta: Optional[JsonRecord] = None) -> None method posthog.mcp.posthog_mcp.PostHogMCP.capture_tools_list(*, tool_names: Optional[List[str]] = None, parameters: Any = None, response: Any = None, duration_ms: Optional[float] = None, is_error: bool = False, error: Any = None, error_type: Optional[str] = None, protocol_version: Optional[str] = None, distinct_id: Optional[str] = None, session_id: Optional[str] = None, client_user_agent: Optional[str] = None, vendor_client: Optional[str] = None, set_properties: Optional[JsonRecord] = None, groups: Optional[Dict[str, str]] = None, properties: Optional[JsonRecord] = None, timestamp: Optional[datetime] = None) -> None method posthog.mcp.posthog_mcp.PostHogMCP.flush(timeout_seconds: Optional[float] = 10) -> None -method posthog.mcp.posthog_mcp.PostHogMCP.prepare_tool_call(name: str, args: Optional[JsonRecord] = None, *, request_meta: Optional[JsonRecord] = None, original_tool: Any = None, session_id: Optional[str] = None) -> PreparedToolCall -method posthog.mcp.posthog_mcp.PostHogMCP.prepare_tool_list(tools: List[Any], context: Union[bool, MCPAnalyticsContextOptions] = True, report_missing: bool = False, collect_feedback: bool = False) -> List[Any] +method posthog.mcp.posthog_mcp.PostHogMCP.prepare_tool_call(name: str, args: Optional[JsonRecord] = None, *, request_meta: Optional[JsonRecord] = None, original_tool: Any = None, session_id: Optional[str] = None, headers: Any = None, client_info: Optional[Mapping[str, Any]] = None) -> PreparedToolCall +method posthog.mcp.posthog_mcp.PostHogMCP.prepare_tool_list(tools: List[Any], context: Union[bool, MCPAnalyticsContextOptions] = True, report_missing: bool = False, collect_feedback: bool = False, headers: Any = None) -> List[Any] method posthog.mcp.posthog_mcp.PostHogMCP.prepare_tool_result(result: TResult, prepared_call: PreparedToolCall) -> PreparedToolResult[TResult] method posthog.mcp.posthog_mcp.PostHogMCP.shutdown() -> None method posthog.metrics_capture.PostHogMetrics.count(name: str, value: float = 1, unit: Optional[str] = None, attributes: Optional[dict] = None) -> None