fix: handle generated SDK input validation

src/mcp_server_appwrite/error_classification.py

@@ -17,6 +17,7 @@
     "write_confirmation",
     "appwrite_4xx",
     "appwrite_5xx",
+    "sdk_input_validation",
     "sdk_validation",
     "response_too_large",
     "internal",
@@ -27,6 +28,7 @@
         "write_confirmation",
         "appwrite_4xx",
         "appwrite_5xx",
+        "sdk_input_validation",
         "sdk_validation",
         "response_too_large",
         "internal",
@@ -83,6 +85,12 @@ def classify_tool_error(exc: BaseException) -> ErrorCategory:
     )
     if appwrite_error is not None:
         code = _appwrite_status_code(appwrite_error)
+        if (
+            code == 0
+            and appwrite_error.type == "sdk_input_validation"
+            and appwrite_error.response is None
+        ):
+            return "sdk_input_validation"
         if code is not None and 400 <= code < 500:
             return "appwrite_4xx"
         if code is not None and 500 <= code < 600:

src/mcp_server_appwrite/error_monitoring.py

@@ -171,7 +171,7 @@ def _should_capture(exc: BaseException) -> bool:
         return False
 
     category = classify_tool_error(exc)
-    if category in {"write_confirmation", "appwrite_4xx"}:
+    if category in {"write_confirmation", "appwrite_4xx", "sdk_input_validation"}:
         return False
     # Pydantic validation errors are ValueError subclasses, but SDK response
     # validation is actionable model drift and must remain visible.

tests/unit/test_error_classification.py

@@ -41,6 +41,31 @@ def test_appwrite_5xx(self):
                     "appwrite_5xx",
                 )
 
+    def test_sdk_input_validation(self):
+        error = AppwriteException(
+            'Invalid parameter: "filters" must be an array of strings',
+            type="sdk_input_validation",
+        )
+        wrapped = RuntimeError("wrapped")
+        wrapped.__cause__ = error
+
+        for failure in (error, wrapped):
+            with self.subTest(failure=type(failure).__name__):
+                self.assertEqual(classify_tool_error(failure), "sdk_input_validation")
+
+    def test_sdk_input_validation_type_does_not_hide_responses(self):
+        for code, response, expected in (
+            (503, None, "appwrite_5xx"),
+            (0, {}, "internal"),
+            (0, "", "internal"),
+        ):
+            with self.subTest(code=code, response=response):
+                error = AppwriteException(
+                    "upstream failed", code, "sdk_input_validation", response
+                )
+
+                self.assertEqual(classify_tool_error(error), expected)
+
     def test_sdk_validation_takes_precedence_over_code(self):
         class Provider(BaseModel):
             options: dict
@@ -49,7 +74,7 @@ class Provider(BaseModel):
             Provider.model_validate({"options": []})
         except ValidationError as validation_error:
             appwrite_error = AppwriteException(
-                "Unable to parse response into Provider", 0, None
+                "Unable to parse response into Provider", 0, "sdk_input_validation"
             )
             appwrite_error.__cause__ = validation_error
         else:  # pragma: no cover - defensive

tests/unit/test_error_monitoring.py

@@ -70,16 +70,55 @@ def test_wrapped_value_errors_are_not_captured(self):
         capture.assert_not_called()
 
     def test_wrapped_sdk_validation_errors_are_captured(self):
+        error_monitoring._enabled = True
+
         class Payload(BaseModel):
             required: str
 
         try:
             Payload.model_validate({})
         except ValidationError as exc:
+            error = AppwriteException("invalid response", type="sdk_input_validation")
+            error.__cause__ = exc
             wrapped = RuntimeError("wrapped")
-            wrapped.__cause__ = exc
+            wrapped.__cause__ = error
+
+        with patch("sentry_sdk.capture_exception") as capture:
+            captured = error_monitoring.capture_exception(wrapped)
+
+        self.assertTrue(captured)
+        capture.assert_called_once_with(wrapped)
+
+    def test_sdk_input_validation_is_not_captured(self):
+        error_monitoring._enabled = True
+
+        for wrapped in (False, True):
+            with self.subTest(wrapped=wrapped):
+                error = AppwriteException(
+                    "invalid filters", type="sdk_input_validation"
+                )
+                failure = RuntimeError("wrapped") if wrapped else error
+                if wrapped:
+                    failure.__cause__ = error
+                with patch("sentry_sdk.capture_exception") as capture:
+                    captured = error_monitoring.capture_exception(failure)
+
+                self.assertFalse(captured)
+                capture.assert_not_called()
+
+    def test_sdk_input_validation_does_not_hide_unexpected_failures(self):
+        error_monitoring._enabled = True
+        for error in (
+            AppwriteException("network down"),
+            AppwriteException("upstream failed", 503, "sdk_input_validation"),
+            AppwriteException("invalid response", 0, "sdk_input_validation", {}),
+        ):
+            with self.subTest(error=error):
+                with patch("sentry_sdk.capture_exception") as capture:
+                    captured = error_monitoring.capture_exception(error)
 
-        self.assertTrue(error_monitoring._should_capture(wrapped))
+                self.assertTrue(captured)
+                capture.assert_called_once_with(error)
 
     def test_client_disconnects_are_not_captured(self):
         error_monitoring._enabled = True

tests/unit/test_server.py

@@ -537,6 +537,113 @@ async def run_check():
         with patch.dict(server_module.SERVICE_CLASSES, {"functions": FunctionsService}):
             asyncio.run(run_check())
 
+    def test_call_tool_returns_sdk_input_validation_for_unencoded_queries(self):
+        client = build_introspection_client()
+        manager = register_services(client, profile=API_KEY_PROFILE)
+        server = build_mcp_server(build_operator(manager, client), transport="stdio")
+        entry = server.get_request_handler("tools/call")
+        self.assertIsNotNone(entry)
+
+        async def run_check():
+            ctx = Mock()
+            ctx.protocol_version = "2026-07-28"
+            ctx.meta = None
+            ctx.session.client_params = None
+            for queries in (
+                [{"method": "limit", "values": [1]}],
+                [
+                    '{"method":"limit","values":[1]}',
+                    {"method": "offset", "values": [1]},
+                ],
+                {"method": "limit", "values": [1]},
+                '{"method":"limit","values":[1]}',
+                [1],
+            ):
+                with self.subTest(queries=queries):
+                    params = types.CallToolRequestParams(
+                        name="appwrite_call_tool",
+                        arguments={
+                            "tool_name": "tables_db_list_rows",
+                            "arguments": {
+                                "database_id": "database-id",
+                                "table_id": "table-id",
+                                "queries": queries,
+                            },
+                        },
+                    )
+                    result = await entry.handler(ctx, params)
+
+                    self.assertTrue(result.is_error)
+                    self.assertIn("type=sdk_input_validation", result.content[0].text)
+                    self.assertIn("queries", result.content[0].text)
+                    self.assertIn("string", result.content[0].text)
+
+        with (
+            patch(
+                "requests.sessions.Session.request",
+                side_effect=AssertionError(
+                    "Invalid inputs must not send HTTP requests"
+                ),
+            ) as request,
+            patch.object(server_module.error_monitoring, "_enabled", True),
+            patch("sentry_sdk.capture_exception") as capture,
+        ):
+            asyncio.run(run_check())
+
+        request.assert_not_called()
+        capture.assert_not_called()
+
+    def test_call_tool_preserves_encoded_queries(self):
+        client = build_introspection_client()
+        manager = register_services(client, profile=API_KEY_PROFILE)
+        server = build_mcp_server(build_operator(manager, client), transport="stdio")
+        entry = server.get_request_handler("tools/call")
+        self.assertIsNotNone(entry)
+        query = '{"method":"equal","attribute":"name","values":["Zoë"]}'
+        received_queries = []
+
+        class TablesDbService:
+            def __init__(self, client):
+                pass
+
+            def list_rows(self, database_id, table_id, queries=None):
+                received_queries.append(queries)
+                return {"total": 0, "rows": []}
+
+        async def run_check():
+            ctx = Mock()
+            ctx.protocol_version = "2026-07-28"
+            ctx.meta = None
+            ctx.session.client_params = None
+            for arguments, expected in (
+                ({"queries": [query]}, [query]),
+                ({"queries": []}, []),
+                ({"queries": None}, None),
+                ({}, None),
+            ):
+                with self.subTest(arguments=arguments):
+                    params = types.CallToolRequestParams(
+                        name="appwrite_call_tool",
+                        arguments={
+                            "tool_name": "tables_db_list_rows",
+                            "arguments": {
+                                "database_id": "database-id",
+                                "table_id": "table-id",
+                                **arguments,
+                            },
+                        },
+                    )
+                    result = await entry.handler(ctx, params)
+
+                    self.assertFalse(result.is_error)
+                    self.assertEqual(received_queries[-1], expected)
+                    self.assertEqual(
+                        json.loads(result.content[0].text), {"total": 0, "rows": []}
+                    )
+
+        with patch.dict(server_module.SERVICE_CLASSES, {"tables_db": TablesDbService}):
+            asyncio.run(run_check())
+
     def test_format_tool_result_serializes_json(self):
         result = _format_tool_result(
             "tables_db_list_rows", {"total": 1, "rows": []}, {}