feat(mcp): set $mcp_error_status from the raised exception's cause chain

.sampo/changesets/mcp-error-status.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+`posthog.mcp`: a failed tool call now sets `$mcp_error_status` to the HTTP status behind it, read from the raised exception or its cause chain (`status_code`, `statusCode` or `status`, on the exception or its `response`, as `httpx.HTTPStatusError` and `requests.HTTPError` carry it). Failures from an upstream API now group by status in MCP Analytics instead of landing in "unknown".

posthog/mcp/_capture.py

@@ -69,6 +69,7 @@ def capture_event(
         "is_error": event_input.get("is_error"),
         "error": event_input.get("error"),
         "error_type": event_input.get("error_type"),
+        "error_status": event_input.get("error_status"),
         "conversation_id": event_input.get("conversation_id"),
         "properties": event_input.get("properties"),
     }

posthog/mcp/_exceptions.py

@@ -10,7 +10,7 @@
 
 from __future__ import annotations
 
-from typing import Any, List
+from typing import Any, Iterator, List, Optional, Set
 
 from posthog.exception_utils import exceptions_from_error_tuple
 
@@ -38,6 +38,53 @@ def capture_exception(error: Any) -> ErrorProperties:
     return _from_message(_safe_str(error))
 
 
+_STATUS_ATTRIBUTES = ("status_code", "statusCode", "status")
+_MAX_CHAIN_LINKS = 50
+
+
+def http_status(error: Any) -> Optional[int]:
+    """The HTTP status behind a failure, for ``$mcp_error_status``: the first
+    one found walking the raised exception's cause chain inward, so a status the
+    tool raised itself beats the one on the error it wrapped (httpx and requests
+    keep theirs on ``response.status_code``)."""
+    for link in _cause_chain(error):
+        for holder in (link, _attribute(link, "response")):
+            for name in _STATUS_ATTRIBUTES:
+                status = _attribute(holder, name)
+                if (
+                    isinstance(status, int)
+                    and not isinstance(status, bool)
+                    and 100 <= status <= 599
+                ):
+                    return int(status)
+    return None
+
+
+def _cause_chain(error: Any) -> Iterator[BaseException]:
+    """The chain a traceback prints, capped like ``$exception_list``."""
+    seen: Set[int] = set()
+    link = error
+    while (
+        isinstance(link, BaseException)
+        and id(link) not in seen
+        and len(seen) < _MAX_CHAIN_LINKS
+    ):
+        seen.add(id(link))
+        yield link
+        if link.__cause__ is not None:
+            link = link.__cause__
+        else:
+            link = None if link.__suppress_context__ else link.__context__
+
+
+def _attribute(value: Any, name: str) -> Any:
+    # A third-party property can raise anything; a failed probe must not drop the event.
+    try:
+        return getattr(value, name, None)
+    except Exception:
+        return None
+
+
 def _from_message(message: str) -> ErrorProperties:
     return {
         "$exception_list": [

posthog/mcp/_instrumentation.py

@@ -30,7 +30,7 @@
     resolve_conversation_id,
 )
 from ._event_types import MCPAnalyticsEventType
-from ._exceptions import capture_exception
+from ._exceptions import capture_exception, http_status
 from .feedback import (
     build_feedback_event_properties,
     build_feedback_intent,
@@ -826,6 +826,7 @@ async def record_tool_call(
         if error is not None:
             event["is_error"] = True
             event["error"] = capture_exception(error)
+            event["error_status"] = http_status(error)
 
         props = await resolve_event_properties(data, request, extra)
         input_aliases = None

posthog/mcp/_posthog_events.py

@@ -252,6 +252,8 @@ def _add_error_details(event: Event, properties: Dict[str, Any]) -> None:
     message = first.get("value")
     if message:
         properties[_P.ERROR_MESSAGE] = message
+    if event.get("error_status") is not None:
+        properties[_P.ERROR_STATUS] = event["error_status"]
 
 
 def _add_custom_properties(event: Event, properties: Dict[str, Any]) -> None:

posthog/mcp/constants.py

@@ -68,6 +68,7 @@ class PostHogMCPAnalyticsProperty:
     CONVERSATION_ID = "$mcp_conversation_id"
     DURATION_MS = "$mcp_duration_ms"
     ERROR_MESSAGE = "$mcp_error_message"
+    ERROR_STATUS = "$mcp_error_status"
     ERROR_TYPE = "$mcp_error_type"
     FEEDBACK_DETAILS = "$mcp_feedback_details"
     FEEDBACK_FRICTION_POINTS = "$mcp_feedback_friction_points"

posthog/mcp/posthog_mcp.py

@@ -33,7 +33,7 @@
     resolve_conversation_id,
 )
 from ._event_types import MCPAnalyticsEventType
-from ._exceptions import capture_exception
+from ._exceptions import capture_exception, http_status
 from ._instrumentation import (
     VIRTUAL_TOOL_FEEDBACK,
     VIRTUAL_TOOL_MISSING_CAPABILITY,
@@ -242,6 +242,7 @@ def capture_tool_call(
             event["error"] = capture_exception(
                 error if error is not None else f"Tool {tool_name} returned an error"
             )
+            event["error_status"] = http_status(error)
         self._emit(event)
 
     def capture_initialize(

posthog/mcp/types.py

@@ -63,7 +63,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, error_type, event_name, event_type, groups, id, identify_actor_data,
+# duration, error, error_status, 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, llm_model,

posthog/test/mcp/test_error_status.py

@@ -0,0 +1,145 @@
+"""``$mcp_error_status``: the HTTP status behind a failed call, read off the
+raised exception's cause chain (posthog-python#1223, sdk-specs mcp-analytics
+"Failure capture")."""
+
+from http import HTTPStatus
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+from posthog.mcp import PostHogMCP
+from posthog.mcp.constants import PostHogMCPAnalyticsProperty as P
+from posthog.test.mcp._helpers import events_named, flush_background
+
+
+class Failure(Exception):
+    def __init__(self, message="failed", **attributes):
+        super().__init__(message)
+        for name, value in attributes.items():
+            setattr(self, name, value)
+
+
+def _http_status_error(status):
+    request = httpx.Request("GET", "https://api.example.com/invoices/1")
+    response = httpx.Response(status, request=request)
+    return httpx.HTTPStatusError(
+        f"{status} from upstream", request=request, response=response
+    )
+
+
+def _raised_from(outer, inner):
+    try:
+        try:
+            raise inner
+        except Exception as error:
+            raise outer from error
+    except Exception as error:
+        return error
+
+
+def _raised_during(outer, inner):
+    try:
+        try:
+            raise inner
+        except Exception:
+            raise outer
+    except Exception as error:
+        return error
+
+
+def _raised_from_none(outer, inner):
+    try:
+        try:
+            raise inner
+        except Exception:
+            raise outer from None
+    except Exception as error:
+        return error
+
+
+def _cause_without_suppressed_context():
+    error = _raised_during(RuntimeError("lookup failed"), Failure(status=500))
+    error.__cause__ = Failure(status=404)
+    error.__suppress_context__ = False
+    return error
+
+
+def _chain_of(length, status):
+    error = Failure("innermost", status=status)
+    for _ in range(length - 1):
+        error = _raised_from(Failure("wrapper"), error)
+    return error
+
+
+class UnreadableResponse(Exception):
+    @property
+    def response(self):
+        raise RuntimeError("response not loaded")
+
+
+def _cycle():
+    first, second = Failure("first"), Failure("second")
+    first.__cause__, second.__cause__ = second, first
+    return first
+
+
+@pytest.mark.parametrize(
+    "error, status",
+    [
+        (_http_status_error(503), 503),
+        (_raised_from(RuntimeError("lookup failed"), _http_status_error(503)), 503),
+        (_raised_during(RuntimeError("lookup failed"), _http_status_error(429)), 429),
+        (
+            _raised_from_none(RuntimeError("lookup failed"), _http_status_error(503)),
+            None,
+        ),
+        (_raised_from(Failure(status_code=404), _http_status_error(503)), 404),
+        (Failure(status_code=502), 502),
+        (Failure(statusCode=502), 502),
+        (Failure(status=502), 502),
+        (Failure(status_code=None, status=502), 502),
+        (Failure(response=SimpleNamespace(status=502)), 502),
+        (Failure(status="failed", response=SimpleNamespace(status_code=401)), 401),
+        (Failure(status=HTTPStatus.BAD_GATEWAY), 502),
+        (Failure(status_code=42), None),
+        (Failure(status_code=600), None),
+        (Failure(status_code=True), None),
+        (Failure(status_code="503"), None),
+        (Failure(status_code=503.0), None),
+        (Failure("no status here"), None),
+        (_cause_without_suppressed_context(), 404),
+        (_chain_of(50, 503), 503),
+        (_chain_of(51, 503), None),
+        (_cycle(), None),
+        (UnreadableResponse("failed"), None),
+        ("upstream returned 503", None),
+    ],
+)
+async def test_failed_call_carries_the_status_from_the_cause_chain(error, status):
+    client = PostHogMCP("phc_test")
+    captured = []
+    client.capture = lambda event, **kw: captured.append({"event": event, **kw})
+
+    client.capture_tool_call("fetch_invoice", is_error=True, error=error)
+    await flush_background()
+
+    [call] = events_named(captured, "$mcp_tool_call")
+    assert call["properties"].get(P.ERROR_STATUS) == status
+
+
+async def test_explicit_status_wins_over_the_derived_one():
+    client = PostHogMCP("phc_test")
+    captured = []
+    client.capture = lambda event, **kw: captured.append({"event": event, **kw})
+
+    client.capture_tool_call(
+        "fetch_invoice",
+        is_error=True,
+        error=_http_status_error(503),
+        properties={P.ERROR_STATUS: 418},
+    )
+    await flush_background()
+
+    [call] = events_named(captured, "$mcp_tool_call")
+    assert call["properties"][P.ERROR_STATUS] == 418

posthog/test/mcp/test_fastmcp_tool_errors.py

@@ -7,6 +7,7 @@
 import logging
 import weakref
 
+import httpx
 import pytest
 
 pytest.importorskip("fastmcp")
@@ -33,6 +34,14 @@ def _tool_error():
     raise ToolError("bad input")
 
 
+def _upstream_status():
+    request = httpx.Request("GET", "https://api.example.com/invoices/1")
+    try:
+        httpx.Response(503, request=request).raise_for_status()
+    except httpx.HTTPStatusError as exc:
+        raise RuntimeError("invoice lookup failed") from exc
+
+
 def _tool_error_worded_like_fastmcp():
     try:
         _upstream()
@@ -41,34 +50,45 @@ def _tool_error_worded_like_fastmcp():
 
 
 @pytest.mark.parametrize(
-    "raise_in_tool, mask_error_details, error_type, message, chain",
+    "raise_in_tool, mask_error_details, error_type, message, chain, status",
     [
         (
             _chained,
             False,
             "RuntimeError",
             "lookup failed",
             ["RuntimeError", "ConnectionError"],
+            None,
         ),
         (
             _chained,
             True,
             "RuntimeError",
             "lookup failed",
             ["RuntimeError", "ConnectionError"],
+            None,
         ),
-        (_tool_error, False, "ToolError", "bad input", ["ToolError"]),
+        (_tool_error, False, "ToolError", "bad input", ["ToolError"], None),
         (
             _tool_error_worded_like_fastmcp,
             False,
             "ToolError",
             "Error calling tool: lookup failed",
             ["ToolError", "ConnectionError"],
+            None,
+        ),
+        (
+            _upstream_status,
+            True,
+            "RuntimeError",
+            "invoice lookup failed",
+            ["RuntimeError", "HTTPStatusError"],
+            503,
         ),
     ],
 )
 async def test_failed_call_carries_the_raised_exception(
-    raise_in_tool, mask_error_details, error_type, message, chain
+    raise_in_tool, mask_error_details, error_type, message, chain, status
 ):
     server = FastMCP("example-server", mask_error_details=mask_error_details)
 
@@ -89,6 +109,7 @@ def fail() -> str:
     assert call["properties"]["$mcp_is_error"] is True
     assert call["properties"]["$mcp_error_type"] == error_type
     assert call["properties"]["$mcp_error_message"] == message
+    assert call["properties"].get("$mcp_error_status") == status
     [exception] = events_named(sink, "$exception")
     exception_list = exception["properties"]["$exception_list"]
     raised = [e for e in exception_list if e["type"] in chain]

references/public_api_snapshot.txt

@@ -1029,6 +1029,7 @@ attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.CLIENT_VERSION = '$m
 attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.CONVERSATION_ID = '$mcp_conversation_id'
 attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.DURATION_MS = '$mcp_duration_ms'
 attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.ERROR_MESSAGE = '$mcp_error_message'
+attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.ERROR_STATUS = '$mcp_error_status'
 attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.ERROR_TYPE = '$mcp_error_type'
 attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.FEEDBACK_DETAILS = '$mcp_feedback_details'
 attribute posthog.mcp.constants.PostHogMCPAnalyticsProperty.FEEDBACK_FRICTION_POINTS = '$mcp_feedback_friction_points'