feat(mcp): capture server build metadata

.sampo/changesets/mcp-server-build.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Add optional MCP server build metadata. Set `server_build` to record an immutable deployment identifier as `$mcp_server_build` on MCP events from automatic instrumentation, custom dispatchers, and custom events captured through `PostHogMCP`.

posthog/mcp/README.md

@@ -50,6 +50,23 @@ from posthog.mcp import MCPAnalyticsOptions, instrument
 instrument(server, posthog, MCPAnalyticsOptions(capture_model=False, enable_conversation_id=False))
 ```
 
+Set `server_build` to connect each MCP event to the exact deployed code. Use an
+immutable value such as a Git commit SHA or a container image digest.
+
+```python
+import os
+
+instrument(
+    server,
+    posthog,
+    MCPAnalyticsOptions(server_build=os.environ.get("GIT_SHA")),
+)
+```
+
+The value must contain 1 to 256 characters. The SDK records it as
+`$mcp_server_build` on all MCP events. `PostHogMCP.capture()` also adds it to
+custom events unless the event provides its own value.
+
 Model capture adds an `llm_model` argument to compatible tool schemas, required on the official
 high-level adapters and optional elsewhere. Dispatch never enforces it, so servers keep working;
 strict-schema clients see the new field. Set `capture_model=False` to leave schemas untouched.

posthog/mcp/__init__.py

@@ -38,6 +38,7 @@
 from posthog.client import Client
 
 from ._capture import capture_event
+from ._server_build import validate_server_build
 from .constants import (
     POSTHOG_MCP_ANALYTICS_SOURCE,
     PostHogMCPAnalyticsEvent,
@@ -248,6 +249,7 @@ def instrument(
     :param options: Optional :class:`MCPAnalyticsOptions`.
     """
     opts = options or MCPAnalyticsOptions()
+    server_build = validate_server_build(opts.server_build)
 
     # Install the logger first so the version advisory below (and any warning) is
     # actually visible rather than going to the default no-op sink.
@@ -299,7 +301,10 @@ def instrument(
         if data is None:
             sink = McpEventSink(client) if client is not None else None
             data = MCPAnalyticsData(
-                options=opts, sink=sink, session_id=new_session_id()
+                options=opts,
+                sink=sink,
+                session_id=new_session_id(),
+                server_build=server_build,
             )
 
         if is_fastmcp_v2(server) and uses_v2_handler_registry(key):

posthog/mcp/_capture.py

@@ -45,6 +45,7 @@ def capture_event(
         "duration": duration,
         "sdk_language": "Python",
         "sdk_version": __version__,
+        "server_build": data.server_build,
         "server_name": data.server_name,
         "server_version": data.server_version,
         "client_name": event_input.get("client_name"),

posthog/mcp/_internal.py

@@ -101,6 +101,9 @@ class MCPAnalyticsData:
     # server can't accumulate one entry per session forever.
     initialized_sessions: "OrderedDict[str, None]" = field(default_factory=OrderedDict)
     server_name: Optional[str] = None
+    # Validated once during setup. Keep it separate from the caller-owned
+    # options object so later option changes cannot alter event attribution.
+    server_build: Optional[str] = None
     server_version: Optional[str] = None
     # A strong wrapper reference would retain the low-level WeakKeyDictionary key.
     standalone_fastmcp: Optional["weakref.ReferenceType[Any]"] = None

posthog/mcp/_posthog_events.py

@@ -66,6 +66,7 @@ def _build_capture_event(event: Event) -> PostHogCaptureEvent:
     _add_groups(event, properties)
     _add_common_properties(event, properties)
     _add_custom_properties(event, properties)
+    _add_server_build(event, properties)
 
     event_name = (
         event.get("event_name") or _BUILT_IN_EVENT_NAME_BY_TYPE[event["event_type"]]
@@ -252,6 +253,12 @@ def _add_custom_properties(event: Event, properties: Dict[str, Any]) -> None:
             properties[key] = value
 
 
+def _add_server_build(event: Event, properties: Dict[str, Any]) -> None:
+    """Apply the validated build after custom properties so it cannot be replaced."""
+    if event.get("server_build"):
+        properties[_P.SERVER_BUILD] = event["server_build"]
+
+
 def _build_exception_event(event: Event) -> PostHogCaptureEvent:
     properties: Dict[str, Any] = {}
     _add_session_id(event, properties)
@@ -285,6 +292,7 @@ def _build_exception_event(event: Event) -> PostHogCaptureEvent:
         properties[_P.PROTOCOL_VERSION] = event["protocol_version"]
 
     _add_custom_properties(event, properties)
+    _add_server_build(event, properties)
 
     return {
         "event": PostHogMCPAnalyticsEvent.EXCEPTION,

posthog/mcp/_server_build.py

@@ -0,0 +1,20 @@
+"""Validate server build identifiers for MCP analytics events."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+MAX_SERVER_BUILD_LENGTH = 256
+
+
+def validate_server_build(server_build: Any) -> Optional[str]:
+    """Reject build identifiers that the event pipeline cannot record exactly."""
+    if server_build is None:
+        return None
+    if not isinstance(server_build, str) or not server_build:
+        raise TypeError("server_build must be a non-empty string.")
+    if len(server_build) > MAX_SERVER_BUILD_LENGTH:
+        raise ValueError(
+            f"server_build must not exceed {MAX_SERVER_BUILD_LENGTH} characters."
+        )
+    return server_build

posthog/mcp/_truncation.py

@@ -20,6 +20,8 @@
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
+from ._server_build import MAX_SERVER_BUILD_LENGTH
+
 MAX_DEPTH = 10
 MAX_BREADTH = 100
 MAX_STRING_LENGTH = 32_768  # 32KB
@@ -250,6 +252,13 @@ def _collect_string_paths(
         return
     if isinstance(obj, dict):
         for key, value in obj.items():
+            if (
+                not current_path
+                and key == "server_build"
+                and isinstance(value, str)
+                and len(value) <= MAX_SERVER_BUILD_LENGTH
+            ):
+                continue
             _collect_string_paths(value, current_path + [str(key)], results)
 
 

posthog/mcp/constants.py

@@ -94,6 +94,7 @@ class PostHogMCPAnalyticsProperty:
     RESOURCE_NAME = "$mcp_resource_name"
     RESPONSE = "$mcp_response"
     SERVER_NAME = "$mcp_server_name"
+    SERVER_BUILD = "$mcp_server_build"
     SERVER_VERSION = "$mcp_server_version"
     SESSION_ID = "$session_id"
     SOURCE = "$mcp_source"

posthog/mcp/posthog_mcp.py

@@ -15,7 +15,10 @@
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional, Set, Tuple, Union
 
-from posthog.client import Client
+from typing_extensions import Unpack
+
+from ..args import OptionalCaptureArgs
+from ..client import Client
 
 from ._context_parameters import (
     add_context_parameter_to_schema,
@@ -43,6 +46,8 @@
     resolve_model,
 )
 from ._sink import McpCaptureOptions, McpEventSink
+from ._server_build import validate_server_build
+from .constants import PostHogMCPAnalyticsProperty
 from .feedback import (
     build_feedback_event_properties,
     build_feedback_intent,
@@ -70,8 +75,9 @@
 class PostHogMCP(Client):
     """A drop-in posthog ``Client`` with ``capture_tool_call`` / ``capture_initialize``
     / ``capture_tools_list`` / ``capture_missing_capability`` / ``capture_feedback``
-    plus ``prepare_tool_list`` and ``prepare_tool_call`` helpers. ``capture``,
-    ``flush``, ``shutdown``, feature flags, etc. all work unchanged."""
+    plus ``prepare_tool_list`` and ``prepare_tool_call`` helpers. ``capture`` adds
+    the configured server build to custom events. Other inherited methods work
+    unchanged."""
 
     def __init__(
         self,
@@ -80,8 +86,10 @@ def __init__(
         mcp_exception_autocapture: bool = True,
         capture_model: Union[bool, MCPAnalyticsModelOptions] = True,
         collect_feedback: Union[bool, CollectFeedbackOptions] = False,
+        server_build: Optional[str] = None,
         **kwargs: Any,
     ) -> None:
+        self._server_build = validate_server_build(server_build)
         super().__init__(api_key, **kwargs)
         apply_mcp_lib_identity(self)
         self._mcp_sink = McpEventSink(self)
@@ -133,6 +141,19 @@ def shutdown(self) -> None:
 
     # --- capture methods -----------------------------------------------------
 
+    def capture(
+        self, event: str, **kwargs: Unpack[OptionalCaptureArgs]
+    ) -> Optional[str]:
+        """Capture a custom event with the configured server build as a default."""
+        if self._server_build is not None:
+            properties = dict(kwargs.get("properties") or {})
+            if properties.get(PostHogMCPAnalyticsProperty.SERVER_BUILD) is None:
+                properties[PostHogMCPAnalyticsProperty.SERVER_BUILD] = (
+                    self._server_build
+                )
+            kwargs["properties"] = properties
+        return super().capture(event, **kwargs)
+
     def capture_tool_call(
         self,
         tool_name: str,
@@ -549,6 +570,7 @@ def _base_event(
             # off the request automatically.
             "client_user_agent": client_user_agent,
             "vendor_client": vendor_client,
+            "server_build": self._server_build,
         }
         if distinct_id:
             event["identify_actor_given_id"] = distinct_id

posthog/mcp/types.py

@@ -209,9 +209,11 @@ class MCPAnalyticsOptions:
     # defaults; the object form renames the tool, replaces its description,
     # declares host-specific extra_properties, or wires an on_feedback handler.
     # Covers what `report_missing` covers (as feedback_type "missing_capability"),
-    # so new integrations should enable only one of the two. New field appended
-    # last: positional construction of the earlier fields must keep working.
+    # so new integrations should enable only one of the two.
     collect_feedback: Union[bool, CollectFeedbackOptions] = False
+    # Exact deployment identifier recorded as `$mcp_server_build`. Use an
+    # immutable value such as a Git commit SHA or a container image digest.
+    server_build: Optional[str] = None
 
 
 @dataclass

posthog/test/mcp/test_features_m4.py

@@ -124,6 +124,29 @@ async def test_fastmcp_conversation_id_captured():
     assert calls and calls[0]["properties"].get("$mcp_conversation_id")  # minted
 
 
+async def test_fastmcp_captures_server_build_on_all_events():
+    server = make_fastmcp()
+    client = FakeClient()
+    options = MCPAnalyticsOptions(server_build="sha-abc123")
+    instrument(
+        server,
+        client,
+        options,
+    )
+    options.server_build = "changed-after-setup"
+
+    list_handler = server._mcp_server.request_handlers[mcp_types.ListToolsRequest]
+    await list_handler(mcp_types.ListToolsRequest(method="tools/list"))
+    await server._tool_manager.call_tool("add", {"a": 1, "b": 2}, convert_result=True)
+    await _flush()
+
+    assert client.events
+    assert all(
+        event["properties"]["$mcp_server_build"] == "sha-abc123"
+        for event in client.events
+    )
+
+
 async def test_lowlevel_conversation_id_captured_and_prompt_back():
     server = make_lowlevel()
     client = FakeClient()

posthog/test/mcp/test_posthog_mcp.py

@@ -47,15 +47,58 @@ async def test_capture_tool_call_success():
 
 
 async def test_capture_tool_call_error_fans_out_exception():
-    client, captured = make_client()
+    client, captured = make_client(server_build="sha-abc123")
     client.capture_tool_call(
-        "broken", is_error=True, error=RuntimeError("kaboom"), distinct_id="u"
+        "broken",
+        is_error=True,
+        error=RuntimeError("kaboom"),
+        distinct_id="u",
+        properties={"$mcp_server_build": "custom-value"},
     )
     await _flush()
 
     assert _events(captured, "$mcp_tool_call")[0]["properties"]["$mcp_is_error"] is True
     exc = _events(captured, "$exception")
     assert exc and exc[0]["properties"]["$exception_list"][0]["value"] == "kaboom"
+    assert all(
+        event["properties"]["$mcp_server_build"] == "sha-abc123"
+        for event in [*_events(captured, "$mcp_tool_call"), *exc]
+    )
+
+
+@pytest.mark.parametrize("server_build", ["", 123, "x" * 257])
+def test_server_build_validation(server_build):
+    with pytest.raises((TypeError, ValueError)):
+        make_client(server_build=server_build)
+
+
+def test_capture_adds_server_build_to_custom_events():
+    captured = []
+
+    def before_send(event):
+        captured.append(event)
+        return event
+
+    client = PostHogMCP(
+        "phc_test",
+        send=False,
+        before_send=before_send,
+        server_build="default-build",
+    )
+    client.capture("custom event", properties={"existing": True})
+    client.capture(
+        "event build",
+        properties={"$mcp_server_build": "event-build"},
+    )
+    client.capture(
+        "null build",
+        properties={"$mcp_server_build": None},
+    )
+
+    assert captured[0]["properties"]["existing"] is True
+    assert captured[0]["properties"]["$mcp_server_build"] == "default-build"
+    assert captured[1]["properties"]["$mcp_server_build"] == "event-build"
+    assert captured[2]["properties"]["$mcp_server_build"] == "default-build"
 
 
 async def test_mcp_events_use_mcp_library_identity():

posthog/test/mcp/test_truncation.py

@@ -145,6 +145,20 @@ def test_truncate_event_enforces_byte_budget_with_many_strings():
     assert _json_byte_size(out) <= MAX_EVENT_BYTES
 
 
+def test_truncate_event_preserves_valid_server_build():
+    server_build = "b" * 256
+    event = {
+        "event_type": "$mcp_tool_call",
+        "server_build": server_build,
+        "parameters": {f"field_{i}": "z" * 5000 for i in range(60)},
+    }
+
+    out = truncate_event(event)
+
+    assert _json_byte_size(out) <= MAX_EVENT_BYTES
+    assert out["server_build"] == server_build
+
+
 def test_truncate_event_caps_single_huge_string_under_budget():
     out = truncate_event({"parameters": {"blob": "z" * 300_000}})
     assert _json_byte_size(out) <= MAX_EVENT_BYTES