Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .sampo/changesets/quiet-mcp-identify.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
pypi/posthog: minor
---

MCP analytics no longer sends standalone `$identify` events, so insights, funnels, or cohorts built on MCP `$identify` stop receiving new data. The `identify` callback still sets `distinct_id`, `$set` and `$groups` on every event.
2 changes: 1 addition & 1 deletion examples/mcp_analytics_demo.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ async def main() -> None:
list_handler = server._mcp_server.request_handlers[mcp_types.ListToolsRequest]
await list_handler(mcp_types.ListToolsRequest(method="tools/list"))

# tool calls -> $mcp_initialize (lazy, once), $identify, $mcp_tool_call x3, $exception
# tool calls -> $mcp_initialize (lazy, once), $mcp_tool_call x3, $exception
await server._tool_manager.call_tool(
"add",
{"a": 2, "b": 3, "context": "adding two numbers to demo the python mcp sdk"},
Expand Down
1 change: 0 additions & 1 deletion posthog/mcp/_event_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
class MCPAnalyticsEventType:
"""Protocol-shaped event types observed by the SDK (internal dispatch keys)."""

IDENTIFY = "posthog:identify"
CUSTOM = "posthog:custom"
MCP_FEEDBACK = "mcp:feedback"
MCP_MISSING_CAPABILITY = "mcp:missing_capability"
Expand Down
10 changes: 4 additions & 6 deletions posthog/mcp/_instrumentation.py
Original file line number Diff line number Diff line change
Expand Up @@ -377,9 +377,9 @@ async def prepare_request(
the session id to stamp on the event for this request.

``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 — identify,
initialize, and the call itself — lands in the conversation's session rather
than this instance's.
it anchors the session (ADR-0004) so every event of the request — initialize
and the call itself — lands in the conversation's session rather than this
instance's.

Callers pass it only for a handle the agent **echoed**. A freshly minted one
is unproven: this runs before the call, so delivery cannot be known yet, and
Expand Down Expand Up @@ -412,9 +412,7 @@ async def prepare_request(
and not _is_sse_request(extra)
):
_warn_stateless_session_not_wired(data)
identify_event = await handle_identify(data, session_id, request, extra)
if identify_event:
fire_and_forget(capture_event(data, identify_event), data)
await handle_identify(data, session_id, request, extra)
await _maybe_emit_initialize(
data, session_id, client_name, client_version, extra, protocol_version
)
Expand Down
85 changes: 10 additions & 75 deletions posthog/mcp/_internal.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
from __future__ import annotations

import asyncio
import json
import weakref
from collections import OrderedDict
from dataclasses import dataclass, field
Expand Down Expand Up @@ -139,25 +138,6 @@ def set_server_tracking_data(server: Any, data: MCPAnalyticsData) -> None:
_server_tracking[server] = data


def are_identities_equal(a: UserIdentity, b: UserIdentity) -> bool:
if a.distinct_id != b.distinct_id:
return False
if json.dumps(a.groups or {}, sort_keys=True) != json.dumps(
b.groups or {}, sort_keys=True
):
return False
a_props = a.properties or {}
b_props = b.properties or {}
if set(a_props.keys()) != set(b_props.keys()):
return False
for key in a_props:
if json.dumps(a_props[key], sort_keys=True, default=str) != json.dumps(
b_props[key], sort_keys=True, default=str
):
return False
return True


def merge_identities(
previous: Optional[UserIdentity], nxt: UserIdentity
) -> UserIdentity:
Expand All @@ -176,35 +156,18 @@ async def _maybe_await(value: Any) -> Any:
return value


def _captured_extra(extra: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""Project ``extra`` down to JSON-safe scalars before it is captured.

Callbacks (``identify``, ``event_properties``, ``intent_fallback``) receive
the full dict — on MCP SDK v2 that includes the raw request ``ctx`` so hosts
can read headers. Captured event parameters must not: the sanitizer leaves
opaque objects untouched and truncation stringifies them, which would ship
whatever the object's repr carries (headers, auth material, transport state)
to PostHog without key-based redaction."""
if extra is None:
return None
return {
key: value
for key, value in extra.items()
if value is None or isinstance(value, (str, int, float, bool))
}


async def handle_identify(
data: MCPAnalyticsData,
session_id: str,
request: Dict[str, Any],
extra: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
"""Resolve the optional ``identify`` callback, dedupe against the identity
cache, and return an ``$identify`` event to emit only when the identity has
materially changed (otherwise ``None``)."""
) -> None:
"""Resolve the optional ``identify`` callback and merge the result into the
session's cached identity. Every event of the session then carries it as
``distinct_id``/``$set``, so no standalone ``$identify`` is sent: without an
``$anon_distinct_id`` it would merge no person in ingestion."""
if not data.options.identify:
return None
return

try:
identify = data.options.identify
Expand All @@ -217,29 +180,16 @@ async def handle_identify(
log(
f"Warning: Supplied identify function returned null for session {session_id}"
)
return None
return

previous = data.identified_sessions.get(session_id)
merged = merge_identities(previous, identity_result)
has_changed = not (previous and are_identities_equal(previous, merged))
data.identified_sessions.set(session_id, merged)

if has_changed:
from ._event_types import MCPAnalyticsEventType

log(f"Identified session {session_id}")
return {
"session_id": session_id,
"resource_name": _get_request_resource_name(request),
"event_type": MCPAnalyticsEventType.IDENTIFY,
"parameters": {"request": request, "extra": _captured_extra(extra)},
"timestamp": datetime.now(timezone.utc),
}
data.identified_sessions.set(
session_id, merge_identities(previous, identity_result)
)
except Exception as error: # noqa: BLE001
log(
f"Error: identify function threw while identifying session {session_id} - {error}"
)
return None


async def resolve_event_properties(
Expand All @@ -254,18 +204,3 @@ async def resolve_event_properties(
except Exception as e: # noqa: BLE001
log(f"event_properties callback error: {e}")
return None


def _get_request_resource_name(request: Any) -> str:
"""The thing the request acts on: a tool/prompt ``name``, or the ``uri`` of a
resource read — which is the only name a ``resources/read`` request carries."""
if not isinstance(request, dict):
return "Unknown"
params = request.get("params")
if not isinstance(params, dict):
return "Unknown"
for key in ("name", "uri"):
value = params.get(key)
if isinstance(value, str):
return value
return "Unknown"
1 change: 0 additions & 1 deletion posthog/mcp/_posthog_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@

_BUILT_IN_EVENT_NAME_BY_TYPE = {
MCPAnalyticsEventType.CUSTOM: PostHogMCPAnalyticsEvent.CUSTOM,
MCPAnalyticsEventType.IDENTIFY: PostHogMCPAnalyticsEvent.IDENTIFY,
MCPAnalyticsEventType.MCP_FEEDBACK: PostHogMCPAnalyticsEvent.FEEDBACK,
MCPAnalyticsEventType.MCP_MISSING_CAPABILITY: PostHogMCPAnalyticsEvent.MISSING_CAPABILITY,
MCPAnalyticsEventType.MCP_INITIALIZE: PostHogMCPAnalyticsEvent.INITIALIZE,
Expand Down
1 change: 1 addition & 0 deletions posthog/mcp/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ class PostHogMCPAnalyticsEvent:
CUSTOM = "$mcp_custom"
EXCEPTION = "$exception"
FEEDBACK = "$mcp_feedback"
# Deprecated: the SDK no longer sends `$identify`.
IDENTIFY = "$identify"
INITIALIZE = "$mcp_initialize"
MISSING_CAPABILITY = "$mcp_missing_capability"
Expand Down
3 changes: 1 addition & 2 deletions posthog/test/mcp/test_fastmcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -256,8 +256,7 @@ async def test_identify_sets_distinct_id_and_groups():
assert calls[0]["distinct_id"] == "user_42"
assert calls[0]["properties"]["$groups"] == {"organization": "org_7"}
assert "$process_person_profile" not in calls[0]["properties"]
# an $identify event was emitted
assert _events(client, "$identify")
assert not _events(client, "$identify")


async def test_instrument_is_idempotent():
Expand Down
16 changes: 4 additions & 12 deletions posthog/test/mcp/test_lowlevel.py
Original file line number Diff line number Diff line change
Expand Up @@ -291,7 +291,7 @@ async def test_failed_resource_listing_is_captured() -> None:
assert "listing unavailable" in json.dumps(exceptions[0]["properties"])


async def test_identify_on_a_resource_read_is_named_by_the_uri() -> None:
async def test_identify_attributes_a_resource_read_without_an_identify_event() -> None:
server = make_server()
client = FakeClient()
instrument(
Expand All @@ -304,21 +304,13 @@ async def test_identify_on_a_resource_read_is_named_by_the_uri() -> None:

await server.request_handlers[mcp_types.ReadResourceRequest](
mcp_types.ReadResourceRequest(
params=mcp_types.ReadResourceRequestParams(
uri="https://fakeuser:[email protected]/guide"
)
params=mcp_types.ReadResourceRequestParams(uri="https://example.com/guide")
)
)
await _flush()

identified = _events(client, "$identify")
assert len(identified) == 1
# A resources/read request carries no `name`, so the uri is the only thing
# that can name it — sanitized like any other captured URL.
assert (
identified[0]["properties"]["$mcp_resource_name"]
== "https://%5Bredacted%[email protected]/guide"
)
assert not _events(client, "$identify")
assert _events(client, "$mcp_resource_read")[0]["distinct_id"] == "user_42"


async def test_tool_call_success_captures_intent():
Expand Down
5 changes: 0 additions & 5 deletions posthog/test/mcp/test_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -703,11 +703,6 @@ def test_redact_pii_is_not_quadratic_on_pathological_input():
"Get_Organization_Memberships",
"Get_Organization_Memberships",
),
(
MCPAnalyticsEventType.IDENTIFY,
"https://fakeuser:[email protected]/doc",
"https://%5Bredacted%[email protected]/doc",
),
(
MCPAnalyticsEventType.MCP_RESOURCES_READ,
"https://example.com/guide?token=fakesecret",
Expand Down
8 changes: 0 additions & 8 deletions posthog/test/mcp/test_units.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
from posthog.mcp._internal import (
IdentityCache,
MCPAnalyticsData,
are_identities_equal,
merge_identities,
)
from posthog.mcp.session import (
Expand Down Expand Up @@ -385,13 +384,6 @@ async def test_resolve_session_id_rolls_over_after_inactivity():
# --- identity cache / merge --------------------------------------------------


def test_are_identities_equal():
a = UserIdentity(distinct_id="u1", properties={"plan": "pro"}, groups={"org": "o1"})
b = UserIdentity(distinct_id="u1", properties={"plan": "pro"}, groups={"org": "o1"})
assert are_identities_equal(a, b)
assert not are_identities_equal(a, UserIdentity(distinct_id="u2"))


def test_merge_identities_merges_properties_and_keeps_distinct_id():
merged = merge_identities(
UserIdentity(distinct_id="u1", properties={"a": 1}),
Expand Down
16 changes: 3 additions & 13 deletions posthog/test/mcp/test_v2_mcpserver.py
Original file line number Diff line number Diff line change
Expand Up @@ -235,14 +235,12 @@ async def test_identify_sets_distinct_id_and_groups():
calls = _events(client, "$mcp_tool_call")
assert calls[0]["distinct_id"] == "user_42"
assert calls[0]["properties"]["$groups"] == {"organization": "org_7"}
assert _events(client, "$identify")
assert not _events(client, "$identify")


async def test_identify_callback_sees_ctx_but_capture_does_not():
async def test_identify_callback_sees_ctx():
"""The v2 adapters hand the raw request ``ctx`` to callbacks via ``extra`` so
hosts can read headers — but the captured $identify parameters must carry
only a scalar projection, or the stringified context (headers, transport
state) would ship to PostHog without key-based redaction."""
hosts can read headers."""
server = make_server()
client = FakeClient()
seen = {}
Expand All @@ -258,16 +256,8 @@ def identify(request, extra):
)
await _flush()

# the callback got the live context object...
assert seen["extra"]["ctx"] is not None

# ...but the captured event only carries JSON-safe scalars
identify_events = _events(client, "$identify")
assert identify_events
captured_extra = identify_events[0]["properties"]["$mcp_parameters"]["extra"]
assert "ctx" not in captured_extra
assert set(captured_extra) <= {"session_id"}


async def test_structured_output_tool_gets_the_conversation_handle():
"""A tool that declares an output schema is served to clients that read
Expand Down
45 changes: 44 additions & 1 deletion posthog/test/mcp/test_v2_wire_dual_era.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,15 @@
from contextlib import asynccontextmanager

import httpx
import pytest
from mcp.server.mcpserver import MCPServer

from posthog.mcp import (
decode_session_id,
derive_session_id_from_conversation,
instrument,
)
from posthog.mcp.types import MCPAnalyticsOptions
from posthog.mcp.types import MCPAnalyticsOptions, UserIdentity
from posthog.test.mcp._helpers import (
FakeClient,
events_named as _events,
Expand Down Expand Up @@ -241,6 +242,48 @@ async def test_modern_result_shape_survives_instrumentation():
assert result.get("structuredContent") == bare_result.get("structuredContent")


@pytest.mark.parametrize("era", ["legacy", "modern"])
async def test_identify_sends_no_identify_event_across_instances(era):
"""Each call lands on a fresh instance with no session token, so every
identity is first-seen. Attribution must still come from the callback."""
client = FakeClient()
options = MCPAnalyticsOptions(
enable_conversation_id=False,
identify=lambda request, extra: UserIdentity(
distinct_id="user_42", properties={"plan": "paid"}
),
)

for request_id in (1, 2):
server = make_server()
instrument(server, client, options)
async with wire(server) as http:
if era == "modern":
response = await modern_call(
http, "add", {"a": 1, "b": 2, "context": "pod"}, request_id
)
else:
call = rpc(
"tools/call",
{"name": "add", "arguments": {"a": 1, "b": 2, "context": "pod"}},
request_id,
)
headers = {
**legacy_headers(),
"mcp-protocol-version": LEGACY_PROTOCOL_VERSION,
}
response = await http.post("/mcp", json=call, headers=headers)
assert response.status_code == 200
await _flush()

assert not _events(client, "$identify")
calls = _events(client, "$mcp_tool_call")
assert len(calls) == 2
for call in calls:
assert call["distinct_id"] == "user_42"
assert call["properties"]["$set"] == {"plan": "paid"}


# --- legacy era (2025-11-25) on the v2 SDK -------------------------------------


Expand Down
Loading