PostHog · gesh · Aug 21, 2026 · Aug 20, 2026 · Aug 20, 2026 · Aug 20, 2026
@@ -0,0 +1,5 @@
+---
+posthog: minor
+---
+
+feat(mcp): emit `$mcp_error_message` and `$mcp_error_type` on failed MCP events. The reason a tool call failed previously lived only on the sibling `$exception` event, so PostHog's failures view — which reads the scalars off the primary event — showed empty error rows for every Python-backed MCP server, and switching off `enable_exception_autocapture` removed the reason entirely. Both values are read from the same `$exception_list` the sibling carries, so the two surfaces can never disagree, and the message inherits the existing 2048-character cap. `PostHogMCP.capture_tool_call()` and `capture_tools_list()` take a new optional `error_type` for custom dispatchers that want a coarse category (`"validation"`, `"timeout"`) instead of the thrown class name. Exception messages are also redacted before they leave — previously nothing sanitized the error payload, so the `$exception` sibling had been shipping them raw. Credential-looking words go through the SDK's own detector (entropy, known key formats, PEM markers), per word, so a message like `auth failed for sk-...` keeps its diagnostic text and loses only the key. Parity with `@posthog/mcp`, which sanitizes exception values the same way.
@@ -63,6 +63,7 @@ def capture_event(
         "user_intent_source": event_input.get("user_intent_source"),
         "is_error": event_input.get("is_error"),
         "error": event_input.get("error"),
+        "error_type": event_input.get("error_type"),
         "conversation_id": event_input.get("conversation_id"),
         "properties": event_input.get("properties"),
     }

@@ -139,6 +139,8 @@ def _add_common_properties(event: Event, properties: Dict[str, Any]) -> None:
         properties[_P.INTENT_SOURCE] = event["user_intent_source"]
     if event.get("is_error") is not None:
         properties[_P.IS_ERROR] = event["is_error"]
+    if event.get("is_error"):
+        _add_error_details(event, properties)
     if event.get("parameters") is not None:
         properties[_P.PARAMETERS] = event["parameters"]
     if event.get("response") is not None:
@@ -149,6 +151,36 @@ def _add_common_properties(event: Event, properties: Dict[str, Any]) -> None:
         properties["$set"] = {**identify_actor_data}
 
 
+def _add_error_details(event: Event, properties: Dict[str, Any]) -> None:
+    """Surface the failure reason on the primary event itself.
+
+    Without these the dashboard has to join to the ``$exception`` sibling to
+    know *why* a call failed — and that sibling can be switched off with
+    ``enable_exception_autocapture``, or never emitted when no error value was
+    passed. Both values are read off the ``$exception_list`` the sibling would
+    carry, so the two always agree; the message is already bounded to
+    ``_MAX_ERROR_MESSAGE_LENGTH`` because truncation runs before this mapping.
+    """
+    first: Dict[str, Any] = {}
+    error = event.get("error")
+    if isinstance(error, dict):
+        exception_list = error.get("$exception_list")
+        if isinstance(exception_list, list) and exception_list:
+            candidate = exception_list[0]
+            if isinstance(candidate, dict):
+                first = candidate
+
+    # An explicit coarse category (e.g. "validation", "timeout") beats the
+    # thrown type; a custom dispatcher can pass one that means something to the
+    # product, where the class name rarely does.
+    error_type = event.get("error_type") or first.get("type")
+    if error_type:
+        properties[_P.ERROR_TYPE] = error_type
+    message = first.get("value")
+    if message:
+        properties[_P.ERROR_MESSAGE] = message
+
+
 def _add_custom_properties(event: Event, properties: Dict[str, Any]) -> None:
     custom = event.get("properties")
     if custom:

@@ -39,7 +39,38 @@ def _should_redact_key(key: str) -> bool:
 def _sanitize_string(value: str) -> str:
     if len(value) >= _SIZE_GATE and _BASE64_PATTERN.match(value):
         return "[binary data redacted - not supported by PostHog MCP analytics]"
-    return _POSTHOG_TOKEN_PATTERN.sub(_REDACTED_VALUE, value)
+    return _redact_secret_tokens(_POSTHOG_TOKEN_PATTERN.sub(_REDACTED_VALUE, value))
+
+
+def _redact_secret_tokens(value: str) -> str:
+    """Redact credential-looking words, leaving the surrounding text intact.
+
+    The PostHog-token pattern above only knows ``phc_``/``phx_``; a failure
+    message like ``auth failed for sk-proj-...`` carries someone else's key.
+    Rather than enumerate every vendor's format — an arms race that fails
+    quietly in both directions — this reuses the SDK's own detector
+    (``exception_utils._looks_like_secret``: entropy, known formats such as AWS
+    key ids, PEM markers), which the code-variables path already ships.
+
+    Applied per whitespace-separated token, not to the whole string: redacting
+    an entire exception message would destroy the diagnostic value that
+    ``$mcp_error_message`` exists to provide, and ordinary prose is left alone
+    because no single word in it looks like a credential.
+    """
+    if " " not in value:
+        return _REDACTED_VALUE if _is_secret(value) else value
+    return " ".join(
+        _REDACTED_VALUE if _is_secret(word) else word for word in value.split(" ")
+    )
+
+
+def _is_secret(word: str) -> bool:
+    try:
+        from posthog.exception_utils import _looks_like_secret
+
+        return bool(word) and _looks_like_secret(word)
+    except Exception:  # noqa: BLE001 - redaction must never break capture
+        return False
 
 
 def sanitize_captured_value(value: Any) -> Any:
@@ -64,8 +95,8 @@ def sanitize_captured_value(value: Any) -> Any:
 
 
 def sanitize_event(event: Dict[str, Any]) -> Dict[str, Any]:
-    """Sanitize an event's response, parameters, and user_intent. Returns a new
-    shallow copy; does not mutate the input."""
+    """Sanitize an event's response, parameters, user_intent and error. Returns
+    a new shallow copy; does not mutate the input."""
     result = {**event}
 
     if result.get("response") is not None:
@@ -79,9 +110,41 @@ def sanitize_event(event: Dict[str, Any]) -> Dict[str, Any]:
     if result.get("user_intent") is not None:
         result["user_intent"] = sanitize_captured_value(result["user_intent"])
 
+    # An exception message is free text a server wrote, and it reaches PostHog
+    # on the $exception sibling and — since it is also surfaced as
+    # $mcp_error_message — on the primary event, so run it through the same
+    # sanitizer as every other captured value.
+    #
+    # That sanitizer redacts PostHog tokens and sensitive-looking keys; it is
+    # deliberately not a general credential scrubber, because enumerating every
+    # vendor's key format is an arms race that fails quietly in both directions.
+    # A host with strict requirements should gate free text in `before_send`.
+    # Same scope as @posthog/mcp's sanitizeCapturedValue.
+    if result.get("error") is not None:
+        result["error"] = _sanitize_exception_values(result["error"])
+
     return result
 
 
+def _sanitize_exception_values(error: Any) -> Any:
+    """Redact the ``value`` of every frame in an ``$exception_list``, leaving
+    the rest of the error-tracking shape untouched."""
+    if not isinstance(error, dict):
+        return error
+    exception_list = error.get("$exception_list")
+    if not isinstance(exception_list, list):
+        return error
+    return {
+        **error,
+        "$exception_list": [
+            {**exception, "value": sanitize_captured_value(exception.get("value"))}
+            if isinstance(exception, dict)
+            else exception
+            for exception in exception_list
+        ],
+    }
+
+
 def _sanitize_response(response: Any) -> Any:
     if response is None or not isinstance(response, (dict, list, str)):
         return sanitize_captured_value(response)

@@ -41,6 +41,7 @@
     ("server_version", _MAX_METADATA_LENGTH),
     ("client_name", _MAX_METADATA_LENGTH),
     ("client_version", _MAX_METADATA_LENGTH),
+    ("error_type", _MAX_METADATA_LENGTH),
 )
 
 _NORMALIZED_FIELDS = ("parameters", "response", "identify_actor_data", "error")

@@ -60,6 +60,8 @@ class PostHogMCPAnalyticsProperty:
     PROTOCOL_VERSION = "$mcp_protocol_version"
     CONVERSATION_ID = "$mcp_conversation_id"
     DURATION_MS = "$mcp_duration_ms"
+    ERROR_MESSAGE = "$mcp_error_message"
+    ERROR_TYPE = "$mcp_error_type"
     IS_ERROR = "$mcp_is_error"
     INTENT = "$mcp_intent"
     INTENT_SOURCE = "$mcp_intent_source"

@@ -87,6 +87,7 @@ def capture_tool_call(
         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,
         protocol_version: Optional[str] = None,
@@ -115,6 +116,7 @@ def capture_tool_call(
         event["response"] = response
         event["duration"] = duration_ms
         event["is_error"] = is_error
+        event["error_type"] = error_type
         _apply_intent(event, intent, intent_source)
         if is_error:
             event["error"] = capture_exception(
@@ -165,6 +167,7 @@ def capture_tools_list(
         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,
@@ -190,6 +193,7 @@ def capture_tools_list(
         event["response"] = response
         event["duration"] = duration_ms
         event["is_error"] = is_error
+        event["error_type"] = error_type
         if is_error:
             event["error"] = capture_exception(
                 error if error is not None else "tools/list failed"

@@ -40,7 +40,7 @@
 # plain dict (constructed and read with ``.get()`` throughout) to mirror the TS
 # plain-object pipeline. Snake_case keys map to the ``$mcp_*`` wire keys in
 # ``posthog_events``. Known keys: client_name, client_version, conversation_id,
-# duration, error, event_name, event_type, groups, id, identify_actor_data,
+# 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.