fix(cloud): raise AirbyteCloudError on malformed skill docs responses

airbyte/_direct_connectors/api_util.py

@@ -18,6 +18,7 @@
 
 from airbyte._direct_connectors.models import (
     ConnectorEnablement,
+    DirectAccessGuidance,
     ExternalApiExecuteResult,
     ExternalSearchResult,
     ExternalSearchStatusResult,
@@ -36,6 +37,7 @@
     AirbyteCloudApiError,
     AirbyteCloudError,
     AirbyteLibInputError,
+    AirbyteMalformedSkillDocsError,
 )
 from airbyte.registry import ConnectorType
 
@@ -350,21 +352,34 @@ def read_cloud_skill_docs(
     skill_id: str,
     credentials: _AirbyteCredentials,
     section: str | None = None,
-) -> dict[str, Any]:
+) -> DirectAccessGuidance:
     """Read skill docs for a workspace via the Cloud Config API.
 
     The `skill_id` is sent as the `id` query parameter, with `section` added only when
-    provided.
+    provided. A response that does not match `DirectAccessGuidance` raises
+    `AirbyteMalformedSkillDocsError`.
     """
     params: dict[str, Any] = {"id": skill_id}
     if section is not None:
         params["section"] = section
-    return make_cloud_agent_request(
+    path = f"/workspaces/{workspace_id}/skills/docs"
+    response = make_cloud_agent_request(
         method="GET",
-        path=f"/workspaces/{workspace_id}/skills/docs",
+        path=path,
         credentials=credentials,
         params=params,
     )
+    try:
+        return DirectAccessGuidance.model_validate(response)
+    except ValidationError as ex:
+        raise AirbyteMalformedSkillDocsError(
+            message="Malformed Airbyte Cloud skill docs response.",
+            skill_id=skill_id,
+            context={
+                "path": path,
+                "errors": ex.errors(include_url=False, include_input=False),
+            },
+        ) from ex
 
 
 class _ConnectorLookup(NamedTuple):

airbyte/cloud/connectors.py

@@ -759,12 +759,10 @@ def _context_layer_inspect(
             else connector_docs.destination_skill_id(self.connector_id)
         )
         try:
-            docs = DirectAccessGuidance.model_validate(
-                agents_api_util.read_cloud_skill_docs(
-                    workspace_id=self.workspace.workspace_id,
-                    skill_id=skill_id,
-                    credentials=self.workspace._credentials,  # noqa: SLF001
-                )
+            docs = agents_api_util.read_cloud_skill_docs(
+                workspace_id=self.workspace.workspace_id,
+                skill_id=skill_id,
+                credentials=self.workspace._credentials,  # noqa: SLF001
             )
         except exc.AirbyteCloudError as error:
             if not agents_api_util.is_not_enabled_error(error):
@@ -886,13 +884,11 @@ def read_agent_skill_docs(self, *, section: str | None = None) -> DirectAccessGu
                     "definition_id": self.definition_id,
                 },
             )
-        return DirectAccessGuidance.model_validate(
-            agents_api_util.read_cloud_skill_docs(
-                workspace_id=self.workspace.workspace_id,
-                skill_id=skill_id,
-                credentials=self.workspace._credentials,  # noqa: SLF001
-                section=section,
-            )
+        return agents_api_util.read_cloud_skill_docs(
+            workspace_id=self.workspace.workspace_id,
+            skill_id=skill_id,
+            credentials=self.workspace._credentials,  # noqa: SLF001
+            section=section,
         )
 
     def get_direct_access_guidance(self, *, section: str | None = None) -> DirectAccessGuidance:
@@ -931,13 +927,11 @@ def get_direct_access_guidance(self, *, section: str | None = None) -> DirectAcc
                 )
             skill_id = connector_docs.destination_skill_id(self.connector_id)
             try:
-                server_docs = DirectAccessGuidance.model_validate(
-                    agents_api_util.read_cloud_skill_docs(
-                        workspace_id=self.workspace.workspace_id,
-                        skill_id=skill_id,
-                        credentials=self.workspace._credentials,  # noqa: SLF001
-                        section=section,
-                    )
+                server_docs = agents_api_util.read_cloud_skill_docs(
+                    workspace_id=self.workspace.workspace_id,
+                    skill_id=skill_id,
+                    credentials=self.workspace._credentials,  # noqa: SLF001
+                    section=section,
                 )
             except exc.AirbyteCloudError as error:
                 if not agents_api_util.is_not_enabled_error(error) or section is not None:

airbyte/cloud/workspaces.py

@@ -629,13 +629,11 @@ def get_agent_skill_docs(
                 connector_docs.connector_id_from_skill_id(docs_skill_id)
             )
             return destination.read_agent_skill_docs(section=section)
-        return DirectAccessGuidance.model_validate(
-            agents_api_util.read_cloud_skill_docs(
-                workspace_id=self.workspace_id,
-                skill_id=docs_skill_id,
-                credentials=self._credentials,
-                section=section,
-            )
+        return agents_api_util.read_cloud_skill_docs(
+            workspace_id=self.workspace_id,
+            skill_id=docs_skill_id,
+            credentials=self._credentials,
+            section=section,
         )
 
     # Deploy sources and destinations

airbyte/exceptions.py

@@ -612,6 +612,13 @@ class AirbyteCloudApiError(AirbyteCloudError):
     status_code: int | None = None
 
 
+@dataclass
+class AirbyteMalformedSkillDocsError(AirbyteCloudError):
+    """Airbyte Cloud returned a skill docs response that does not match the expected shape."""
+
+    skill_id: str | None = None
+
+
 @dataclass
 class AirbyteExternalAccessNotEnabledError(AirbyteCloudError):
     """The connector is not enabled for external access, so it cannot execute direct actions."""

tests/unit_tests/test_cloud_credentials.py

@@ -40,6 +40,7 @@
 from airbyte.exceptions import (
     AirbyteCloudApiError,
     AirbyteCloudError,
+    AirbyteMalformedSkillDocsError,
     AirbyteMissingResourceError,
     AirbyteLibInputError,
 )
@@ -2412,6 +2413,31 @@ def test_cloud_connector_enablement_rejects_malformed_response(
     assert (exc_info.value.context or {})["path"] == "/sources/source-1/enablement"
 
 
+def test_read_cloud_skill_docs_rejects_malformed_response(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A skill-docs response missing `metadata` raises `AirbyteMalformedSkillDocsError`."""
+    monkeypatch.setattr(
+        cloud_workspaces.agents_api_util,
+        "make_cloud_agent_request",
+        lambda **_: {"content": [], "leaked": "payload-secret"},
+    )
+    credentials = cloud_credentials._AirbyteCredentials.from_auth(  # noqa: SLF001
+        bearer_token="token", env_vars=False
+    )
+
+    with pytest.raises(
+        AirbyteMalformedSkillDocsError, match="Malformed Airbyte Cloud skill docs"
+    ) as exc_info:
+        cloud_workspaces.agents_api_util.read_cloud_skill_docs(
+            workspace_id="workspace-1",
+            skill_id="connector-source:source-1",
+            credentials=credentials,
+        )
+    assert exc_info.value.skill_id == "connector-source:source-1"
+    assert "payload-secret" not in str(exc_info.value)
+
+
 def test_cloud_connector_enablement_parses_destination_response(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

tests/unit_tests/test_cloud_describe.py

@@ -255,9 +255,9 @@ def test_direct_access_guidance_id_source_with_context_layer(
     _patch_context_layer(monkeypatch)
     calls: list[dict[str, Any]] = []
 
-    def fake_read_docs(**kwargs: Any) -> dict[str, Any]:
+    def fake_read_docs(**kwargs: Any) -> DirectAccessGuidance:
         calls.append(kwargs)
-        return SKILL_DOCS_RESPONSE
+        return DirectAccessGuidance.model_validate(SKILL_DOCS_RESPONSE)
 
     monkeypatch.setattr(
         agents_api_util,
@@ -438,7 +438,9 @@ def test_get_direct_access_guidance_destination_forwards_section_to_sonar(
         "section_id": "streams",
         "content": [{"type": "paragraph", "text": "Streams from Sonar."}],
     }
-    read_docs = MagicMock(return_value=section_response)
+    read_docs = MagicMock(
+        return_value=DirectAccessGuidance.model_validate(section_response)
+    )
     monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
     monkeypatch.setattr(CloudWorkspace, "list_connections", lambda *_, **__: [])
     destination = _seed_destination(
@@ -485,7 +487,11 @@ def test_get_direct_access_guidance_destination_merges_server_docs(
     """A successful docs read gets a short local intro; the outline is unchanged."""
     workspace = _make_workspace(monkeypatch)
     _patch_context_layer(monkeypatch)
-    read_docs = MagicMock(return_value=DESTINATION_SKILL_DOCS_RESPONSE)
+    read_docs = MagicMock(
+        return_value=DirectAccessGuidance.model_validate(
+            DESTINATION_SKILL_DOCS_RESPONSE
+        )
+    )
     monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
     monkeypatch.setattr(CloudWorkspace, "list_connections", lambda *_, **__: [])
     destination = _seed_destination(

tests/unit_tests/test_cloud_guidance.py

@@ -105,9 +105,9 @@ def test_get_agent_skill_docs_reads_docs(
     workspace = _make_workspace(monkeypatch)
     calls: list[dict[str, Any]] = []
 
-    def fake_read_docs(**kwargs: Any) -> dict[str, Any]:
+    def fake_read_docs(**kwargs: Any) -> DirectAccessGuidance:
         calls.append(kwargs)
-        return SKILL_DOCS_RESPONSE
+        return DirectAccessGuidance.model_validate(SKILL_DOCS_RESPONSE)
 
     monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", fake_read_docs)
 
@@ -127,9 +127,9 @@ def test_get_agent_skill_docs_passes_section(
     workspace = _make_workspace(monkeypatch)
     calls: list[dict[str, Any]] = []
 
-    def fake_read_docs(**kwargs: Any) -> dict[str, Any]:
+    def fake_read_docs(**kwargs: Any) -> DirectAccessGuidance:
         calls.append(kwargs)
-        return SKILL_DOCS_RESPONSE
+        return DirectAccessGuidance.model_validate(SKILL_DOCS_RESPONSE)
 
     monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", fake_read_docs)