Classify MCP tool errors by operational category

src/mcp_server_appwrite/error_classification.py

@@ -0,0 +1,100 @@
+"""Classify public tool failures into bounded operational categories.
+
+The public operator wraps several distinct failure modes before they reach the
+telemetry and error-monitoring layers.  Keep the classification logic here so
+Prometheus and Sentry agree about which failures are expected and actionable.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from typing import Literal
+
+from appwrite_console.exception import AppwriteException
+
+ErrorCategory = Literal[
+    "write_confirmation",
+    "appwrite_4xx",
+    "appwrite_5xx",
+    "sdk_validation",
+    "internal",
+]
+
+ERROR_CATEGORIES: frozenset[str] = frozenset(
+    {
+        "write_confirmation",
+        "appwrite_4xx",
+        "appwrite_5xx",
+        "sdk_validation",
+        "internal",
+    }
+)
+
+
+class WriteConfirmationRequired(RuntimeError):
+    """A mutating hidden tool was called without explicit confirmation."""
+
+
+def classify_tool_error(exc: BaseException) -> ErrorCategory:
+    """Return the bounded operational category for an exception chain."""
+    chain = tuple(_exception_chain(exc))
+
+    if any(isinstance(item, WriteConfirmationRequired) for item in chain):
+        return "write_confirmation"
+
+    if any(_is_sdk_validation_error(item) for item in chain):
+        return "sdk_validation"
+
+    appwrite_error = next(
+        (item for item in chain if isinstance(item, AppwriteException)), None
+    )
+    if appwrite_error is not None:
+        code = _appwrite_status_code(appwrite_error)
+        if code is not None and 400 <= code < 500:
+            return "appwrite_4xx"
+        if code is not None and 500 <= code < 600:
+            return "appwrite_5xx"
+
+    return "internal"
+
+
+def _exception_chain(exc: BaseException) -> Iterator[BaseException]:
+    """Walk causes and contexts defensively, including malformed cycles."""
+    pending: list[BaseException] = [exc]
+    seen: set[int] = set()
+    while pending:
+        current = pending.pop()
+        if id(current) in seen:
+            continue
+        seen.add(id(current))
+        yield current
+
+        # Push context first so an explicit cause is inspected first.
+        if isinstance(current.__context__, BaseException):
+            pending.append(current.__context__)
+        if isinstance(current.__cause__, BaseException):
+            pending.append(current.__cause__)
+
+
+def _is_sdk_validation_error(exc: BaseException) -> bool:
+    error_type = type(exc)
+    if error_type.__name__ == "ValidationError" and error_type.__module__.startswith(
+        "pydantic"
+    ):
+        return True
+
+    # The console SDK normally chains the Pydantic error, but retain a narrow
+    # fallback for SDK versions that only preserve their parse-error message.
+    return isinstance(exc, AppwriteException) and str(exc).startswith(
+        "Unable to parse response into "
+    )
+
+
+def _appwrite_status_code(exc: AppwriteException) -> int | None:
+    raw_code = getattr(exc, "code", None)
+    if raw_code is None:
+        return None
+    try:
+        return int(raw_code)
+    except (TypeError, ValueError):
+        return None

src/mcp_server_appwrite/error_monitoring.py

@@ -15,6 +15,8 @@
 
 from appwrite_console.exception import AppwriteException
 
+from .error_classification import classify_tool_error
+
 _enabled = False
 
 _SENSITIVE_KEYS = {
@@ -98,6 +100,7 @@ def capture_exception(
         import sentry_sdk
 
         with sentry_sdk.new_scope() as scope:
+            scope.set_tag("mcp.error_category", classify_tool_error(exc))
             for key, value in (tags or {}).items():
                 if value is not None:
                     scope.set_tag(key, str(value))
@@ -160,8 +163,7 @@ def _should_capture(exc: BaseException) -> bool:
         return False
     if isinstance(exc, ValueError):
         return False
-    appwrite_error = _find_appwrite_exception(exc)
-    if appwrite_error is not None and _is_appwrite_client_error(appwrite_error):
+    if classify_tool_error(exc) in {"write_confirmation", "appwrite_4xx"}:
         return False
     return True
 
@@ -184,14 +186,6 @@ def _find_exception(
     return None
 
 
-def _is_appwrite_client_error(exc: AppwriteException) -> bool:
-    try:
-        code = int(getattr(exc, "code", 0) or 0)
-    except (TypeError, ValueError):
-        return False
-    return 400 <= code < 500
-
-
 def _already_captured(exc: BaseException) -> bool:
     current: BaseException | None = exc
     seen: set[int] = set()

src/mcp_server_appwrite/operator.py

@@ -28,6 +28,11 @@
     VERBS,
 )
 from .docs_search import DocsSearch
+from .error_classification import (
+    ErrorCategory,
+    WriteConfirmationRequired,
+    classify_tool_error,
+)
 from .tool_manager import ToolManager
 
 ToolContent = types.TextContent | types.ImageContent | types.EmbeddedResource
@@ -281,6 +286,7 @@ def execute_public_tool(
         start = time.monotonic()
         status = "success"
         error_type: str | None = None
+        error_category: ErrorCategory | None = None
         output_chars = 0
         telemetry.tool_call_started(name)
         try:
@@ -290,13 +296,15 @@ def execute_public_tool(
         except Exception as exc:
             status = "error"
             error_type = type(exc).__name__
+            error_category = classify_tool_error(exc)
             raise
         finally:
             telemetry.record_tool_call(
                 name,
                 status,
                 time.monotonic() - start,
                 error_type=error_type,
+                error_category=error_category,
                 input_chars=len(json.dumps(arguments)) if arguments else 0,
                 output_chars=output_chars,
             )
@@ -484,7 +492,7 @@ def _call_hidden_tool(self, raw_arguments: dict[str, Any]) -> list[ToolContent]:
             raw_arguments.get("confirm_write", raw_arguments.get("confirmWrite", False))
         )
         if entry.classification != "read" and not confirm_write:
-            raise RuntimeError(
+            raise WriteConfirmationRequired(
                 f"Tool {tool_name} is {entry.classification}. Re-run appwrite_call_tool with confirm_write=true if you intend to mutate Appwrite state."
             )
 

src/mcp_server_appwrite/telemetry.py

@@ -56,6 +56,7 @@
 from typing import Any, Iterable
 
 from .constants import ACTIVE_WINDOW_SECONDS, KNOWN_MCP_CLIENTS
+from .error_classification import ERROR_CATEGORIES, ErrorCategory
 
 _enabled = False
 _lock = threading.Lock()
@@ -254,7 +255,10 @@ def _build_instruments(meter: Any, transport: str, version: str) -> None:
     _instruments["tool_errors"] = meter.create_counter(
         "mcp.tool.errors",
         unit="{error}",
-        description="Failed public operator tool invocations by error type.",
+        description=(
+            "Failed public operator tool invocations by exception type and "
+            "operational category."
+        ),
     )
     _instruments["tool_inflight"] = meter.create_up_down_counter(
         "mcp.tool.inflight",
@@ -649,6 +653,7 @@ def record_tool_call(
     duration_s: float,
     *,
     error_type: str | None = None,
+    error_category: ErrorCategory | None = None,
     input_chars: int | None = None,
     output_chars: int | None = None,
 ) -> None:
@@ -663,10 +668,15 @@ def record_tool_call(
     )
     _safe_record("tool_duration", duration_s, {"tool_name": tool_name})
     if status == "error":
+        category = error_category if error_category in ERROR_CATEGORIES else "internal"
         _safe_add(
             "tool_errors",
             1,
-            {"tool_name": tool_name, "error_type": error_type or "unknown"},
+            {
+                "tool_name": tool_name,
+                "error_type": error_type or "unknown",
+                "error_category": category,
+            },
         )
 
     input_tokens = _estimate_tokens(input_chars or 0)

tests/unit/test_error_classification.py

@@ -0,0 +1,77 @@
+import unittest
+
+from appwrite_console.exception import AppwriteException
+from pydantic import BaseModel, ValidationError
+
+from mcp_server_appwrite.error_classification import (
+    WriteConfirmationRequired,
+    classify_tool_error,
+)
+
+
+class ErrorClassificationTests(unittest.TestCase):
+    def test_write_confirmation(self):
+        self.assertEqual(
+            classify_tool_error(WriteConfirmationRequired("confirm_write=true")),
+            "write_confirmation",
+        )
+
+    def test_wrapped_appwrite_4xx(self):
+        for code in (400, 401, 404, 409, 429, 499):
+            with self.subTest(code=code):
+                appwrite_error = AppwriteException("client error", str(code), None)
+                wrapped = RuntimeError("wrapped")
+                wrapped.__cause__ = appwrite_error
+
+                self.assertEqual(classify_tool_error(wrapped), "appwrite_4xx")
+
+    def test_appwrite_5xx(self):
+        for code in (500, 503, 599):
+            with self.subTest(code=code):
+                self.assertEqual(
+                    classify_tool_error(
+                        AppwriteException("upstream failed", code, "server_error")
+                    ),
+                    "appwrite_5xx",
+                )
+
+    def test_sdk_validation_takes_precedence_over_code(self):
+        class Provider(BaseModel):
+            options: dict
+
+        try:
+            Provider.model_validate({"options": []})
+        except ValidationError as validation_error:
+            appwrite_error = AppwriteException(
+                "Unable to parse response into Provider", 0, None
+            )
+            appwrite_error.__cause__ = validation_error
+        else:  # pragma: no cover - defensive
+            self.fail("Expected Pydantic validation to fail")
+
+        self.assertEqual(classify_tool_error(appwrite_error), "sdk_validation")
+
+    def test_sdk_validation_message_fallback(self):
+        error = AppwriteException(
+            "Unable to parse response into ProviderList: invalid model", 0, None
+        )
+        self.assertEqual(classify_tool_error(error), "sdk_validation")
+
+    def test_code_less_appwrite_and_unexpected_errors_are_internal(self):
+        self.assertEqual(
+            classify_tool_error(AppwriteException("network down", 0, None)),
+            "internal",
+        )
+        self.assertEqual(classify_tool_error(TypeError("boom")), "internal")
+
+    def test_exception_cycle_is_safe(self):
+        first = RuntimeError("first")
+        second = AppwriteException("not found", 404, "not_found")
+        first.__cause__ = second
+        second.__context__ = first
+
+        self.assertEqual(classify_tool_error(first), "appwrite_4xx")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_error_monitoring.py

@@ -5,6 +5,7 @@
 from appwrite_console.exception import AppwriteException
 
 from mcp_server_appwrite import error_monitoring
+from mcp_server_appwrite.error_classification import WriteConfirmationRequired
 
 
 class ErrorMonitoringTests(unittest.TestCase):
@@ -62,6 +63,9 @@ def __enter__(self):
             def __exit__(self, *args):
                 return False
 
+            def set_tag(self, key, value):
+                pass
+
         scope = FakeScope()
         with patch("sentry_sdk.capture_exception") as capture:
             with patch("sentry_sdk.new_scope", return_value=scope):
@@ -90,6 +94,16 @@ def test_appwrite_4xx_errors_are_not_captured(self):
         self.assertFalse(captured)
         capture.assert_not_called()
 
+    def test_write_confirmation_is_not_captured(self):
+        error_monitoring._enabled = True
+        exc = WriteConfirmationRequired("confirm_write=true")
+
+        with patch("sentry_sdk.capture_exception") as capture:
+            captured = error_monitoring.capture_exception(exc)
+
+        self.assertFalse(captured)
+        capture.assert_not_called()
+
     def test_wrapped_appwrite_4xx_errors_are_not_captured(self):
         error_monitoring._enabled = True
         exc = AppwriteException("not found", 404, "user_target_not_found")
@@ -163,6 +177,7 @@ def set_transaction_name(self, value):
         self.assertTrue(captured)
         capture.assert_called_once_with(exc)
         self.assertEqual(scope.tags["mcp.method"], "tools/call")
+        self.assertEqual(scope.tags["mcp.error_category"], "internal")
         self.assertEqual(scope.contexts["appwrite_mcp"]["arguments"], "[Filtered]")
         self.assertEqual(scope.contexts["appwrite_mcp"]["safe"], "ok")
         self.assertEqual(scope.transaction, "mcp.tools/call:appwrite_call_tool")
@@ -210,6 +225,7 @@ def set_transaction_name(self, value):
         capture.assert_called_once_with(exc)
         self.assertEqual(scope.tags["appwrite.project_id"], "project-1")
         self.assertEqual(scope.tags["appwrite.organization_id"], "org-1")
+        self.assertEqual(scope.tags["mcp.error_category"], "appwrite_5xx")
         self.assertEqual(
             scope.contexts["appwrite_mcp"]["appwrite"]["project_id"], "project-1"
         )

tests/unit/test_operator.py

@@ -3,6 +3,7 @@
 
 import mcp.types as types
 
+from mcp_server_appwrite.error_classification import WriteConfirmationRequired
 from mcp_server_appwrite.operator import CATALOG_URI, Operator, ResultStore
 from mcp_server_appwrite.tool_manager import ToolManager
 
@@ -397,7 +398,7 @@ def test_search_tools_scores_get_queries_against_get_tools(self):
     def test_call_tool_requires_confirm_write(self):
         runtime = self.make_runtime(lambda name, arguments, *_: [])
 
-        with self.assertRaisesRegex(RuntimeError, "confirm_write=true"):
+        with self.assertRaisesRegex(WriteConfirmationRequired, "confirm_write=true"):
             runtime.execute_public_tool(
                 "appwrite_call_tool",
                 {"tool_name": "tables_db_create", "arguments": {"database_id": "db"}},

tests/unit/test_server.py

@@ -16,6 +16,7 @@
 
 from mcp_server_appwrite import server as server_module
 from mcp_server_appwrite.catalog_policy import API_KEY_PROFILE, OAUTH_PROFILE
+from mcp_server_appwrite.error_classification import WriteConfirmationRequired
 from mcp_server_appwrite.server import (
     _coerce_argument,
     _configure_uploads,
@@ -504,7 +505,7 @@ def has_public_tool(self, name):
                 return True
 
             def execute_public_tool(self, name, arguments):
-                raise RuntimeError(
+                raise WriteConfirmationRequired(
                     "Tool tables_db_create is write. Re-run appwrite_call_tool "
                     "with confirm_write=true if you intend to mutate Appwrite state."
                 )