feat(mcp): add a capture_initialize option

.sampo/changesets/mcp-capture-initialize-option.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Add a `capture_initialize` option to MCP analytics. It is on by default. Set it to `False` to stop sending `$mcp_initialize`.

posthog/mcp/README.md

@@ -41,8 +41,8 @@ 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.
 
 
-Intent, model capture, conversation correlation, and MCP exception capture are on by default.
-Missing-capability reporting and feedback collection remain off.
+Intent, model capture, conversation correlation, MCP exception capture, and `$mcp_initialize`
+are on by default. Missing-capability reporting and feedback collection remain off.
 
 ```python
 from posthog.mcp import MCPAnalyticsOptions, instrument
@@ -88,6 +88,14 @@ The MCP SDK answers `initialize` itself, so the SDK observes it from outside:
 - **MCP SDK 1.x over stdio**: from the session's first request. A stdio server
   serves one client, so this misses only a client that never sends a request.
 
+Set `capture_initialize=False` to stop sending `$mcp_initialize`. Many clients reconnect before
+every few tool calls, so this event can outnumber `$mcp_tool_call`. Nothing else changes: tool
+calls keep their sessions and client names.
+
+```python
+instrument(server, posthog, MCPAnalyticsOptions(capture_initialize=False))
+```
+
 `identify` and `event_properties` run for the handshake too. On MCP SDK 1.x the
 SDK has no request context for it, so `extra["ctx"]` carries only `.request`.
 `get_request_headers(extra)` works the same as for any other request.

posthog/mcp/_instrumentation.py

@@ -260,14 +260,19 @@ async def record_initialize(
     stateless fleet may never see this client again. The capture itself runs after,
     in the background, as for every other event."""
 
+    request = {"method": "initialize", "params": params}
+    await handle_identify(data, 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.capture_initialize:
+        return
+
     def text(value: Any) -> Optional[str]:
         return value if isinstance(value, str) and value else None
 
     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
-    request = {"method": "initialize", "params": params}
-    await handle_identify(data, session_id, request, extra)
     data.mark_session_initialized(session_id)
     event: Dict[str, Any] = {
         "event_type": MCPAnalyticsEventType.MCP_INITIALIZE,
@@ -321,6 +326,8 @@ async def _maybe_emit_initialize(
     """Emit ``$mcp_initialize`` once per session from its first request, where no
     seam observes the real ``initialize`` (see
     :func:`_first_request_stands_in_for_initialize`)."""
+    if not data.options.capture_initialize:
+        return
     if not _first_request_stands_in_for_initialize(data, session_source, extra):
         return
     if session_id in data.initialized_sessions:

posthog/mcp/types.py

@@ -246,6 +246,9 @@ class MCPAnalyticsOptions:
     # its input schema to remove only PostHog-owned arguments on a fresh server
     # instance that did not serve tools/list.
     resolve_original_tool: Optional[Callable[[str], Any]] = None
+    # Send `$mcp_initialize` for each `initialize` handshake. On by default.
+    # False sends no `$mcp_initialize`. Nothing else changes.
+    capture_initialize: bool = True
 
 
 @dataclass

posthog/test/mcp/test_initialize_handshake.py

@@ -988,3 +988,77 @@ async def test_a_nested_method_before_the_top_level_initialize_is_counted(kind):
     assert response.status_code == 200, response.text
     (init,) = _events(client, "$mcp_initialize")
     assert init["properties"]["$mcp_client_name"] == "nested-client"
+
+
+# --- capture_initialize=False --------------------------------------------------------
+
+
+@pytest.mark.parametrize("kind", KINDS)
+async def test_capture_initialize_off_sends_no_initialize_and_keeps_sessions(kind):
+    # The multi-pod scenario with the option off: no $mcp_initialize. Nothing else
+    # changes: the token is still minted, so each connection's tool calls keep its
+    # session and its client name on every pod.
+    client = FakeClient()
+    options = MCPAnalyticsOptions(
+        enable_conversation_id=False, capture_initialize=False
+    )
+    sessions = []
+    async with fleet(client, pods=2, kind=kind, options=options) as lb:
+        for _ in range(2):
+            headers, _ = await handshake(lb)
+            token = decode_session_id(headers.get("mcp-session-id"))
+            assert token is not None
+            sessions.append(token.session_id)
+            await list_tools(lb, headers)
+            await call_add(lb, headers, 3)
+            await call_add(lb, headers, 4)
+    await _flush()
+
+    assert _events(client, "$mcp_initialize") == []
+    calls = _events(client, "$mcp_tool_call")
+    assert [c["properties"]["$session_id"] for c in calls] == [
+        sessions[0],
+        sessions[0],
+        sessions[1],
+        sessions[1],
+    ]
+    assert {c["properties"]["$mcp_client_name"] for c in calls} == {"codex-mcp-client"}
+
+
+@pytest.mark.parametrize("kind", KINDS)
+async def test_capture_initialize_off_keeps_the_handshake_identity(kind):
+    # The option stops only the event. An identify that resolves the user only
+    # from the handshake still attributes the session's later tool calls.
+    def identify(request, extra):
+        if request.get("method") == "initialize":
+            return UserIdentity(distinct_id="user_from_handshake")
+        return None
+
+    client = FakeClient()
+    options = MCPAnalyticsOptions(
+        enable_conversation_id=False, identify=identify, capture_initialize=False
+    )
+    async with fleet(client, kind=kind, options=options) as lb:
+        headers, _ = await handshake(lb)
+        await call_add(lb, headers)
+    await _flush()
+
+    assert _events(client, "$mcp_initialize") == []
+    (call,) = _events(client, "$mcp_tool_call")
+    assert call["distinct_id"] == "user_from_handshake"
+
+
+@pytest.mark.skipif(MCP_MAJOR >= 2, reason="the first-request fallback is mcp 1.x")
+@pytest.mark.parametrize("capture", [True, False])
+async def test_capture_initialize_off_also_gates_the_first_request_fallback(capture):
+    # No HTTP, as on stdio: the first tool call stands in for the handshake.
+    client = FakeClient()
+    options = MCPAnalyticsOptions(
+        enable_conversation_id=False, capture_initialize=capture
+    )
+    _, _, server = make_pod(client, options=options)
+    await server._tool_manager.call_tool("add", {"a": 1, "b": 2, "context": "stdio"})
+    await _flush()
+
+    assert len(_events(client, "$mcp_tool_call")) == 1
+    assert len(_events(client, "$mcp_initialize")) == (1 if capture else 0)

references/public_api_snapshot.txt

@@ -1039,6 +1039,7 @@ attribute posthog.mcp.types.MCPAnalyticsContextOptions.description: Optional[str
 attribute posthog.mcp.types.MCPAnalyticsModelOptions.description: Optional[str] = None
 attribute posthog.mcp.types.MCPAnalyticsModelSource = Literal['client_metadata', 'self_reported']
 attribute posthog.mcp.types.MCPAnalyticsOptions.before_send: Optional[BeforeSendFn] = None
+attribute posthog.mcp.types.MCPAnalyticsOptions.capture_initialize: bool = True
 attribute posthog.mcp.types.MCPAnalyticsOptions.capture_model: Union[bool, MCPAnalyticsModelOptions] = True
 attribute posthog.mcp.types.MCPAnalyticsOptions.collect_feedback: Union[bool, CollectFeedbackOptions] = False
 attribute posthog.mcp.types.MCPAnalyticsOptions.context: Union[bool, MCPAnalyticsContextOptions] = True
@@ -1306,7 +1307,7 @@ class posthog.mcp.types.CollectFeedbackOptions(tool_name: Optional[str] = None,
 class posthog.mcp.types.FeedbackReport(feedback_type: str = 'other', summary: str = '', sentiment: Optional[str] = None, friction_points: Optional[str] = None, suggested_improvement: Optional[str] = None, details: Optional[str] = None, tool_name: Optional[str] = None, task_completed: Optional[bool] = None, extras: JsonRecord = dict(), raw: JsonRecord = dict())
 class posthog.mcp.types.MCPAnalyticsContextOptions(description: Optional[str] = None)
 class posthog.mcp.types.MCPAnalyticsModelOptions(description: Optional[str] = None)
-class posthog.mcp.types.MCPAnalyticsOptions(logger: Optional[LoggerFn] = None, report_missing: bool = False, missing_capability_tool_name: Optional[str] = None, enable_conversation_id: bool = True, enable_exception_autocapture: bool = True, context: Union[bool, MCPAnalyticsContextOptions] = True, identify: Optional[Union[IdentifyFn, UserIdentity]] = None, intent_fallback: Optional[IntentFallbackFn] = None, before_send: Optional[BeforeSendFn] = None, event_properties: Optional[EventPropertiesFn] = None, capture_model: Union[bool, MCPAnalyticsModelOptions] = True, collect_feedback: Union[bool, CollectFeedbackOptions] = False, server_build: Optional[str] = None, should_record_input_key: Optional[ShouldRecordInputKeyFn] = None, resolve_input_aliases: Optional[ResolveInputAliasesFn] = None, resolve_original_tool: Optional[Callable[[str], Any]] = None)
+class posthog.mcp.types.MCPAnalyticsOptions(logger: Optional[LoggerFn] = None, report_missing: bool = False, missing_capability_tool_name: Optional[str] = None, enable_conversation_id: bool = True, enable_exception_autocapture: bool = True, context: Union[bool, MCPAnalyticsContextOptions] = True, identify: Optional[Union[IdentifyFn, UserIdentity]] = None, intent_fallback: Optional[IntentFallbackFn] = None, before_send: Optional[BeforeSendFn] = None, event_properties: Optional[EventPropertiesFn] = None, capture_model: Union[bool, MCPAnalyticsModelOptions] = True, collect_feedback: Union[bool, CollectFeedbackOptions] = False, server_build: Optional[str] = None, should_record_input_key: Optional[ShouldRecordInputKeyFn] = None, resolve_input_aliases: Optional[ResolveInputAliasesFn] = None, resolve_original_tool: Optional[Callable[[str], Any]] = None, capture_initialize: bool = True)
 class posthog.mcp.types.PreparedToolCall(args: Optional[JsonRecord] = None, intent: Optional[str] = None, intent_source: Optional[str] = None, is_missing_capability: bool = False, llm_model: Optional[str] = None, llm_model_source: Optional[MCPAnalyticsModelSource] = None, is_feedback: bool = False, feedback_report: Optional[FeedbackReport] = None, session_id: Optional[str] = None, conversation_id: Optional[str] = None, _conversation_state: Optional[PreparedConversationState] = None)
 class posthog.mcp.types.ToolInputOptions(should_record_input_key: Optional[ShouldRecordInputKeyFn] = None, input_aliases: Optional[InputAliasMap] = None)
 class posthog.mcp.types.UserIdentity(distinct_id: str, properties: Optional[JsonRecord] = None, groups: Optional[Dict[str, str]] = None)