feat: identify MCP and AI analytics events by library (#917)

* feat: identify MCP and AI analytics libraries

* fix: keep library identity plumbing private

* fix: propagate MCP identity to all requests

Author: marandaneto

.sampo/changesets/virtuous-duchess-tursas.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Label MCP events and requests as `posthog-python-mcp`, and AI events as `posthog-ai`

posthog/ai/utils.py

@@ -10,6 +10,13 @@
 from posthog.ai.types import FormattedMessage, StreamingEventData, TokenUsage
 from posthog.client import Client as PostHogClient
 
+from ..version import VERSION as _POSTHOG_VERSION
+
+
+_AI_LIB_PROPERTIES = {
+    "$ai_lib": "posthog-ai",
+    "$ai_lib_version": _POSTHOG_VERSION,
+}
 
 _TOKEN_PROPERTY_KEYS = frozenset(
     {
@@ -63,7 +70,11 @@ def _ai_lane_enabled(ph_client) -> bool:
 
 
 def _capture_ai_event(ph_client, event: str, **kwargs):
-    """Capture a wrapper-emitted AI event, falling back to `capture()` for duck-typed clients without `capture_ai`."""
+    """Capture a wrapper-emitted AI event with the PostHog AI library identity."""
+    kwargs["properties"] = {
+        **_AI_LIB_PROPERTIES,
+        **(kwargs.get("properties") or {}),
+    }
     if _ai_lane_enabled(ph_client):
         capture_ai = getattr(ph_client, "capture_ai", None)
         if callable(capture_ai):

posthog/capture_v1.py

@@ -348,6 +348,7 @@ def _post_v1(
     request_id: str,
     compression: CaptureCompression = CaptureCompression.NONE,
     timeout: int = 15,
+    sdk_info: str = USER_AGENT,
     session: Optional["requests.Session"] = None,
 ) -> "requests.Response":
     """Perform a single ``POST /i/v1/analytics/events`` attempt.
@@ -364,9 +365,9 @@ def _post_v1(
     data = json.dumps(batch_body, cls=DatetimeSerializer)
     headers = {
         "Content-Type": "application/json",
-        "User-Agent": USER_AGENT,
+        "User-Agent": sdk_info,
         "Authorization": f"Bearer {api_key}",
-        _HEADER_SDK_INFO: USER_AGENT,
+        _HEADER_SDK_INFO: sdk_info,
         _HEADER_ATTEMPT: str(attempt),
         _HEADER_REQUEST_ID: request_id,
         _HEADER_REQUEST_TIMESTAMP: datetime.now(timezone.utc).isoformat(),
@@ -470,6 +471,7 @@ def _send_v1_batch(
     timeout: int = 15,
     max_retries: int = 3,
     historical_migration: bool = False,
+    sdk_info: str = USER_AGENT,
     session: Optional["requests.Session"] = None,
 ) -> None:
     """Deliver ``batch`` to the v1 endpoint with partial retry.
@@ -521,6 +523,7 @@ def _send_v1_batch(
                 request_id=request_id,
                 compression=compression,
                 timeout=timeout,
+                sdk_info=sdk_info,
                 session=session,
             )
         except Exception as e:

posthog/client.py

@@ -78,10 +78,13 @@
 from posthog.request import (
     AI_EVENTS_ENDPOINT,
     EVENTS_ENDPOINT,
+    USER_AGENT as _USER_AGENT,
     APIError,
     QuotaLimitError,
     RequestsConnectionError,
     RequestsTimeout,
+    _get as _get_with_identity,
+    _remote_config as _remote_config_with_identity,
     batch_post,
     determine_server_host,
     flags,
@@ -374,6 +377,7 @@ def __init__(
         max_msg_size,
         capture_mode,
         capture_compression,
+        sdk_info,
         eager_start,
     ):
         self.name = name
@@ -391,6 +395,7 @@ def __init__(
         self.max_msg_size = max_msg_size
         self.capture_mode = capture_mode
         self.capture_compression = capture_compression
+        self.sdk_info = sdk_info
         self._max_queue_size = max_queue_size
         self._thread_count = thread_count
         self._eager_start = eager_start
@@ -426,6 +431,7 @@ def _start_locked(self) -> None:
                 capture_mode=self.capture_mode,
                 capture_compression=self.capture_compression,
             )
+            consumer._sdk_info = self.sdk_info
             consumer._set_drain_signal(self._drain_signal)
             self.consumers.append(consumer)
 
@@ -913,6 +919,9 @@ def __init__(
         # `/i/v1/analytics/events`). Resolved here so the env-var fallback is
         # applied once; V0 is the default and keeps upgrades transparent.
         self.capture_mode = _resolve_capture_mode(capture_mode)
+        self._library_id = "posthog-python"
+        self._library_version = VERSION
+        self._sdk_info = f"{self._library_id}/{self._library_version}"
         # v1-only request compression; falls back to the legacy `gzip` flag when
         # neither the kwarg nor POSTHOG_CAPTURE_COMPRESSION is set.
         self.capture_compression = _resolve_capture_compression(
@@ -1033,6 +1042,7 @@ def __init__(
             max_retries=self.max_retries,
             timeout=timeout,
             historical_migration=historical_migration,
+            sdk_info=self._sdk_info,
         )
         self._analytics_lane = _Lane(
             name="analytics",
@@ -1068,6 +1078,19 @@ def __init__(
 
         self._warn_if_duplicate_async_client()
 
+    def _set_library_identity(self, library_id: str, library_version: str) -> None:
+        """Override the SDK identity stamped on events and outbound requests."""
+        self._library_id = library_id
+        self._library_version = library_version
+        self._sdk_info = f"{library_id}/{library_version}"
+        for lane in self._lanes:
+            lane.sdk_info = self._sdk_info
+            for consumer in lane.consumers:
+                consumer._sdk_info = self._sdk_info
+
+    def _request_identity_kwargs(self) -> Dict[str, str]:
+        return {"_user_agent": self._sdk_info} if self._sdk_info != _USER_AGENT else {}
+
     @property
     def queue(self) -> Queue:
         """The analytics lane's queue (kept for backwards compatibility)."""
@@ -1481,6 +1504,8 @@ def _get_flags_decision(
 
         if flag_keys_to_evaluate:
             request_data["flag_keys_to_evaluate"] = flag_keys_to_evaluate
+        if self._sdk_info != _USER_AGENT:
+            request_data["_user_agent"] = self._sdk_info
 
         resp_data = flags(
             self.api_key,
@@ -2279,8 +2304,8 @@ def _enqueue(self, msg, disable_geoip, lane=None, property_allowlist=None):
 
         if not msg.get("properties"):
             msg["properties"] = {}
-        msg["properties"]["$lib"] = "posthog-python"
-        msg["properties"]["$lib_version"] = VERSION
+        msg["properties"]["$lib"] = self._library_id
+        msg["properties"]["$lib_version"] = self._library_version
 
         if disable_geoip is None:
             disable_geoip = self.disable_geoip
@@ -2356,6 +2381,7 @@ def send_sync() -> None:
                         timeout=self.timeout,
                         max_retries=self.max_retries,
                         historical_migration=self.historical_migration,
+                        sdk_info=self._sdk_info,
                     )
                     return
 
@@ -2367,6 +2393,7 @@ def send_sync() -> None:
                     batch=[msg],
                     historical_migration=self.historical_migration,
                     path=lane.endpoint,
+                    **self._request_identity_kwargs(),
                 )
 
             if lane.run_sync_if_open(send_sync):
@@ -2925,12 +2952,14 @@ def _fetch_feature_flags_from_api(self):
 
         cache_data_to_store: Optional[FlagDefinitionCacheData] = None
         try:
-            response = get(
+            request_get = _get_with_identity if self._request_identity_kwargs() else get
+            response = request_get(
                 personal_api_key,
                 f"/flags/definitions?token={self.api_key}&send_cohorts",
                 self.host,
                 timeout=10,
                 etag=request_etag,
+                **self._request_identity_kwargs(),
             )
 
             with self._flag_definition_publication_lock:
@@ -3865,12 +3894,18 @@ def get_remote_config_payload(self, key: str):
             return None
 
         try:
-            return remote_config(
+            request_remote_config = (
+                _remote_config_with_identity
+                if self._request_identity_kwargs()
+                else remote_config
+            )
+            return request_remote_config(
                 self.personal_api_key,
                 self.api_key,
                 self.host,
                 key,
                 timeout=self.feature_flags_request_timeout_seconds,
+                **self._request_identity_kwargs(),
             )
         except Exception as e:
             self.log.exception(

posthog/consumer.py

@@ -10,6 +10,7 @@
 from posthog.capture_v1 import _backoff, _send_v1_batch
 from posthog.request import (
     EVENTS_ENDPOINT,
+    USER_AGENT as _USER_AGENT,
     APIError,
     DatetimeSerializer,
     batch_post,
@@ -132,6 +133,7 @@ def __init__(
         self.max_msg_size = max_msg_size
         self.capture_mode = capture_mode
         self.capture_compression = capture_compression
+        self._sdk_info = _USER_AGENT
         self._drain_signal: Optional[_DrainSignal] = None
         self._drain_on_stop = False
         # It's important to set running in the constructor: if we are asked to
@@ -300,6 +302,7 @@ def request(self, batch):
                 timeout=self.timeout,
                 max_retries=self.retries,
                 historical_migration=self.historical_migration,
+                sdk_info=self._sdk_info,
             )
             return
         self._send(batch, self.endpoint)
@@ -332,6 +335,11 @@ def is_retryable(exc):
                     batch=batch,
                     historical_migration=self.historical_migration,
                     path=path,
+                    **(
+                        {"_user_agent": self._sdk_info}
+                        if self._sdk_info != _USER_AGENT
+                        else {}
+                    ),
                 )
                 return
             except Exception as e:

posthog/mcp/README.md

@@ -16,6 +16,11 @@ analytics = instrument(server, posthog)
 Install is just `pip install posthog`. `instrument()` needs the MCP SDK at runtime,
 but anyone wrapping a server already has it.
 
+MCP analytics events report `$lib: "posthog-python-mcp"`. Because `$lib` is a
+client-level identity, `instrument()` relabels every event sent by the client passed
+to it. Use a client dedicated to MCP analytics if the application also captures
+unrelated events.
+
 ## Stateless / multi-pod servers
 
 A stateless MCP server issues no session id, so `$session_id` fragments across pods

posthog/mcp/__init__.py

@@ -303,6 +303,11 @@ def instrument(
                 "on SDK 2.x, or jlowin's fastmcp.FastMCP) or a low-level mcp.server.Server."
             )
 
+        if client is not None:
+            from ._lib_identity import apply_mcp_lib_identity
+
+            apply_mcp_lib_identity(client)
+
         # Zero-config stateless minting: wrap the server's ASGI-app factories so a
         # stateless/multi-pod deployment keeps one $session_id + the client harness
         # across pods with no extra setup. No-op for stdio / low-level servers.

posthog/mcp/_lib_identity.py

@@ -0,0 +1,15 @@
+"""Apply the MCP-specific identity to the underlying PostHog client."""
+
+from __future__ import annotations
+
+from ..client import Client
+
+from .constants import POSTHOG_MCP_LIB_NAME
+from .version import __version__
+
+
+def apply_mcp_lib_identity(client: Client) -> None:
+    """Relabel every event sent by ``client`` as coming from ``posthog.mcp``."""
+    set_identity = getattr(client, "_set_library_identity", None)
+    if set_identity is not None:
+        set_identity(POSTHOG_MCP_LIB_NAME, __version__)

posthog/mcp/constants.py

@@ -33,6 +33,7 @@
 )
 
 POSTHOG_MCP_ANALYTICS_SOURCE = "posthog_mcp_analytics"
+POSTHOG_MCP_LIB_NAME = "posthog-python-mcp"
 
 
 class PostHogMCPAnalyticsEvent:

posthog/mcp/posthog_mcp.py

@@ -24,6 +24,7 @@
 from ._event_types import MCPAnalyticsEventType
 from ._exceptions import capture_exception
 from ._instrumentation import drain_pending_sync, fire_and_forget
+from ._lib_identity import apply_mcp_lib_identity
 from ._sink import McpCaptureOptions, McpEventSink
 from .tools import build_report_missing_descriptor
 from .types import (
@@ -51,6 +52,7 @@ def __init__(
         **kwargs: Any,
     ) -> None:
         super().__init__(api_key, **kwargs)
+        apply_mcp_lib_identity(self)
         self._mcp_sink = McpEventSink(self)
         self._missing_capability_tool_name = (
             missing_capability_tool_name or _GET_MORE_TOOLS_NAME