fix: tolerate non-conforming apps across all app read paths (#48)

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)")
@@ -174,14 +239,34 @@ async def get_application(ctx: Context, app_id: str, expand: Optional[str] = Non
         if expand:
             query_params["expand"] = expand
 
-        app, _, err = await client.get_application(app_id, **query_params)
+        # The typed client.get_application validates the record into a strict SDK
+        # model, so an App Catalog SAML app with a partial settings.signOn (or a
+        # custom SWA whose name is outside the template enum) raises and the call
+        # fails outright. Fetch the raw record through the request executor and
+        # parse it via _safe_parse_app, falling back to the raw dict on failure.
+        executor = client.get_request_executor()
+        query_string = urlencode(
+            {_camel_case_param(k): v for k, v in query_params.items()}
+        )
+        url = f"/api/v1/apps/{app_id}" + (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:
+            logger.error(f"Okta API error while getting application {app_id}: {request_err}")
+            return {"error": str(request_err)}
 
-        if err:
-            logger.error(f"Okta API error while getting application {app_id}: {err}")
-            return {"error": str(err)}
+        _, response_body, response_err = await executor.execute(request)
+        if response_err:
+            logger.error(f"Okta API error while getting application {app_id}: {response_err}")
+            return {"error": str(response_err)}
+
+        item = json.loads(response_body) if response_body else None
+        if item is None:
+            return {"error": f"Application {app_id} not found"}
 
         logger.info(f"Successfully retrieved application: {app_id}")
-        return app
+        return _safe_parse_app(item)
     except Exception as e:
         logger.error(f"Exception while getting application {app_id}: {type(e).__name__}: {e}")
         return {"error": str(e)}

src/okta_mcp_server/tools/groups/groups.py

@@ -5,12 +5,15 @@
 # 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 Optional
+from urllib.parse import urlencode
 
 from loguru import logger
 from mcp.server.fastmcp import Context
 
 from okta_mcp_server.server import mcp
+from okta_mcp_server.tools.applications.applications import _camel_case_param, _safe_parse_app
 from okta_mcp_server.utils.client import get_okta_client
 from okta_mcp_server.utils.elicitation import DeleteConfirmation, elicit_or_fallback
 from okta_mcp_server.utils.messages import DELETE_GROUP
@@ -484,7 +487,39 @@ async def list_group_apps(
         query_params = build_query_params(after=after, limit=effective_limit)
         logger.debug(f"Calling Okta API to list applications for group {group_id}")
 
-        apps, response, err = await client.list_assigned_applications_for_group(group_id, **query_params)
+        async def _fetch_group_apps_page(params):
+            """Fetch one page of /api/v1/groups/{id}/apps and parse each app permissively.
+
+            The typed ``client.list_assigned_applications_for_group`` validates the whole
+            page into strict SDK models in one pass, so a single non-conforming app — a
+            SAML app with a partial ``settings.signOn`` or a custom SWA whose ``name`` is
+            outside the SDK enum — aborts the entire response. Fetching the raw page
+            through the request executor and parsing per item via ``_safe_parse_app``
+            avoids that. Mirrors the resilient path in ``list_applications``.
+            """
+            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 = f"/api/v1/groups/{group_id}/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
+
+        apps, response, err = await _fetch_group_apps_page(query_params)
 
         if err:
             logger.error(f"Okta API error while listing applications for group {group_id}: {err}")
@@ -501,7 +536,7 @@ async def list_group_apps(
             async def _next_page(cursor):
                 p = {k: v for k, v in query_params.items() if k != "after"}
                 p["after"] = cursor
-                return await client.list_assigned_applications_for_group(group_id, **p)
+                return await _fetch_group_apps_page(p)
 
             async def _on_page(pages, total):
                 logger.info(f"[list_group_apps] Page {pages} fetched — {total} apps so far")

src/okta_mcp_server/utils/pagination.py

@@ -25,15 +25,53 @@ def extract_after_cursor(response) -> Optional[str]:
     Returns:
         str: The 'after' cursor value, or None if no next page
     """
-    # --- Okta SDK v3: ApiResponse with Link header ---
-    if response and hasattr(response, "headers") and response.headers:
+    # --- Raw aiohttp response (returned by the request executor) ---
+    # The request executor returns aiohttp's response object directly. aiohttp
+    # pre-parses the (possibly multiple) Link headers into ``.links``, a mapping
+    # keyed by rel — e.g. ``links["next"]["url"]`` is a yarl URL. This is what the
+    # SDK itself uses (OktaAPIResponse.extract_pagination). Reading the raw
+    # ``headers.get("Link")`` is NOT enough: Okta sends ``self`` and ``next`` as
+    # SEPARATE Link headers, and a multidict ``.get`` returns only the first
+    # (self), so the next cursor would be missed.
+    links = getattr(response, "links", None) if response is not None else None
+    if links:
+        try:
+            nxt = links.get("next")
+            url = nxt.get("url") if nxt is not None and hasattr(nxt, "get") else None
+            if url is not None:
+                cursor = parse_qs(urlparse(str(url)).query).get("after", [None])[0]
+                if cursor:
+                    return cursor
+        except Exception as e:
+            logger.warning(f"Failed to parse aiohttp links cursor: {e}")
+
+    # --- Okta SDK v3: Link-header cursor ---
+    # Resolve a headers mapping from whichever response shape we got:
+    #   * ApiResponse exposes a ``.headers`` attribute
+    #   * OktaAPIResponse (what the request executor returns) exposes headers via
+    #     ``get_headers()`` / ``_resp_headers`` and does NOT have ``.headers``
+    # Reading both ensures the cursor is found regardless of how the page was
+    # fetched (typed client vs. raw request executor).
+    headers = None
+    if response is not None:
+        if getattr(response, "headers", None):
+            headers = response.headers
+        elif hasattr(response, "get_headers"):
+            try:
+                headers = response.get_headers()
+            except Exception:
+                headers = None
+        if not headers and getattr(response, "_resp_headers", None):
+            headers = response._resp_headers
+
+    if headers:
         link_header = ""
         try:
-            link_header = response.headers.get("Link", "") or response.headers.get("link", "")
+            link_header = headers.get("Link", "") or headers.get("link", "")
         except Exception:
-            for key in response.headers:
+            for key in headers:
                 if key.lower() == "link":
-                    link_header = response.headers[key]
+                    link_header = headers[key]
                     break
 
         if link_header and 'rel="next"' in link_header:

tests/test_get_application.py

@@ -0,0 +1,102 @@
+# 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 get_application — resilient parsing of a single app record (#48)."""
+
+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 get_application
+
+
+# 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"}},
+}
+
+# An App Catalog SAML app with a partial settings.signOn fails strict SDK
+# validation (the model marks ~15 signOn fields required) — the record that
+# previously made get_application fail outright (#48).
+BAD_SPARSE_SAML = {
+    "id": "0oaSAML0001",
+    "label": "Sparse SAML",
+    "name": "sparsesaml",
+    "signOnMode": "SAML_2_0",
+    "settings": {"signOn": {"defaultRelayState": ""}},
+}
+
+
+def _make_ctx():
+    manager = MagicMock()
+    ctx = MagicMock()
+    ctx.request_context.lifespan_context.okta_auth_manager = manager
+    return ctx
+
+
+def _client_returning(body, execute_error=None):
+    """Build a fake Okta client whose request executor returns the given record."""
+    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=(MagicMock(), body, None))
+    client = MagicMock()
+    client.get_request_executor = MagicMock(return_value=executor)
+    return client
+
+
+class TestGetApplicationResilientParsing:
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_good_app_returns_typed_model(self, mock_get_client):
+        mock_get_client.return_value = _client_returning(json.dumps(GOOD_BOOKMARK))
+
+        result = await get_application(_make_ctx(), "0oaBOOKMARK001")
+
+        assert isinstance(result, okta_models.BookmarkApplication)
+
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_non_conforming_app_falls_back_to_raw_dict(self, mock_get_client):
+        mock_get_client.return_value = _client_returning(json.dumps(BAD_SPARSE_SAML))
+
+        result = await get_application(_make_ctx(), "0oaSAML0001")
+
+        assert isinstance(result, dict)
+        assert result["id"] == "0oaSAML0001"
+        assert "_deserialization_warning" in result
+
+    @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: 404 not found")
+
+        result = await get_application(_make_ctx(), "0oaMISSING")
+
+        assert result == {"error": "Error: 404 not found"}
+
+    @pytest.mark.asyncio
+    @patch("okta_mcp_server.tools.applications.applications.get_okta_client")
+    async def test_expand_is_sent_as_query_param(self, mock_get_client):
+        client = _client_returning(json.dumps(GOOD_BOOKMARK))
+        mock_get_client.return_value = client
+
+        await get_application(_make_ctx(), "0oaBOOKMARK001", expand="user/abc")
+
+        url = client.get_request_executor.return_value.create_request.call_args.kwargs["url"]
+        assert "/api/v1/apps/0oaBOOKMARK001" in url
+        assert "expand=user" in url