fix: tolerate non-conforming apps when listing applications

src/okta_mcp_server/tools/applications/applications.py

@@ -5,7 +5,9 @@
 # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and limitations under the License.
 
+import json
 from typing import Any, Dict, Optional
+from urllib.parse import urlencode
 
 import okta.models as okta_models
 from loguru import logger
@@ -40,6 +42,39 @@ def _build_application_model(app_config: Dict[str, Any]) -> Any:
     return model_cls(**app_config)
 
 
+def _camel_case_param(name: str) -> str:
+    """Convert a snake_case query-param name to the camelCase Okta expects.
+
+    build_query_params keeps tool argument names verbatim (e.g. include_non_deleted),
+    but the Okta API expects camelCase query keys (includeNonDeleted). The typed SDK
+    client performs this mapping internally; since the listing path now issues the
+    request directly, it must do the same. Names without underscores are unchanged.
+    """
+    head, *rest = name.split("_")
+    return head + "".join(part.capitalize() for part in rest)
+
+
+def _safe_parse_app(item: Dict[str, Any]) -> Any:
+    """Deserialize a single application dict, falling back to the raw dict.
+
+    The Okta SDK bulk-deserializes list responses into strict pydantic models and
+    aborts the entire page if any single record fails validation — for example a
+    SAML app whose ``settings.signOn`` omits fields the model marks required, or a
+    provisioning ``features`` value outside the SDK's enum. Parsing each record on
+    its own keeps one non-conforming app from breaking the whole listing; records
+    that fail strict parsing are returned as their raw dict with a warning marker.
+    """
+    try:
+        model = okta_models.Application.from_dict(item)
+        return model if model is not None else item
+    except Exception as e:
+        label = item.get("label") or item.get("name") or item.get("id", "<unknown>")
+        logger.warning(
+            f"Application '{label}' failed strict deserialization, returning raw dict: {type(e).__name__}: {e}"
+        )
+        return {**item, "_deserialization_warning": f"{type(e).__name__}: {e}"}
+
+
 from okta_mcp_server.utils.client import get_okta_client
 from okta_mcp_server.utils.elicitation import DeactivateConfirmation, DeleteConfirmation, elicit_or_fallback
 from okta_mcp_server.utils.messages import DEACTIVATE_APPLICATION, DELETE_APPLICATION
@@ -108,8 +143,38 @@ async def list_applications(
             include_non_deleted=include_non_deleted,
         )
 
+        async def _fetch_apps_page(params):
+            """Fetch one page of /api/v1/apps and parse each app permissively.
+
+            The typed ``client.list_applications`` validates the whole page into
+            strict SDK models in one pass, so a single non-conforming app aborts
+            the entire response. Fetching the raw page through the request
+            executor and parsing per item via ``_safe_parse_app`` avoids that.
+            """
+            executor = client.get_request_executor()
+            query_string = urlencode(
+                {
+                    _camel_case_param(k): ("true" if v is True else "false" if v is False else v)
+                    for k, v in params.items()
+                }
+            )
+            url = "/api/v1/apps" + (f"?{query_string}" if query_string else "")
+            request, request_err = await executor.create_request(
+                method="GET", url=url, body={}, headers={}, oauth=False
+            )
+            if request_err:
+                return None, None, request_err
+
+            page_response, response_body, response_err = await executor.execute(request)
+            if response_err:
+                return None, page_response, response_err
+
+            raw_items = json.loads(response_body) if response_body else []
+            parsed_items = [_safe_parse_app(item) for item in raw_items]
+            return parsed_items, page_response, None
+
         logger.debug("Calling Okta API to list applications")
-        apps, response, err = await client.list_applications(**query_params)
+        apps, response, err = await _fetch_apps_page(query_params)
 
         if err:
             logger.error(f"Okta API error while listing applications: {err}")
@@ -129,7 +194,7 @@ async def list_applications(
             async def _next_page(cursor):
                 p = dict(query_params)
                 p["after"] = cursor
-                return await client.list_applications(**p)
+                return await _fetch_apps_page(p)
 
             async def _on_page(pages, total):
                 await ctx.info(f"Fetching applications... {total} fetched so far ({pages} pages)")

tests/test_list_applications.py

@@ -0,0 +1,136 @@
+# The Okta software accompanied by this notice is provided pursuant to the following terms:
+# Copyright © 2026-Present, Okta, Inc.
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0.
+# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and limitations under the License.
+
+"""Tests for list_applications — resilient per-item parsing of the apps page."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import okta.models as okta_models
+import pytest
+
+from okta_mcp_server.tools.applications.applications import _safe_parse_app, list_applications
+
+
+# A bookmark app parses cleanly into a typed model.
+GOOD_BOOKMARK = {
+    "id": "0oaBOOKMARK001",
+    "label": "Bookmark App",
+    "name": "bookmark",
+    "signOnMode": "BOOKMARK",
+    "settings": {"app": {"url": "https://example.com"}},
+}
+
+# A SAML app with a partial settings.signOn fails strict SDK validation
+# (the model marks ~15 signOn fields required). This is the record that
+# previously aborted the whole listing.
+BAD_SPARSE_SAML = {
+    "id": "0oaSAML0001",
+    "label": "Sparse SAML",
+    "name": "sparsesaml",
+    "signOnMode": "SAML_2_0",
+    "settings": {"signOn": {"defaultRelayState": ""}},
+}
+
+
+class _Resp:
+    """Minimal stand-in for the executor response (only headers are read)."""
+
+    def __init__(self, headers=None):
+        self.headers = headers or {}
+
+
+def _make_ctx():
+    from tests.conftest import FakeLifespanContext, FakeOktaAuthManager
+
+    request_context = MagicMock()
+    request_context.lifespan_context = FakeLifespanContext(
+        okta_auth_manager=FakeOktaAuthManager()
+    )
+    ctx = MagicMock()
+    ctx.request_context = request_context
+    return ctx
+
+
+def _client_returning(body, response=None, execute_error=None):
+    """Build a fake Okta client whose request executor returns the given page body."""
+    executor = MagicMock()
+    executor.create_request = AsyncMock(return_value=({"method": "GET"}, None))
+    if execute_error is not None:
+        executor.execute = AsyncMock(return_value=(None, None, execute_error))
+    else:
+        executor.execute = AsyncMock(return_value=(response or _Resp(), body, None))
+    client = MagicMock()
+    client.get_request_executor = MagicMock(return_value=executor)
+    return client
+
+
+class TestSafeParseApp:
+    def test_good_record_parses_to_model(self):
+        result = _safe_parse_app(GOOD_BOOKMARK)
+        assert isinstance(result, okta_models.BookmarkApplication)
+
+    def test_bad_record_falls_back_to_raw_dict(self):
+        result = _safe_parse_app(BAD_SPARSE_SAML)
+        assert isinstance(result, dict)
+        assert result["id"] == "0oaSAML0001"
+        assert "_deserialization_warning" in result
+
+
+class TestListApplicationsResilientParsing:
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_one_bad_app_does_not_abort_the_listing(self, mock_get_client):
+        body = json.dumps([GOOD_BOOKMARK, BAD_SPARSE_SAML])
+        mock_get_client.return_value = _client_returning(body)
+
+        result = await list_applications(ctx=_make_ctx())
+
+        assert result["total_fetched"] == 2
+        items = result["items"]
+
+        bad = [i for i in items if isinstance(i, dict) and "_deserialization_warning" in i]
+        assert len(bad) == 1
+        assert bad[0]["id"] == "0oaSAML0001"
+
+        good = [i for i in items if not isinstance(i, dict)]
+        assert len(good) == 1
+        assert isinstance(good[0], okta_models.BookmarkApplication)
+
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_executor_error_is_returned(self, mock_get_client):
+        mock_get_client.return_value = _client_returning(None, execute_error="Error: 403 forbidden")
+
+        result = await list_applications(ctx=_make_ctx())
+
+        assert result == {"error": "Error: 403 forbidden"}
+
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_empty_page_returns_empty_envelope(self, mock_get_client):
+        mock_get_client.return_value = _client_returning("[]")
+
+        result = await list_applications(ctx=_make_ctx())
+
+        assert result["total_fetched"] == 0
+        assert result["items"] == []
+
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_snake_case_params_are_sent_as_camel_case(self, mock_get_client):
+        client = _client_returning("[]")
+        mock_get_client.return_value = client
+
+        await list_applications(ctx=_make_ctx(), include_non_deleted=True)
+
+        executor = client.get_request_executor.return_value
+        url = executor.create_request.call_args.kwargs["url"]
+        assert "includeNonDeleted=true" in url
+        assert "include_non_deleted" not in url