From 1b5ff3a135028930c2656367d7346fe7b00438be Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Thu, 24 Sep 2026 13:53:27 -0700
Subject: [PATCH 01/10] fix(cloud): clarify unsupported writes and resilient
 connector guidance

---
 airbyte/cloud/connectors.py             |  55 +++++----
 airbyte/mcp/cloud.py                    |  25 ++--
 tests/unit_tests/test_cloud_describe.py | 144 ++++++++++++++++++++++++
 tests/unit_tests/test_mcp_cloud.py      |  50 +++++++-
 4 files changed, 241 insertions(+), 33 deletions(-)

diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py
index 0c66db1d7..3ba979ee3 100644
--- a/airbyte/cloud/connectors.py
+++ b/airbyte/cloud/connectors.py
@@ -164,17 +164,29 @@ def __init__(
         self._connector_definition: _ConnectorDefinitionLike | None = None
         """The connector definition lookup result. (Cached; `None` until fetched.)"""
 
-    def _get_enabled_features(self) -> frozenset[ConnectorFeature]:
-        """Return the enabled features, resolving them through the workspace on first use."""
-        if self._enabled_features is None:
-            self._enabled_features = self.workspace._get_connector_features(self)  # noqa: SLF001
+    def get_enabled_features(
+        self, *, warnings: list[str] | None = None
+    ) -> frozenset[ConnectorFeature]:
+        """Resolve connector features, optionally collecting unavailable-docs warnings.
+
+        Successful features are cached. Unavailable docs (403/404) return no features
+        without caching absence; other probe failures propagate so callers can distinguish
+        unknown support from an empty feature set.
+        """
+        if self._enabled_features is not None:
+            if warnings is not None and self._context_layer_details is not None:
+                warnings.extend(self._context_layer_details.warnings)
+            return self._enabled_features
 
-        return self._enabled_features
+        features = self.workspace._get_connector_features(self, warnings=warnings)  # noqa: SLF001
+        if features:
+            self._enabled_features = features
+        return features
 
     @property
     def enabled_features(self) -> frozenset[ConnectorFeature]:
-        """The features enabled for this connector. Resolved on first access and cached."""
-        return self._get_enabled_features()
+        """The connector's enabled features; successful lookups are cached."""
+        return self.get_enabled_features()
 
     def is_feature_enabled(self, feature: ConnectorFeature) -> bool:
         """Whether `feature` is enabled for this connector.
@@ -421,11 +433,10 @@ def execute_api_action(
         skip_truncation: bool = True,
         intent: str | None = None,
     ) -> ExternalApiExecuteResult:
-        """Run a write action (`create`, `update`, or `delete`) on one entity type.
+        """Reject unsupported Cloud writes (`create`, `update`, or `delete`).
 
-        Requires external access to be enabled for this connector in its organization's
-        Context Layer settings. Connectors without an entity API fail with the Agents
-        API's own error.
+        Cloud connector execution currently supports read actions only. This method
+        raises `PyAirbyteInputError` without sending a request.
         """
         try:
             resolved_action = ExternalApiWriteAction(action)
@@ -436,15 +447,11 @@ def execute_api_action(
                 context={"entity_type": entity_type, "action": action},
             ) from None
 
-        return self._execute_direct_action(
-            entity_type=entity_type,
-            action=resolved_action.value,
-            api_args=api_args,
-            select_fields=select_fields,
-            exclude_fields=exclude_fields,
-            skip_truncation=skip_truncation,
-            intent=intent,
-            read_only=False,
+        _ = api_args, select_fields, exclude_fields, skip_truncation, intent
+        raise exc.PyAirbyteInputError(
+            message="Cloud connector write actions are not supported yet.",
+            guidance="Use execute_api_query for read actions (list, get, search).",
+            context={"entity_type": entity_type, "action": resolved_action.value},
         )
 
     def execute_sql_query(
@@ -742,11 +749,12 @@ def _context_layer_inspect(
 
         The connector's docs skill is probed: a skill doc exists only for connectors the
         Context layer knows. A 403 or 404 `AirbyteError` from the docs read means the
-        connector is not enabled for agent access, so a warning is appended to
+        docs are unavailable, so a warning is appended to
         `warnings` and `None` is returned. Any other failure (auth, server, malformed
         response, transport) is raised to the caller.
         """
         if self._context_layer_details is not None and not force_refresh:
+            warnings.extend(self._context_layer_details.warnings)
             return self._context_layer_details
         skill_id = (
             connector_docs.source_skill_id(self.connector_id)
@@ -764,7 +772,9 @@ def _context_layer_inspect(
         except exc.AirbyteError as error:
             if not agents_api_util.is_not_enabled_error(error):
                 raise
-            warnings.append(f"Connector direct-access docs lookup failed: {error}")
+            warnings.append(
+                "Connector direct-access docs are unavailable (access denied or not found)."
+            )
             return None
         parsed = _DirectConnectorInspectResult(
             connector_id=self.connector_id,
@@ -778,6 +788,7 @@ def _context_layer_inspect(
             warnings=list(docs.metadata.warnings),
         )
         self._context_layer_details = parsed
+        warnings.extend(parsed.warnings)
         return parsed
 
     def _fetch_enablement(self) -> ConnectorEnablement | None:
diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index 85a1c5eeb..e78121d6f 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -1348,6 +1348,14 @@ def _describe_cloud_connector(
         warnings.append(f"Integration name lookup failed: {error}")
         integration_name = None
 
+    try:
+        enabled_features: list[ConnectorFeature] | Literal["unknown"] = sorted(
+            connector.get_enabled_features(warnings=warnings)
+        )
+    except (AirbyteError, requests.RequestException, ValueError):
+        enabled_features = "unknown"
+        warnings.append("Connector feature lookup failed; enabled features are unknown.")
+
     result = CloudConnectorDetailsResult(
         connector_id=connector.connector_id,
         connector_type=connector_type.value,
@@ -1355,6 +1363,7 @@ def _describe_cloud_connector(
         connector_url=connector.connector_url,
         connector_definition_id=connector.definition_id,
         integration_name=integration_name,
+        enabled_features=enabled_features,
     )
 
     try:
@@ -1390,11 +1399,11 @@ def _describe_cloud_connector(
         except AirbyteError as error:
             warnings.append(f"Connection listing failed: {error}")
 
-    if with_direct_access_guidance:
+    if with_direct_access_guidance and enabled_features != "unknown":
         try:
             docs = connector.get_direct_access_guidance()
-        except (PyAirbyteError, requests.RequestException) as error:
-            warnings.append(f"Direct access docs are unavailable: {error}")
+        except (PyAirbyteError, requests.RequestException, ValueError):
+            warnings.append("Direct access docs are unavailable.")
         else:
             result.direct_access_guidance = render_connector_docs_result(docs)
 
@@ -1729,14 +1738,10 @@ def _execute_external_api_action(  # noqa: PLR0913  # Explicit args mirror the c
         ),
     ] = None,
 ) -> ExternalApiExecuteResult:
-    """Run a write action through a deployed Cloud connector's direct API.
-
-    Creates, updates, or deletes data in the external system.
+    """Cloud connector write actions are not supported yet.
 
-    Before calling, read the connector's action docs with `get_agent_skill_docs`
-    (or `describe_cloud_connector` with `with_direct_access_guidance=True`) to
-    learn the entity types, actions, and required `api_args`; argument names
-    differ per connector and are not guessable.
+    Returns a clear error without executing a write. Use `execute_external_api_query`
+    for supported read actions (`list`, `get`, or `search`).
     """
     connector = _get_cloud_workspace(ctx, workspace_id).get_connector(connector_id)
     return connector.execute_api_action(
diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py
index 4c7d4711b..500958621 100644
--- a/tests/unit_tests/test_cloud_describe.py
+++ b/tests/unit_tests/test_cloud_describe.py
@@ -14,6 +14,7 @@
 from airbyte._direct_connectors import api_util as agents_api_util
 from airbyte._direct_connectors import connector_docs
 from airbyte.cloud import workspaces as cloud_workspaces
+from airbyte.mcp import cloud as cloud_mcp
 from airbyte.cloud.connections import CloudConnection
 from airbyte.cloud.connectors import (
     CloudConnector,
@@ -807,3 +808,146 @@ def test_is_feature_enabled_wrong_connector_kind_short_circuits(
 
     assert connector.is_feature_enabled(feature) is False
     get_features.assert_not_called()
+
+
+@pytest.mark.parametrize("status", [403, 404])
+def test_unavailable_features_warn_without_caching_absence(
+    monkeypatch: pytest.MonkeyPatch, status: int
+) -> None:
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    read_docs = MagicMock(
+        side_effect=[
+            AirbyteError(context={"status_code": status, "response_text": "secret"}),
+            SKILL_DOCS_RESPONSE,
+        ]
+    )
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+    source = _seed_source(workspace, "source-1", "GitHub")
+    warnings: list[str] = []
+
+    assert source.get_enabled_features(warnings=warnings) == frozenset()
+    assert source._enabled_features is None  # noqa: SLF001
+    assert len(warnings) == 1
+    assert "unavailable" in warnings[0]
+    assert "secret" not in warnings[0]
+    assert "not enabled" not in warnings[0]
+    assert ConnectorFeature.DIRECT_API_QUERY in source.enabled_features
+    assert read_docs.call_count == 2
+    assert source.enabled_features
+    assert read_docs.call_count == 2
+
+
+@pytest.mark.parametrize(
+    "failure",
+    [
+        AirbyteError(context={"status_code": 401, "response_text": "secret"}),
+        AirbyteError(context={"status_code": 500, "response_text": "secret"}),
+        requests.Timeout("secret"),
+        ValueError("secret"),
+    ],
+)
+def test_describe_survives_unknown_features_and_recovers(
+    monkeypatch: pytest.MonkeyPatch, failure: Exception
+) -> None:
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    read_docs = MagicMock(side_effect=[failure, SKILL_DOCS_RESPONSE])
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+    source = _seed_source(workspace, "source-1", "GitHub")
+    source._connector_definition = SimpleNamespace(name="GitHub")  # noqa: SLF001
+    monkeypatch.setattr(connector_docs, "build_connection_details", lambda _: [])
+
+    result = cloud_mcp._describe_cloud_connector(  # noqa: SLF001
+        source,
+        with_config=False,
+        with_replication_details=True,
+        with_direct_access_guidance=True,
+        with_data_replication_docs=False,
+    )
+
+    assert result.connector_id == "source-1"
+    assert result.replication_details == []
+    assert result.enabled_features == "unknown"
+    assert result.warnings == [
+        "Connector feature lookup failed; enabled features are unknown."
+    ]
+    assert read_docs.call_count == 1
+    assert source._enabled_features is None  # noqa: SLF001
+    assert ConnectorFeature.DIRECT_API_QUERY in source.enabled_features
+    assert read_docs.call_count == 2
+
+
+def test_malformed_probe_is_not_cached_or_silently_filtered(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    source = _seed_source(workspace, "source-1", "GitHub")
+    monkeypatch.setattr(workspace, "list_sources", lambda: [source])
+    monkeypatch.setattr(workspace, "list_destinations", lambda: [])
+    read_docs = MagicMock(side_effect=[{}, SKILL_DOCS_RESPONSE])
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+
+    with pytest.raises(ValueError):
+        workspace.list_connectors(feature_filter=ConnectorFeature.DIRECT_ACCESS)
+    assert source._enabled_features is None  # noqa: SLF001
+    assert workspace.list_connectors(feature_filter=ConnectorFeature.DIRECT_ACCESS) == [
+        source
+    ]
+    assert read_docs.call_count == 2
+
+
+@pytest.mark.parametrize("section", [None, "overview", "sql-passthrough"])
+def test_destination_server_guidance_is_not_duplicated(
+    monkeypatch: pytest.MonkeyPatch, section: str | None
+) -> None:
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    server_section = (
+        "actions.record.sql_select" if section == "sql-passthrough" else section
+    )
+    server_response = {
+        "metadata": {
+            "id": "connector-destination:dest-1",
+            "kind": "connector_destination",
+            "title": "Warehouse",
+        },
+        "outline": [
+            {
+                "id": "actions.record.sql_select",
+                "title": "SQL select",
+                "available": True,
+            }
+        ],
+        "section_id": server_section,
+        "content": [{"type": "paragraph", "text": "Authoritative SQL instructions."}],
+    }
+    read_docs = MagicMock(return_value=server_response)
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+    destination = _seed_destination(workspace, "dest-1", SNOWFLAKE_DEFINITION_ID)
+    # A successful server docs response must not trigger local configuration/connection reads.
+    monkeypatch.setattr(
+        connector_docs,
+        "_destination_load_context",
+        MagicMock(side_effect=AssertionError("local SQL rendering")),
+    )
+    monkeypatch.setattr(
+        workspace,
+        "list_connections",
+        MagicMock(side_effect=AssertionError("local connection listing")),
+    )
+
+    docs = destination.get_direct_access_guidance(section=section)
+
+    assert read_docs.call_count == 1
+    assert read_docs.call_args.kwargs["section"] == server_section
+    assert docs.content == [
+        {"type": "paragraph", "text": "Authoritative SQL instructions."}
+    ]
+    assert docs.section_id == section
+    assert {entry.id for entry in docs.outline} == {
+        "actions.record.sql_select",
+        "connections",
+        "streams",
+    }
diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py
index 0725545ca..9bcf0f5e0 100644
--- a/tests/unit_tests/test_mcp_cloud.py
+++ b/tests/unit_tests/test_mcp_cloud.py
@@ -27,7 +27,17 @@
     ExternalSearchType,
     _SQL_PASSTHROUGH_DESTINATION_DIALECTS,
 )
-from airbyte.cloud.connectors import CheckResult, ConnectorFeature, ConnectorType
+from airbyte.cloud.connectors import (
+    CheckResult,
+    CloudSource,
+    ConnectorFeature,
+    ConnectorType,
+)
+from airbyte.cloud.workspaces import CloudWorkspace
+from airbyte.mcp._error_handling import (
+    MCP_TOOL_USER_FACING_ERRORS,
+    format_user_facing_error,
+)
 from airbyte.cloud.models import (
     CloudDefaultContextInfo,
     CloudOrganizationInfo,
@@ -1753,6 +1763,15 @@ def integration_name(self) -> str:
             raise self._integration_name
         return self._integration_name
 
+    def get_enabled_features(
+        self, *, warnings: list[str]
+    ) -> frozenset[ConnectorFeature]:
+        if self.workspace._has_context_layer_api():
+            details = self._context_layer_inspect(warnings=warnings)
+            if details is not None:
+                warnings.extend(details.warnings)
+        return self.enabled_features
+
     def is_feature_enabled(self, feature: ConnectorFeature) -> bool:
         return feature in self.enabled_features
 
@@ -2259,3 +2278,32 @@ def enabled_features(self, value: frozenset[ConnectorFeature]) -> None:
 
     assert result.enabled_features == cloud_mcp.FEATURES_UNKNOWN
     assert any("enabled features are unknown" in w for w in result.warnings)
+
+
+@pytest.mark.parametrize("action", list(ExternalApiWriteAction))
+def test_cloud_write_tool_reports_unsupported_without_network(
+    monkeypatch: pytest.MonkeyPatch, action: ExternalApiWriteAction
+) -> None:
+    workspace = CloudWorkspace(workspace_id="workspace-1", bearer_token="token")
+    source = CloudSource(workspace=workspace, connector_id="source-1")
+    monkeypatch.setattr(workspace, "get_connector", lambda _: source)
+    monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_: workspace)
+    request = MagicMock(side_effect=AssertionError("unsupported write sent to server"))
+    monkeypatch.setattr(requests, "request", request)
+
+    with pytest.raises(PyAirbyteInputError) as raised:
+        cloud_mcp.execute_external_api_action(
+            None,
+            connector_id="source-1",
+            entity_type="issues",
+            action=action,
+            api_args={"title": "example"},
+            workspace_id="workspace-1",
+        )
+
+    request.assert_not_called()
+    assert isinstance(raised.value, MCP_TOOL_USER_FACING_ERRORS)
+    assert format_user_facing_error(raised.value) == (
+        "Cloud connector write actions are not supported yet. "
+        "Use execute_api_query for read actions (list, get, search)."
+    )

From 88e8167dc3477bd47c36f3b8045869060cec8020 Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Thu, 24 Sep 2026 13:57:22 -0700
Subject: [PATCH 02/10] fix(cloud): name MCP read tool in unsupported-write
 guidance

---
 airbyte/cloud/connectors.py        | 4 +++-
 tests/unit_tests/test_mcp_cloud.py | 2 +-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py
index 3ba979ee3..d2dcfd453 100644
--- a/airbyte/cloud/connectors.py
+++ b/airbyte/cloud/connectors.py
@@ -450,7 +450,9 @@ def execute_api_action(
         _ = api_args, select_fields, exclude_fields, skip_truncation, intent
         raise exc.PyAirbyteInputError(
             message="Cloud connector write actions are not supported yet.",
-            guidance="Use execute_api_query for read actions (list, get, search).",
+            guidance=(
+                "Use execute_api_query (Python) or execute_external_api_query (MCP) for reads."
+            ),
             context={"entity_type": entity_type, "action": resolved_action.value},
         )
 
diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py
index 9bcf0f5e0..6c4916d57 100644
--- a/tests/unit_tests/test_mcp_cloud.py
+++ b/tests/unit_tests/test_mcp_cloud.py
@@ -2305,5 +2305,5 @@ def test_cloud_write_tool_reports_unsupported_without_network(
     assert isinstance(raised.value, MCP_TOOL_USER_FACING_ERRORS)
     assert format_user_facing_error(raised.value) == (
         "Cloud connector write actions are not supported yet. "
-        "Use execute_api_query for read actions (list, get, search)."
+        "Use execute_api_query (Python) or execute_external_api_query (MCP) for reads."
     )

From e0bf622e952e33e1867dc2d6a16dc6bdfc93d50f Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Thu, 24 Sep 2026 16:46:08 -0700
Subject: [PATCH 03/10] fix(cloud): preserve execution errors and structured
 guidance warnings

---
 airbyte/_direct_connectors/connector_docs.py  |  2 +-
 airbyte/cloud/connectors.py                   |  8 +-
 tests/unit_tests/test_cloud_describe.py       | 12 ++-
 tests/unit_tests/test_cloud_direct_execute.py | 88 +++----------------
 4 files changed, 24 insertions(+), 86 deletions(-)

diff --git a/airbyte/_direct_connectors/connector_docs.py b/airbyte/_direct_connectors/connector_docs.py
index ab2d303f3..3402e6a40 100644
--- a/airbyte/_direct_connectors/connector_docs.py
+++ b/airbyte/_direct_connectors/connector_docs.py
@@ -215,6 +215,7 @@ def build_direct_access_sql_guidance(
         title=f"{destination.name} (SQL passthrough destination)",
         summary=f"Table layout and naming for the `{destination.name}` destination.",
         tags=["destination"],
+        warnings=[sql_passthrough_notice] if sql_passthrough_notice else [],
     )
     connections = _destination_connections(destination)
     content: list[dict[str, Any]] = []
@@ -247,7 +248,6 @@ def build_direct_access_sql_guidance(
         outline=[],
         section_id=None,
         content=content,
-        warnings=[sql_passthrough_notice] if sql_passthrough_notice else [],
     )
 
 
diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py
index d2dcfd453..89cc8388d 100644
--- a/airbyte/cloud/connectors.py
+++ b/airbyte/cloud/connectors.py
@@ -59,7 +59,6 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol
 
-import requests
 import yaml
 from pydantic import ValidationError
 
@@ -691,11 +690,8 @@ def _execute_direct_action(  # noqa: PLR0913  # Explicit args mirror the public
         """Execute a single entity/action operation through the Cloud Config API.
 
         Raises `AirbyteExternalAccessNotEnabledError` without any network call when the
-        workspace's API roots have no Context layer API. When the Cloud Config API
-        reports the connector as forbidden or not found, the error is re-raised as
-        `AirbyteExternalAccessNotEnabledError` only when external access is actually
-        disabled for the connector; the original `AirbyteError` propagates otherwise,
-        including when the enablement lookup itself fails.
+        workspace's API roots have no Context layer API. Execution errors propagate
+        unchanged: an unavailable docs probe cannot establish disabled access.
         """
         self._require_context_layer_api()
         if read_only and action in {write_action.value for write_action in ExternalApiWriteAction}:
diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py
index 500958621..7783a3a18 100644
--- a/tests/unit_tests/test_cloud_describe.py
+++ b/tests/unit_tests/test_cloud_describe.py
@@ -15,6 +15,10 @@
 from airbyte._direct_connectors import connector_docs
 from airbyte.cloud import workspaces as cloud_workspaces
 from airbyte.mcp import cloud as cloud_mcp
+from airbyte.mcp._docs_results import (
+    render_connector_docs_result,
+    render_agent_skill_docs_result,
+)
 from airbyte.cloud.connections import CloudConnection
 from airbyte.cloud.connectors import (
     CloudConnector,
@@ -409,7 +413,9 @@ def test_get_direct_access_guidance_destination_not_enabled_adds_notice(
     notice = connector_docs.SQL_PASSTHROUGH_NOT_ENABLED_NOTICE
     guidance = destination.get_direct_access_guidance()
     assert guidance.content[0] == {"type": "paragraph", "text": notice}
-    assert guidance.warnings == [notice]
+    assert guidance.metadata.warnings == [notice]
+    assert render_connector_docs_result(guidance).warnings == [notice]
+    assert render_agent_skill_docs_result(guidance).warnings == [notice]
     assert guidance.outline == []
     assert any(block.get("type") == "table" for block in guidance.content)
     assert "execute_external_sql_query" not in _content_text(guidance)
@@ -546,7 +552,9 @@ def test_get_direct_access_guidance_destination_no_context_layer_notices(
     guidance = destination.get_direct_access_guidance()
     read_docs.assert_not_called()
     assert guidance.content[0] == {"type": "paragraph", "text": notice}
-    assert guidance.warnings == [notice]
+    assert guidance.metadata.warnings == [notice]
+    assert render_connector_docs_result(guidance).warnings == [notice]
+    assert render_agent_skill_docs_result(guidance).warnings == [notice]
     assert guidance.outline == []
     assert "execute_external_sql_query" not in _content_text(guidance)
     assert "SHOW TABLES" not in _content_text(guidance)
diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py
index 514e2e283..4652aabf2 100644
--- a/tests/unit_tests/test_cloud_direct_execute.py
+++ b/tests/unit_tests/test_cloud_direct_execute.py
@@ -6,7 +6,6 @@
 import json
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import patch
 
 import pytest
 import requests
@@ -137,15 +136,6 @@ def _seed_destination(
     return destination
 
 
-def _patch_direct_access(*, enabled: bool) -> Any:
-    """Mock `CloudConnector.is_feature_enabled` without hitting the API."""
-    return patch.object(
-        CloudConnector,
-        "is_feature_enabled",
-        return_value=enabled,
-    )
-
-
 @pytest.mark.parametrize(
     ("action_input", "expected_action"),
     [
@@ -270,83 +260,27 @@ def test_execute_direct_action_rejects_write_action_as_read_only(
     assert calls == []
 
 
-_FLAG_LOOKUP_ERROR: Any = object()
-"""Marker: the feature lookup itself fails."""
-
-
-@pytest.mark.parametrize(
-    ("execute_status", "access_flag", "expected_exc", "expected_status"),
-    [
-        pytest.param(
-            403,
-            False,
-            AirbyteExternalAccessNotEnabledError,
-            None,
-            id="forbidden_disabled_raises_not_enabled",
-        ),
-        pytest.param(
-            404,
-            True,
-            AirbyteError,
-            404,
-            id="not_found_enabled_reraises",
-        ),
-        pytest.param(
-            403,
-            _FLAG_LOOKUP_ERROR,
-            AirbyteError,
-            403,
-            id="flag_lookup_failure_reraises",
-        ),
-        pytest.param(
-            500,
-            True,
-            AirbyteError,
-            500,
-            id="other_error_propagates",
-        ),
-    ],
-)
+@pytest.mark.parametrize("execute_status", [403, 404, 500])
+@pytest.mark.parametrize("docs_status", [403, 404, 500])
 def test_execute_error_handling(
     monkeypatch: pytest.MonkeyPatch,
     execute_status: int,
-    access_flag: Any,  # noqa: ANN401
-    expected_exc: type[Exception],
-    expected_status: int | None,
+    docs_status: int,
 ) -> None:
     workspace = _make_workspace(monkeypatch)
     _patch_context_layer(monkeypatch)
-    _patch_execute(
-        monkeypatch,
-        {},
-        error=AirbyteCloudApiError(
-            status_code=execute_status,
-            context={"status_code": execute_status},
-        ),
-    )
+    error = AirbyteCloudApiError(status_code=execute_status)
+    _patch_execute(monkeypatch, {}, error=error)
     source = _seed_source(workspace, "source-1", "GitHub Issues")
 
-    flag_patch = (
-        _patch_direct_access(enabled=access_flag)
-        if isinstance(access_flag, bool)
-        else patch.object(
-            CloudConnector,
-            "is_feature_enabled",
-            side_effect=AirbyteError(context={"status_code": 500}),
-        )
-    )
-    with flag_patch, pytest.raises(expected_exc) as exc_info:
+    def unavailable_docs(**_kwargs: Any) -> dict[str, Any]:
+        raise AirbyteError(context={"status_code": docs_status})
+
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", unavailable_docs)
+    with pytest.raises(AirbyteError) as exc_info:
         source.execute_api_query("issues")
 
-    if expected_exc is AirbyteExternalAccessNotEnabledError:
-        assert isinstance(exc_info.value, AirbyteExternalAccessNotEnabledError)
-        assert exc_info.value.connector_id == "source-1"
-        assert exc_info.value.connector_name == "GitHub Issues"
-    else:
-        assert isinstance(exc_info.value, AirbyteError)
-        assert not isinstance(exc_info.value, AirbyteExternalAccessNotEnabledError)
-        assert isinstance(exc_info.value, AirbyteCloudApiError)
-        assert exc_info.value.status_code == expected_status
+    assert exc_info.value is error
 
 
 @pytest.mark.parametrize("method_name", ["execute_api_query", "execute_api_action"])

From fc9cbc5a2dba76d438b2d237f1026d6ed92160c1 Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Fri, 25 Sep 2026 10:16:16 -0700
Subject: [PATCH 04/10] fix(mcp): report unknown feature probes and normalize
 warnings

---
 airbyte/cloud/connectors.py             | 17 +++++----
 airbyte/mcp/cloud.py                    |  5 +--
 tests/unit_tests/test_cloud_describe.py | 46 ++++++++++++++++++++-----
 3 files changed, 51 insertions(+), 17 deletions(-)

diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py
index 89cc8388d..84db2a61d 100644
--- a/airbyte/cloud/connectors.py
+++ b/airbyte/cloud/connectors.py
@@ -165,16 +165,16 @@ def __init__(
 
     def get_enabled_features(
         self, *, warnings: list[str] | None = None
-    ) -> frozenset[ConnectorFeature]:
+    ) -> frozenset[ConnectorFeature] | None:
         """Resolve connector features, optionally collecting unavailable-docs warnings.
 
-        Successful features are cached. Unavailable docs (403/404) return no features
+        Successful features are cached. Unavailable docs (403/404) return `None`
         without caching absence; other probe failures propagate so callers can distinguish
         unknown support from an empty feature set.
         """
         if self._enabled_features is not None:
             if warnings is not None and self._context_layer_details is not None:
-                warnings.extend(self._context_layer_details.warnings)
+                warnings.extend(str(warning) for warning in self._context_layer_details.warnings)
             return self._enabled_features
 
         features = self.workspace._get_connector_features(self, warnings=warnings)  # noqa: SLF001
@@ -184,8 +184,11 @@ def get_enabled_features(
 
     @property
     def enabled_features(self) -> frozenset[ConnectorFeature]:
-        """The connector's enabled features; successful lookups are cached."""
-        return self.get_enabled_features()
+        """Known enabled features; unavailable probes yield an empty set.
+
+        Use `get_enabled_features()` to distinguish unknown support from no features.
+        """
+        return self.get_enabled_features() or frozenset()
 
     def is_feature_enabled(self, feature: ConnectorFeature) -> bool:
         """Whether `feature` is enabled for this connector.
@@ -752,7 +755,7 @@ def _context_layer_inspect(
         response, transport) is raised to the caller.
         """
         if self._context_layer_details is not None and not force_refresh:
-            warnings.extend(self._context_layer_details.warnings)
+            warnings.extend(str(warning) for warning in self._context_layer_details.warnings)
             return self._context_layer_details
         skill_id = (
             connector_docs.source_skill_id(self.connector_id)
@@ -786,7 +789,7 @@ def _context_layer_inspect(
             warnings=list(docs.metadata.warnings),
         )
         self._context_layer_details = parsed
-        warnings.extend(parsed.warnings)
+        warnings.extend(str(warning) for warning in parsed.warnings)
         return parsed
 
     def _fetch_enablement(self) -> ConnectorEnablement | None:
diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index e78121d6f..7fe4e8545 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -1349,8 +1349,9 @@ def _describe_cloud_connector(
         integration_name = None
 
     try:
-        enabled_features: list[ConnectorFeature] | Literal["unknown"] = sorted(
-            connector.get_enabled_features(warnings=warnings)
+        features = connector.get_enabled_features(warnings=warnings)
+        enabled_features: list[ConnectorFeature] | Literal["unknown"] = (
+            "unknown" if features is None else sorted(features)
         )
     except (AirbyteError, requests.RequestException, ValueError):
         enabled_features = "unknown"
diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py
index 7783a3a18..c17186998 100644
--- a/tests/unit_tests/test_cloud_describe.py
+++ b/tests/unit_tests/test_cloud_describe.py
@@ -834,7 +834,7 @@ def test_unavailable_features_warn_without_caching_absence(
     source = _seed_source(workspace, "source-1", "GitHub")
     warnings: list[str] = []
 
-    assert source.get_enabled_features(warnings=warnings) == frozenset()
+    assert source.get_enabled_features(warnings=warnings) is None
     assert source._enabled_features is None  # noqa: SLF001
     assert len(warnings) == 1
     assert "unavailable" in warnings[0]
@@ -849,6 +849,8 @@ def test_unavailable_features_warn_without_caching_absence(
 @pytest.mark.parametrize(
     "failure",
     [
+        AirbyteError(context={"status_code": 403, "response_text": "secret"}),
+        AirbyteError(context={"status_code": 404, "response_text": "secret"}),
         AirbyteError(context={"status_code": 401, "response_text": "secret"}),
         AirbyteError(context={"status_code": 500, "response_text": "secret"}),
         requests.Timeout("secret"),
@@ -877,9 +879,8 @@ def test_describe_survives_unknown_features_and_recovers(
     assert result.connector_id == "source-1"
     assert result.replication_details == []
     assert result.enabled_features == "unknown"
-    assert result.warnings == [
-        "Connector feature lookup failed; enabled features are unknown."
-    ]
+    assert result.warnings
+    assert all("secret" not in warning for warning in result.warnings)
     assert read_docs.call_count == 1
     assert source._enabled_features is None  # noqa: SLF001
     assert ConnectorFeature.DIRECT_API_QUERY in source.enabled_features
@@ -954,8 +955,37 @@ def test_destination_server_guidance_is_not_duplicated(
         {"type": "paragraph", "text": "Authoritative SQL instructions."}
     ]
     assert docs.section_id == section
-    assert {entry.id for entry in docs.outline} == {
-        "actions.record.sql_select",
-        "connections",
-        "streams",
+    assert {entry.id for entry in docs.outline} == {"actions.record.sql_select"}
+
+
+@pytest.mark.parametrize("cache_path", ["fresh", "inspect", "features"])
+def test_describe_stringifies_structured_probe_warnings(
+    monkeypatch: pytest.MonkeyPatch, cache_path: str
+) -> None:
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    warning = {"code": "partial", "message": "Partial metadata"}
+    response = {
+        **SKILL_DOCS_RESPONSE,
+        "metadata": {**SKILL_DOCS_RESPONSE["metadata"], "warnings": [warning]},
     }
+    read_docs = MagicMock(return_value=response)
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
+    source = _seed_source(workspace, "source-1", "GitHub")
+    source._connector_definition = SimpleNamespace(name="GitHub")  # noqa: SLF001
+    if cache_path == "inspect":
+        source._context_layer_inspect(warnings=[])  # noqa: SLF001
+    elif cache_path == "features":
+        source.get_enabled_features()
+
+    result = cloud_mcp._describe_cloud_connector(  # noqa: SLF001
+        source,
+        with_config=False,
+        with_replication_details=False,
+        with_direct_access_guidance=False,
+        with_data_replication_docs=False,
+    )
+
+    assert result.warnings == [str(warning)]
+    assert result.model_dump()["warnings"] == [str(warning)]
+    assert read_docs.call_count == 1

From fe257b6dd02dcb76bc38a60078a87777772716c3 Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Fri, 25 Sep 2026 10:31:31 -0700
Subject: [PATCH 05/10] fix(cloud): retain destination not-enabled notice

---
 airbyte/cloud/connectors.py             | 2 ++
 tests/unit_tests/test_cloud_describe.py | 1 +
 2 files changed, 3 insertions(+)

diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py
index 84db2a61d..82660d0be 100644
--- a/airbyte/cloud/connectors.py
+++ b/airbyte/cloud/connectors.py
@@ -949,6 +949,8 @@ def get_direct_access_guidance(self, *, section: str | None = None) -> DirectAcc
             except exc.AirbyteError as error:
                 if not agents_api_util.is_not_enabled_error(error) or section is not None:
                     raise
+                # Keep the not-enabled notice for this fallback. A rare missing-docs
+                # 404 is indistinguishable from disabled access without an explicit signal.
                 return connector_docs.build_direct_access_sql_guidance(
                     destination,
                     sql_passthrough_notice=connector_docs.SQL_PASSTHROUGH_NOT_ENABLED_NOTICE,
diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py
index c17186998..5580d0b37 100644
--- a/tests/unit_tests/test_cloud_describe.py
+++ b/tests/unit_tests/test_cloud_describe.py
@@ -411,6 +411,7 @@ def test_get_direct_access_guidance_destination_not_enabled_adds_notice(
     )
 
     notice = connector_docs.SQL_PASSTHROUGH_NOT_ENABLED_NOTICE
+    assert "SQL passthrough is not enabled" in notice
     guidance = destination.get_direct_access_guidance()
     assert guidance.content[0] == {"type": "paragraph", "text": notice}
     assert guidance.metadata.warnings == [notice]

From b0cd1e8f70c1effff7168e3ff37678e3dc97591a Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Fri, 25 Sep 2026 13:29:17 -0700
Subject: [PATCH 06/10] fix: reconcile connector regressions with latest
 release candidate

---
 airbyte/mcp/cloud.py                          |  8 ++---
 tests/unit_tests/test_cloud_describe.py       | 16 +++++----
 tests/unit_tests/test_cloud_direct_execute.py | 34 ++++++++++---------
 tests/unit_tests/test_mcp_cloud.py            |  9 ++---
 4 files changed, 34 insertions(+), 33 deletions(-)

diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index 7fe4e8545..f791521ae 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -1350,11 +1350,11 @@ def _describe_cloud_connector(
 
     try:
         features = connector.get_enabled_features(warnings=warnings)
-        enabled_features: list[ConnectorFeature] | Literal["unknown"] = (
-            "unknown" if features is None else sorted(features)
+        enabled_features: list[ConnectorFeature] | FeaturesUnknown = (
+            FEATURES_UNKNOWN if features is None else sorted(features)
         )
     except (AirbyteError, requests.RequestException, ValueError):
-        enabled_features = "unknown"
+        enabled_features = FEATURES_UNKNOWN
         warnings.append("Connector feature lookup failed; enabled features are unknown.")
 
     result = CloudConnectorDetailsResult(
@@ -1400,7 +1400,7 @@ def _describe_cloud_connector(
         except AirbyteError as error:
             warnings.append(f"Connection listing failed: {error}")
 
-    if with_direct_access_guidance and enabled_features != "unknown":
+    if with_direct_access_guidance and enabled_features != FEATURES_UNKNOWN:
         try:
             docs = connector.get_direct_access_guidance()
         except (PyAirbyteError, requests.RequestException, ValueError):
diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py
index 5580d0b37..a4e25751c 100644
--- a/tests/unit_tests/test_cloud_describe.py
+++ b/tests/unit_tests/test_cloud_describe.py
@@ -827,7 +827,9 @@ def test_unavailable_features_warn_without_caching_absence(
     _patch_context_layer(monkeypatch)
     read_docs = MagicMock(
         side_effect=[
-            AirbyteError(context={"status_code": status, "response_text": "secret"}),
+            AirbyteCloudApiError(
+                status_code=status, context={"response_text": "secret"}
+            ),
             SKILL_DOCS_RESPONSE,
         ]
     )
@@ -850,10 +852,10 @@ def test_unavailable_features_warn_without_caching_absence(
 @pytest.mark.parametrize(
     "failure",
     [
-        AirbyteError(context={"status_code": 403, "response_text": "secret"}),
-        AirbyteError(context={"status_code": 404, "response_text": "secret"}),
-        AirbyteError(context={"status_code": 401, "response_text": "secret"}),
-        AirbyteError(context={"status_code": 500, "response_text": "secret"}),
+        AirbyteCloudApiError(status_code=403, context={"response_text": "secret"}),
+        AirbyteCloudApiError(status_code=404, context={"response_text": "secret"}),
+        AirbyteCloudApiError(status_code=401, context={"response_text": "secret"}),
+        AirbyteCloudApiError(status_code=500, context={"response_text": "secret"}),
         requests.Timeout("secret"),
         ValueError("secret"),
     ],
@@ -908,7 +910,7 @@ def test_malformed_probe_is_not_cached_or_silently_filtered(
     assert read_docs.call_count == 2
 
 
-@pytest.mark.parametrize("section", [None, "overview", "sql-passthrough"])
+@pytest.mark.parametrize("section", [None, "overview", "actions.record.sql_select"])
 def test_destination_server_guidance_is_not_duplicated(
     monkeypatch: pytest.MonkeyPatch, section: str | None
 ) -> None:
@@ -948,7 +950,7 @@ def test_destination_server_guidance_is_not_duplicated(
         MagicMock(side_effect=AssertionError("local connection listing")),
     )
 
-    docs = destination.get_direct_access_guidance(section=section)
+    docs = destination.read_agent_skill_docs(section=section)
 
     assert read_docs.call_count == 1
     assert read_docs.call_args.kwargs["section"] == server_section
diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py
index 4652aabf2..4d14ba528 100644
--- a/tests/unit_tests/test_cloud_direct_execute.py
+++ b/tests/unit_tests/test_cloud_direct_execute.py
@@ -29,6 +29,7 @@
     ConnectorFeature,
     ConnectorType,
     ExternalApiReadOnlyAction,
+    ExternalApiWriteAction,
 )
 from airbyte._direct_connectors.models import (
     _SQL_PASSTHROUGH_DESTINATION_DIALECTS,
@@ -108,7 +109,7 @@ def fake_request(**kwargs: Any) -> requests.Response:
 
     monkeypatch.setattr(requests, "request", fake_request)
 
-    def fake_execute(**kwargs: Any) -> dict[str, Any]:
+    def fake_execute(**kwargs: Any) -> ExternalApiExecuteResult:
         calls.append(kwargs)
         if error is not None:
             raise error
@@ -196,27 +197,23 @@ def test_execute_api_query_forwards_action(
     }
 
 
-def test_execute_api_action_forwards_to_cloud_api(
-    monkeypatch: pytest.MonkeyPatch,
+@pytest.mark.parametrize("action", list(ExternalApiWriteAction))
+def test_execute_api_action_is_unsupported(
+    monkeypatch: pytest.MonkeyPatch, action: ExternalApiWriteAction
 ) -> None:
     workspace = _make_workspace(monkeypatch)
     _patch_context_layer(monkeypatch)
     calls = _patch_execute(monkeypatch, {"data": {"id": 1}})
     source = _seed_source(workspace, "source-1", "GitHub Issues")
 
-    result = source.execute_api_action(
-        "issues",
-        "create",  # type: ignore[arg-type]
-        {"title": "New issue"},
-        intent="file a bug",
-    )
+    with pytest.raises(
+        PyAirbyteInputError, match="write actions are not supported yet"
+    ):
+        source.execute_api_action(
+            "issues", action, {"title": "New issue"}, intent="file a bug"
+        )
 
-    assert isinstance(result, ExternalApiExecuteResult)
-    body = calls[0]["request_body"]
-    assert body["entity"] == "issues"
-    assert body["action"] == "create"
-    assert body["params"] == {"title": "New issue"}
-    assert body["intent"] == "file a bug"
+    assert calls == []
 
 
 @pytest.mark.parametrize(
@@ -309,7 +306,12 @@ def test_direct_methods_raise_without_context_layer(
     for connector in connectors:
         for method in (method_name, "execute_sql_query"):
             args = method_args if method == method_name else ("SELECT 1",)
-            with pytest.raises(AirbyteExternalAccessNotEnabledError):
+            expected_error = (
+                PyAirbyteInputError
+                if method == "execute_api_action"
+                else AirbyteExternalAccessNotEnabledError
+            )
+            with pytest.raises(expected_error):
                 getattr(connector, method)(*args)
 
     assert calls == []
diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py
index 6c4916d57..e457db9a8 100644
--- a/tests/unit_tests/test_mcp_cloud.py
+++ b/tests/unit_tests/test_mcp_cloud.py
@@ -2266,14 +2266,11 @@ def test_describe_cloud_connector_probe_failure_marks_unknown() -> None:
     """A feature-probe failure marks `enabled_features` `"unknown"` with a warning."""
 
     class _FailingFeaturesConnector(_DescribedConnector):
-        @property
-        def enabled_features(self) -> frozenset[ConnectorFeature]:
+        def get_enabled_features(
+            self, *, warnings: list[str]
+        ) -> frozenset[ConnectorFeature]:
             raise AirbyteCloudApiError(status_code=504)
 
-        @enabled_features.setter
-        def enabled_features(self, value: frozenset[ConnectorFeature]) -> None:
-            pass
-
     result = _describe(_FailingFeaturesConnector())
 
     assert result.enabled_features == cloud_mcp.FEATURES_UNKNOWN

From d15debc0718f1b0813bc53994abcc14df1c28404 Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Mon, 28 Sep 2026 12:56:26 -0700
Subject: [PATCH 07/10] refactor(cloud): keep enabled-feature helper private

---
 airbyte/cloud/connectors.py             | 9 +++------
 airbyte/mcp/cloud.py                    | 2 +-
 tests/unit_tests/test_cloud_describe.py | 4 ++--
 tests/unit_tests/test_mcp_cloud.py      | 4 ++--
 4 files changed, 8 insertions(+), 11 deletions(-)

diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py
index 82660d0be..6156e004f 100644
--- a/airbyte/cloud/connectors.py
+++ b/airbyte/cloud/connectors.py
@@ -163,7 +163,7 @@ def __init__(
         self._connector_definition: _ConnectorDefinitionLike | None = None
         """The connector definition lookup result. (Cached; `None` until fetched.)"""
 
-    def get_enabled_features(
+    def _get_enabled_features(
         self, *, warnings: list[str] | None = None
     ) -> frozenset[ConnectorFeature] | None:
         """Resolve connector features, optionally collecting unavailable-docs warnings.
@@ -184,11 +184,8 @@ def get_enabled_features(
 
     @property
     def enabled_features(self) -> frozenset[ConnectorFeature]:
-        """Known enabled features; unavailable probes yield an empty set.
-
-        Use `get_enabled_features()` to distinguish unknown support from no features.
-        """
-        return self.get_enabled_features() or frozenset()
+        """Known enabled features; unavailable probes yield an empty set."""
+        return self._get_enabled_features() or frozenset()
 
     def is_feature_enabled(self, feature: ConnectorFeature) -> bool:
         """Whether `feature` is enabled for this connector.
diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index f791521ae..532d638db 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -1349,7 +1349,7 @@ def _describe_cloud_connector(
         integration_name = None
 
     try:
-        features = connector.get_enabled_features(warnings=warnings)
+        features = connector._get_enabled_features(warnings=warnings)  # noqa: SLF001
         enabled_features: list[ConnectorFeature] | FeaturesUnknown = (
             FEATURES_UNKNOWN if features is None else sorted(features)
         )
diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py
index a4e25751c..43f3fd73d 100644
--- a/tests/unit_tests/test_cloud_describe.py
+++ b/tests/unit_tests/test_cloud_describe.py
@@ -837,7 +837,7 @@ def test_unavailable_features_warn_without_caching_absence(
     source = _seed_source(workspace, "source-1", "GitHub")
     warnings: list[str] = []
 
-    assert source.get_enabled_features(warnings=warnings) is None
+    assert source._get_enabled_features(warnings=warnings) is None  # noqa: SLF001
     assert source._enabled_features is None  # noqa: SLF001
     assert len(warnings) == 1
     assert "unavailable" in warnings[0]
@@ -979,7 +979,7 @@ def test_describe_stringifies_structured_probe_warnings(
     if cache_path == "inspect":
         source._context_layer_inspect(warnings=[])  # noqa: SLF001
     elif cache_path == "features":
-        source.get_enabled_features()
+        source._get_enabled_features()  # noqa: SLF001
 
     result = cloud_mcp._describe_cloud_connector(  # noqa: SLF001
         source,
diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py
index e457db9a8..63a05ffd5 100644
--- a/tests/unit_tests/test_mcp_cloud.py
+++ b/tests/unit_tests/test_mcp_cloud.py
@@ -1763,7 +1763,7 @@ def integration_name(self) -> str:
             raise self._integration_name
         return self._integration_name
 
-    def get_enabled_features(
+    def _get_enabled_features(
         self, *, warnings: list[str]
     ) -> frozenset[ConnectorFeature]:
         if self.workspace._has_context_layer_api():
@@ -2266,7 +2266,7 @@ def test_describe_cloud_connector_probe_failure_marks_unknown() -> None:
     """A feature-probe failure marks `enabled_features` `"unknown"` with a warning."""
 
     class _FailingFeaturesConnector(_DescribedConnector):
-        def get_enabled_features(
+        def _get_enabled_features(
             self, *, warnings: list[str]
         ) -> frozenset[ConnectorFeature]:
             raise AirbyteCloudApiError(status_code=504)

From 58de65bababf249a5a5aa502e3e9320b454f8ff7 Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Mon, 28 Sep 2026 13:02:30 -0700
Subject: [PATCH 08/10] fix(cloud): restore existing write execution path

---
 airbyte/cloud/connectors.py                   | 23 ++++++------
 airbyte/mcp/cloud.py                          | 10 ++++--
 tests/unit_tests/test_cloud_direct_execute.py | 25 ++++++-------
 tests/unit_tests/test_mcp_cloud.py            | 35 -------------------
 4 files changed, 31 insertions(+), 62 deletions(-)

diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py
index 6156e004f..8ede0e4c2 100644
--- a/airbyte/cloud/connectors.py
+++ b/airbyte/cloud/connectors.py
@@ -432,10 +432,11 @@ def execute_api_action(
         skip_truncation: bool = True,
         intent: str | None = None,
     ) -> ExternalApiExecuteResult:
-        """Reject unsupported Cloud writes (`create`, `update`, or `delete`).
+        """Run a write action (`create`, `update`, or `delete`) on one entity type.
 
-        Cloud connector execution currently supports read actions only. This method
-        raises `PyAirbyteInputError` without sending a request.
+        Requires external access to be enabled for this connector in its organization's
+        Context Layer settings. Connectors without an entity API fail with the Agents
+        API's own error.
         """
         try:
             resolved_action = ExternalApiWriteAction(action)
@@ -446,13 +447,15 @@ def execute_api_action(
                 context={"entity_type": entity_type, "action": action},
             ) from None
 
-        _ = api_args, select_fields, exclude_fields, skip_truncation, intent
-        raise exc.PyAirbyteInputError(
-            message="Cloud connector write actions are not supported yet.",
-            guidance=(
-                "Use execute_api_query (Python) or execute_external_api_query (MCP) for reads."
-            ),
-            context={"entity_type": entity_type, "action": resolved_action.value},
+        return self._execute_direct_action(
+            entity_type=entity_type,
+            action=resolved_action.value,
+            api_args=api_args,
+            select_fields=select_fields,
+            exclude_fields=exclude_fields,
+            skip_truncation=skip_truncation,
+            intent=intent,
+            read_only=False,
         )
 
     def execute_sql_query(
diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index 532d638db..dd6b84e74 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -1739,10 +1739,14 @@ def _execute_external_api_action(  # noqa: PLR0913  # Explicit args mirror the c
         ),
     ] = None,
 ) -> ExternalApiExecuteResult:
-    """Cloud connector write actions are not supported yet.
+    """Run a write action through a deployed Cloud connector's direct API.
 
-    Returns a clear error without executing a write. Use `execute_external_api_query`
-    for supported read actions (`list`, `get`, or `search`).
+    Creates, updates, or deletes data in the external system.
+
+    Before calling, read the connector's action docs with `get_agent_skill_docs`
+    (or `describe_cloud_connector` with `with_direct_access_guidance=True`) to
+    learn the entity types, actions, and required `api_args`; argument names
+    differ per connector and are not guessable.
     """
     connector = _get_cloud_workspace(ctx, workspace_id).get_connector(connector_id)
     return connector.execute_api_action(
diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py
index 4d14ba528..f65139e00 100644
--- a/tests/unit_tests/test_cloud_direct_execute.py
+++ b/tests/unit_tests/test_cloud_direct_execute.py
@@ -198,7 +198,7 @@ def test_execute_api_query_forwards_action(
 
 
 @pytest.mark.parametrize("action", list(ExternalApiWriteAction))
-def test_execute_api_action_is_unsupported(
+def test_execute_api_action_forwards_to_cloud_api(
     monkeypatch: pytest.MonkeyPatch, action: ExternalApiWriteAction
 ) -> None:
     workspace = _make_workspace(monkeypatch)
@@ -206,14 +206,16 @@ def test_execute_api_action_is_unsupported(
     calls = _patch_execute(monkeypatch, {"data": {"id": 1}})
     source = _seed_source(workspace, "source-1", "GitHub Issues")
 
-    with pytest.raises(
-        PyAirbyteInputError, match="write actions are not supported yet"
-    ):
-        source.execute_api_action(
-            "issues", action, {"title": "New issue"}, intent="file a bug"
-        )
+    result = source.execute_api_action(
+        "issues", action, {"title": "New issue"}, intent="file a bug"
+    )
 
-    assert calls == []
+    assert isinstance(result, ExternalApiExecuteResult)
+    body = calls[0]["request_body"]
+    assert body["entity"] == "issues"
+    assert body["action"] == action.value
+    assert body["params"] == {"title": "New issue"}
+    assert body["intent"] == "file a bug"
 
 
 @pytest.mark.parametrize(
@@ -306,12 +308,7 @@ def test_direct_methods_raise_without_context_layer(
     for connector in connectors:
         for method in (method_name, "execute_sql_query"):
             args = method_args if method == method_name else ("SELECT 1",)
-            expected_error = (
-                PyAirbyteInputError
-                if method == "execute_api_action"
-                else AirbyteExternalAccessNotEnabledError
-            )
-            with pytest.raises(expected_error):
+            with pytest.raises(AirbyteExternalAccessNotEnabledError):
                 getattr(connector, method)(*args)
 
     assert calls == []
diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py
index 63a05ffd5..9637e47a4 100644
--- a/tests/unit_tests/test_mcp_cloud.py
+++ b/tests/unit_tests/test_mcp_cloud.py
@@ -29,15 +29,9 @@
 )
 from airbyte.cloud.connectors import (
     CheckResult,
-    CloudSource,
     ConnectorFeature,
     ConnectorType,
 )
-from airbyte.cloud.workspaces import CloudWorkspace
-from airbyte.mcp._error_handling import (
-    MCP_TOOL_USER_FACING_ERRORS,
-    format_user_facing_error,
-)
 from airbyte.cloud.models import (
     CloudDefaultContextInfo,
     CloudOrganizationInfo,
@@ -2275,32 +2269,3 @@ def _get_enabled_features(
 
     assert result.enabled_features == cloud_mcp.FEATURES_UNKNOWN
     assert any("enabled features are unknown" in w for w in result.warnings)
-
-
-@pytest.mark.parametrize("action", list(ExternalApiWriteAction))
-def test_cloud_write_tool_reports_unsupported_without_network(
-    monkeypatch: pytest.MonkeyPatch, action: ExternalApiWriteAction
-) -> None:
-    workspace = CloudWorkspace(workspace_id="workspace-1", bearer_token="token")
-    source = CloudSource(workspace=workspace, connector_id="source-1")
-    monkeypatch.setattr(workspace, "get_connector", lambda _: source)
-    monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_: workspace)
-    request = MagicMock(side_effect=AssertionError("unsupported write sent to server"))
-    monkeypatch.setattr(requests, "request", request)
-
-    with pytest.raises(PyAirbyteInputError) as raised:
-        cloud_mcp.execute_external_api_action(
-            None,
-            connector_id="source-1",
-            entity_type="issues",
-            action=action,
-            api_args={"title": "example"},
-            workspace_id="workspace-1",
-        )
-
-    request.assert_not_called()
-    assert isinstance(raised.value, MCP_TOOL_USER_FACING_ERRORS)
-    assert format_user_facing_error(raised.value) == (
-        "Cloud connector write actions are not supported yet. "
-        "Use execute_api_query (Python) or execute_external_api_query (MCP) for reads."
-    )

From 9e15721784eead473e0cf24eb32ddcb5a81ef0d6 Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Mon, 28 Sep 2026 16:06:23 -0700
Subject: [PATCH 09/10] fix: reconcile guidance fixes with authoritative
 enablement

---
 airbyte/cloud/connectors.py                   |  40 ++--
 airbyte/mcp/cloud.py                          |  18 +-
 tests/unit_tests/test_cloud_describe.py       | 184 +++++-------------
 tests/unit_tests/test_cloud_direct_execute.py | 101 ++++++++--
 tests/unit_tests/test_mcp_cloud.py            |  24 +--
 5 files changed, 161 insertions(+), 206 deletions(-)

diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py
index 8ede0e4c2..7bce63afb 100644
--- a/airbyte/cloud/connectors.py
+++ b/airbyte/cloud/connectors.py
@@ -59,6 +59,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol
 
+import requests
 import yaml
 from pydantic import ValidationError
 
@@ -163,29 +164,17 @@ def __init__(
         self._connector_definition: _ConnectorDefinitionLike | None = None
         """The connector definition lookup result. (Cached; `None` until fetched.)"""
 
-    def _get_enabled_features(
-        self, *, warnings: list[str] | None = None
-    ) -> frozenset[ConnectorFeature] | None:
-        """Resolve connector features, optionally collecting unavailable-docs warnings.
+    def _get_enabled_features(self) -> frozenset[ConnectorFeature]:
+        """Return the enabled features, resolving them through the workspace on first use."""
+        if self._enabled_features is None:
+            self._enabled_features = self.workspace._get_connector_features(self)  # noqa: SLF001
 
-        Successful features are cached. Unavailable docs (403/404) return `None`
-        without caching absence; other probe failures propagate so callers can distinguish
-        unknown support from an empty feature set.
-        """
-        if self._enabled_features is not None:
-            if warnings is not None and self._context_layer_details is not None:
-                warnings.extend(str(warning) for warning in self._context_layer_details.warnings)
-            return self._enabled_features
-
-        features = self.workspace._get_connector_features(self, warnings=warnings)  # noqa: SLF001
-        if features:
-            self._enabled_features = features
-        return features
+        return self._enabled_features
 
     @property
     def enabled_features(self) -> frozenset[ConnectorFeature]:
-        """Known enabled features; unavailable probes yield an empty set."""
-        return self._get_enabled_features() or frozenset()
+        """The features enabled for this connector. Resolved on first access and cached."""
+        return self._get_enabled_features()
 
     def is_feature_enabled(self, feature: ConnectorFeature) -> bool:
         """Whether `feature` is enabled for this connector.
@@ -693,8 +682,11 @@ def _execute_direct_action(  # noqa: PLR0913  # Explicit args mirror the public
         """Execute a single entity/action operation through the Cloud Config API.
 
         Raises `AirbyteExternalAccessNotEnabledError` without any network call when the
-        workspace's API roots have no Context layer API. Execution errors propagate
-        unchanged: an unavailable docs probe cannot establish disabled access.
+        workspace's API roots have no Context layer API. When the Cloud Config API
+        reports the connector as forbidden or not found, the error is re-raised as
+        `AirbyteExternalAccessNotEnabledError` only when external access is actually
+        disabled for the connector; the original `AirbyteError` propagates otherwise,
+        including when the enablement lookup itself fails.
         """
         self._require_context_layer_api()
         if read_only and action in {write_action.value for write_action in ExternalApiWriteAction}:
@@ -755,7 +747,6 @@ def _context_layer_inspect(
         response, transport) is raised to the caller.
         """
         if self._context_layer_details is not None and not force_refresh:
-            warnings.extend(str(warning) for warning in self._context_layer_details.warnings)
             return self._context_layer_details
         skill_id = (
             connector_docs.source_skill_id(self.connector_id)
@@ -789,7 +780,6 @@ def _context_layer_inspect(
             warnings=list(docs.metadata.warnings),
         )
         self._context_layer_details = parsed
-        warnings.extend(str(warning) for warning in parsed.warnings)
         return parsed
 
     def _fetch_enablement(self) -> ConnectorEnablement | None:
@@ -949,8 +939,8 @@ def get_direct_access_guidance(self, *, section: str | None = None) -> DirectAcc
             except exc.AirbyteError as error:
                 if not agents_api_util.is_not_enabled_error(error) or section is not None:
                     raise
-                # Keep the not-enabled notice for this fallback. A rare missing-docs
-                # 404 is indistinguishable from disabled access without an explicit signal.
+                # Keep the not-enabled notice for this docs fallback. A rare missing-docs
+                # 404 can reach it even when access is enabled.
                 return connector_docs.build_direct_access_sql_guidance(
                     destination,
                     sql_passthrough_notice=connector_docs.SQL_PASSTHROUGH_NOT_ENABLED_NOTICE,
diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index dd6b84e74..57855e137 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -1348,15 +1348,6 @@ def _describe_cloud_connector(
         warnings.append(f"Integration name lookup failed: {error}")
         integration_name = None
 
-    try:
-        features = connector._get_enabled_features(warnings=warnings)  # noqa: SLF001
-        enabled_features: list[ConnectorFeature] | FeaturesUnknown = (
-            FEATURES_UNKNOWN if features is None else sorted(features)
-        )
-    except (AirbyteError, requests.RequestException, ValueError):
-        enabled_features = FEATURES_UNKNOWN
-        warnings.append("Connector feature lookup failed; enabled features are unknown.")
-
     result = CloudConnectorDetailsResult(
         connector_id=connector.connector_id,
         connector_type=connector_type.value,
@@ -1364,12 +1355,11 @@ def _describe_cloud_connector(
         connector_url=connector.connector_url,
         connector_definition_id=connector.definition_id,
         integration_name=integration_name,
-        enabled_features=enabled_features,
     )
 
     try:
         result.enabled_features = sorted(connector.enabled_features)
-    except (AirbyteError, requests.RequestException) as error:
+    except (AirbyteError, requests.RequestException, ValueError) as error:
         result.enabled_features = FEATURES_UNKNOWN
         warnings.append(_feature_lookup_warning(error))
 
@@ -1382,8 +1372,8 @@ def _describe_cloud_connector(
             context_layer = connector._context_layer_inspect(  # noqa: SLF001
                 warnings=warnings,
             )
-        except (AirbyteError, requests.RequestException) as error:
-            warnings.append(f"Connector direct-access docs lookup failed: {error}")
+        except (AirbyteError, requests.RequestException, ValueError):
+            warnings.append("Connector direct-access docs lookup failed.")
         else:
             if context_layer is not None:
                 warnings.extend(str(warning) for warning in context_layer.warnings)
@@ -1400,7 +1390,7 @@ def _describe_cloud_connector(
         except AirbyteError as error:
             warnings.append(f"Connection listing failed: {error}")
 
-    if with_direct_access_guidance and enabled_features != FEATURES_UNKNOWN:
+    if with_direct_access_guidance:
         try:
             docs = connector.get_direct_access_guidance()
         except (PyAirbyteError, requests.RequestException, ValueError):
diff --git a/tests/unit_tests/test_cloud_describe.py b/tests/unit_tests/test_cloud_describe.py
index 43f3fd73d..ac797ef70 100644
--- a/tests/unit_tests/test_cloud_describe.py
+++ b/tests/unit_tests/test_cloud_describe.py
@@ -411,7 +411,6 @@ def test_get_direct_access_guidance_destination_not_enabled_adds_notice(
     )
 
     notice = connector_docs.SQL_PASSTHROUGH_NOT_ENABLED_NOTICE
-    assert "SQL passthrough is not enabled" in notice
     guidance = destination.get_direct_access_guidance()
     assert guidance.content[0] == {"type": "paragraph", "text": notice}
     assert guidance.metadata.warnings == [notice]
@@ -819,167 +818,82 @@ def test_is_feature_enabled_wrong_connector_kind_short_circuits(
     get_features.assert_not_called()
 
 
-@pytest.mark.parametrize("status", [403, 404])
-def test_unavailable_features_warn_without_caching_absence(
-    monkeypatch: pytest.MonkeyPatch, status: int
-) -> None:
-    workspace = _make_workspace(monkeypatch)
-    _patch_context_layer(monkeypatch)
-    read_docs = MagicMock(
-        side_effect=[
-            AirbyteCloudApiError(
-                status_code=status, context={"response_text": "secret"}
-            ),
-            SKILL_DOCS_RESPONSE,
-        ]
-    )
-    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
-    source = _seed_source(workspace, "source-1", "GitHub")
-    warnings: list[str] = []
-
-    assert source._get_enabled_features(warnings=warnings) is None  # noqa: SLF001
-    assert source._enabled_features is None  # noqa: SLF001
-    assert len(warnings) == 1
-    assert "unavailable" in warnings[0]
-    assert "secret" not in warnings[0]
-    assert "not enabled" not in warnings[0]
-    assert ConnectorFeature.DIRECT_API_QUERY in source.enabled_features
-    assert read_docs.call_count == 2
-    assert source.enabled_features
-    assert read_docs.call_count == 2
-
-
-@pytest.mark.parametrize(
-    "failure",
-    [
-        AirbyteCloudApiError(status_code=403, context={"response_text": "secret"}),
-        AirbyteCloudApiError(status_code=404, context={"response_text": "secret"}),
-        AirbyteCloudApiError(status_code=401, context={"response_text": "secret"}),
-        AirbyteCloudApiError(status_code=500, context={"response_text": "secret"}),
-        requests.Timeout("secret"),
-        ValueError("secret"),
-    ],
-)
-def test_describe_survives_unknown_features_and_recovers(
-    monkeypatch: pytest.MonkeyPatch, failure: Exception
+@pytest.mark.parametrize("cache_path", ["fresh", "inspect"])
+def test_describe_stringifies_structured_probe_warnings(
+    monkeypatch: pytest.MonkeyPatch, cache_path: str
 ) -> None:
     workspace = _make_workspace(monkeypatch)
     _patch_context_layer(monkeypatch)
-    read_docs = MagicMock(side_effect=[failure, SKILL_DOCS_RESPONSE])
+    _patch_enablement(monkeypatch)
+    warning = {"code": "partial", "message": "Partial metadata"}
+    response = {
+        **SKILL_DOCS_RESPONSE,
+        "metadata": {**SKILL_DOCS_RESPONSE["metadata"], "warnings": [warning]},
+    }
+    read_docs = MagicMock(return_value=response)
     monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
     source = _seed_source(workspace, "source-1", "GitHub")
     source._connector_definition = SimpleNamespace(name="GitHub")  # noqa: SLF001
-    monkeypatch.setattr(connector_docs, "build_connection_details", lambda _: [])
+    if cache_path == "inspect":
+        source._context_layer_inspect(warnings=[])  # noqa: SLF001
 
     result = cloud_mcp._describe_cloud_connector(  # noqa: SLF001
         source,
         with_config=False,
-        with_replication_details=True,
-        with_direct_access_guidance=True,
+        with_replication_details=False,
+        with_direct_access_guidance=False,
         with_data_replication_docs=False,
     )
 
-    assert result.connector_id == "source-1"
-    assert result.replication_details == []
-    assert result.enabled_features == "unknown"
-    assert result.warnings
-    assert all("secret" not in warning for warning in result.warnings)
+    assert result.warnings == [str(warning)]
+    assert result.model_dump()["warnings"] == [str(warning)]
     assert read_docs.call_count == 1
-    assert source._enabled_features is None  # noqa: SLF001
-    assert ConnectorFeature.DIRECT_API_QUERY in source.enabled_features
-    assert read_docs.call_count == 2
 
 
-def test_malformed_probe_is_not_cached_or_silently_filtered(
-    monkeypatch: pytest.MonkeyPatch,
+@pytest.mark.parametrize("with_guidance", [False, True])
+def test_describe_malformed_docs_preserves_enabled_features(
+    monkeypatch: pytest.MonkeyPatch, with_guidance: bool
 ) -> None:
     workspace = _make_workspace(monkeypatch)
     _patch_context_layer(monkeypatch)
+    _patch_enablement(monkeypatch)
+    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", lambda **_: {})
     source = _seed_source(workspace, "source-1", "GitHub")
-    monkeypatch.setattr(workspace, "list_sources", lambda: [source])
-    monkeypatch.setattr(workspace, "list_destinations", lambda: [])
-    read_docs = MagicMock(side_effect=[{}, SKILL_DOCS_RESPONSE])
-    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
-
-    with pytest.raises(ValueError):
-        workspace.list_connectors(feature_filter=ConnectorFeature.DIRECT_ACCESS)
-    assert source._enabled_features is None  # noqa: SLF001
-    assert workspace.list_connectors(feature_filter=ConnectorFeature.DIRECT_ACCESS) == [
-        source
-    ]
-    assert read_docs.call_count == 2
-
+    source._connector_definition = SimpleNamespace(name="GitHub")  # noqa: SLF001
 
-@pytest.mark.parametrize("section", [None, "overview", "actions.record.sql_select"])
-def test_destination_server_guidance_is_not_duplicated(
-    monkeypatch: pytest.MonkeyPatch, section: str | None
-) -> None:
-    workspace = _make_workspace(monkeypatch)
-    _patch_context_layer(monkeypatch)
-    server_section = (
-        "actions.record.sql_select" if section == "sql-passthrough" else section
-    )
-    server_response = {
-        "metadata": {
-            "id": "connector-destination:dest-1",
-            "kind": "connector_destination",
-            "title": "Warehouse",
-        },
-        "outline": [
-            {
-                "id": "actions.record.sql_select",
-                "title": "SQL select",
-                "available": True,
-            }
-        ],
-        "section_id": server_section,
-        "content": [{"type": "paragraph", "text": "Authoritative SQL instructions."}],
-    }
-    read_docs = MagicMock(return_value=server_response)
-    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
-    destination = _seed_destination(workspace, "dest-1", SNOWFLAKE_DEFINITION_ID)
-    # A successful server docs response must not trigger local configuration/connection reads.
-    monkeypatch.setattr(
-        connector_docs,
-        "_destination_load_context",
-        MagicMock(side_effect=AssertionError("local SQL rendering")),
-    )
-    monkeypatch.setattr(
-        workspace,
-        "list_connections",
-        MagicMock(side_effect=AssertionError("local connection listing")),
+    result = cloud_mcp._describe_cloud_connector(  # noqa: SLF001
+        source,
+        with_config=False,
+        with_replication_details=False,
+        with_direct_access_guidance=with_guidance,
+        with_data_replication_docs=False,
     )
 
-    docs = destination.read_agent_skill_docs(section=section)
-
-    assert read_docs.call_count == 1
-    assert read_docs.call_args.kwargs["section"] == server_section
-    assert docs.content == [
-        {"type": "paragraph", "text": "Authoritative SQL instructions."}
+    assert result.enabled_features == sorted([
+        ConnectorFeature.DIRECT_ACCESS,
+        ConnectorFeature.DIRECT_API_QUERY,
+    ])
+    assert result.warnings == [
+        "Connector direct-access docs lookup failed.",
+        *(["Direct access docs are unavailable."] if with_guidance else []),
     ]
-    assert docs.section_id == section
-    assert {entry.id for entry in docs.outline} == {"actions.record.sql_select"}
 
 
-@pytest.mark.parametrize("cache_path", ["fresh", "inspect", "features"])
-def test_describe_stringifies_structured_probe_warnings(
-    monkeypatch: pytest.MonkeyPatch, cache_path: str
+@pytest.mark.parametrize("status", [403, 404])
+def test_unavailable_docs_do_not_hide_enabled_features_or_leak_errors(
+    monkeypatch: pytest.MonkeyPatch, status: int
 ) -> None:
     workspace = _make_workspace(monkeypatch)
     _patch_context_layer(monkeypatch)
-    warning = {"code": "partial", "message": "Partial metadata"}
-    response = {
-        **SKILL_DOCS_RESPONSE,
-        "metadata": {**SKILL_DOCS_RESPONSE["metadata"], "warnings": [warning]},
-    }
-    read_docs = MagicMock(return_value=response)
+    _patch_enablement(monkeypatch)
+    read_docs = MagicMock(
+        side_effect=AirbyteCloudApiError(
+            status_code=status, context={"response_text": "secret"}
+        )
+    )
     monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", read_docs)
     source = _seed_source(workspace, "source-1", "GitHub")
     source._connector_definition = SimpleNamespace(name="GitHub")  # noqa: SLF001
-    if cache_path == "inspect":
-        source._context_layer_inspect(warnings=[])  # noqa: SLF001
-    elif cache_path == "features":
-        source._get_enabled_features()  # noqa: SLF001
 
     result = cloud_mcp._describe_cloud_connector(  # noqa: SLF001
         source,
@@ -989,6 +903,10 @@ def test_describe_stringifies_structured_probe_warnings(
         with_data_replication_docs=False,
     )
 
-    assert result.warnings == [str(warning)]
-    assert result.model_dump()["warnings"] == [str(warning)]
-    assert read_docs.call_count == 1
+    assert result.enabled_features == sorted([
+        ConnectorFeature.DIRECT_ACCESS,
+        ConnectorFeature.DIRECT_API_QUERY,
+    ])
+    assert result.warnings == [
+        "Connector direct-access docs are unavailable (access denied or not found)."
+    ]
diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py
index f65139e00..514e2e283 100644
--- a/tests/unit_tests/test_cloud_direct_execute.py
+++ b/tests/unit_tests/test_cloud_direct_execute.py
@@ -6,6 +6,7 @@
 import json
 from types import SimpleNamespace
 from typing import Any
+from unittest.mock import patch
 
 import pytest
 import requests
@@ -29,7 +30,6 @@
     ConnectorFeature,
     ConnectorType,
     ExternalApiReadOnlyAction,
-    ExternalApiWriteAction,
 )
 from airbyte._direct_connectors.models import (
     _SQL_PASSTHROUGH_DESTINATION_DIALECTS,
@@ -109,7 +109,7 @@ def fake_request(**kwargs: Any) -> requests.Response:
 
     monkeypatch.setattr(requests, "request", fake_request)
 
-    def fake_execute(**kwargs: Any) -> ExternalApiExecuteResult:
+    def fake_execute(**kwargs: Any) -> dict[str, Any]:
         calls.append(kwargs)
         if error is not None:
             raise error
@@ -137,6 +137,15 @@ def _seed_destination(
     return destination
 
 
+def _patch_direct_access(*, enabled: bool) -> Any:
+    """Mock `CloudConnector.is_feature_enabled` without hitting the API."""
+    return patch.object(
+        CloudConnector,
+        "is_feature_enabled",
+        return_value=enabled,
+    )
+
+
 @pytest.mark.parametrize(
     ("action_input", "expected_action"),
     [
@@ -197,9 +206,8 @@ def test_execute_api_query_forwards_action(
     }
 
 
-@pytest.mark.parametrize("action", list(ExternalApiWriteAction))
 def test_execute_api_action_forwards_to_cloud_api(
-    monkeypatch: pytest.MonkeyPatch, action: ExternalApiWriteAction
+    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     workspace = _make_workspace(monkeypatch)
     _patch_context_layer(monkeypatch)
@@ -207,13 +215,16 @@ def test_execute_api_action_forwards_to_cloud_api(
     source = _seed_source(workspace, "source-1", "GitHub Issues")
 
     result = source.execute_api_action(
-        "issues", action, {"title": "New issue"}, intent="file a bug"
+        "issues",
+        "create",  # type: ignore[arg-type]
+        {"title": "New issue"},
+        intent="file a bug",
     )
 
     assert isinstance(result, ExternalApiExecuteResult)
     body = calls[0]["request_body"]
     assert body["entity"] == "issues"
-    assert body["action"] == action.value
+    assert body["action"] == "create"
     assert body["params"] == {"title": "New issue"}
     assert body["intent"] == "file a bug"
 
@@ -259,27 +270,83 @@ def test_execute_direct_action_rejects_write_action_as_read_only(
     assert calls == []
 
 
-@pytest.mark.parametrize("execute_status", [403, 404, 500])
-@pytest.mark.parametrize("docs_status", [403, 404, 500])
+_FLAG_LOOKUP_ERROR: Any = object()
+"""Marker: the feature lookup itself fails."""
+
+
+@pytest.mark.parametrize(
+    ("execute_status", "access_flag", "expected_exc", "expected_status"),
+    [
+        pytest.param(
+            403,
+            False,
+            AirbyteExternalAccessNotEnabledError,
+            None,
+            id="forbidden_disabled_raises_not_enabled",
+        ),
+        pytest.param(
+            404,
+            True,
+            AirbyteError,
+            404,
+            id="not_found_enabled_reraises",
+        ),
+        pytest.param(
+            403,
+            _FLAG_LOOKUP_ERROR,
+            AirbyteError,
+            403,
+            id="flag_lookup_failure_reraises",
+        ),
+        pytest.param(
+            500,
+            True,
+            AirbyteError,
+            500,
+            id="other_error_propagates",
+        ),
+    ],
+)
 def test_execute_error_handling(
     monkeypatch: pytest.MonkeyPatch,
     execute_status: int,
-    docs_status: int,
+    access_flag: Any,  # noqa: ANN401
+    expected_exc: type[Exception],
+    expected_status: int | None,
 ) -> None:
     workspace = _make_workspace(monkeypatch)
     _patch_context_layer(monkeypatch)
-    error = AirbyteCloudApiError(status_code=execute_status)
-    _patch_execute(monkeypatch, {}, error=error)
+    _patch_execute(
+        monkeypatch,
+        {},
+        error=AirbyteCloudApiError(
+            status_code=execute_status,
+            context={"status_code": execute_status},
+        ),
+    )
     source = _seed_source(workspace, "source-1", "GitHub Issues")
 
-    def unavailable_docs(**_kwargs: Any) -> dict[str, Any]:
-        raise AirbyteError(context={"status_code": docs_status})
-
-    monkeypatch.setattr(agents_api_util, "read_cloud_skill_docs", unavailable_docs)
-    with pytest.raises(AirbyteError) as exc_info:
+    flag_patch = (
+        _patch_direct_access(enabled=access_flag)
+        if isinstance(access_flag, bool)
+        else patch.object(
+            CloudConnector,
+            "is_feature_enabled",
+            side_effect=AirbyteError(context={"status_code": 500}),
+        )
+    )
+    with flag_patch, pytest.raises(expected_exc) as exc_info:
         source.execute_api_query("issues")
 
-    assert exc_info.value is error
+    if expected_exc is AirbyteExternalAccessNotEnabledError:
+        assert isinstance(exc_info.value, AirbyteExternalAccessNotEnabledError)
+        assert exc_info.value.connector_id == "source-1"
+        assert exc_info.value.connector_name == "GitHub Issues"
+    else:
+        assert isinstance(exc_info.value, AirbyteError)
+        assert not isinstance(exc_info.value, AirbyteExternalAccessNotEnabledError)
+        assert isinstance(exc_info.value, AirbyteCloudApiError)
+        assert exc_info.value.status_code == expected_status
 
 
 @pytest.mark.parametrize("method_name", ["execute_api_query", "execute_api_action"])
diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py
index 9637e47a4..0725545ca 100644
--- a/tests/unit_tests/test_mcp_cloud.py
+++ b/tests/unit_tests/test_mcp_cloud.py
@@ -27,11 +27,7 @@
     ExternalSearchType,
     _SQL_PASSTHROUGH_DESTINATION_DIALECTS,
 )
-from airbyte.cloud.connectors import (
-    CheckResult,
-    ConnectorFeature,
-    ConnectorType,
-)
+from airbyte.cloud.connectors import CheckResult, ConnectorFeature, ConnectorType
 from airbyte.cloud.models import (
     CloudDefaultContextInfo,
     CloudOrganizationInfo,
@@ -1757,15 +1753,6 @@ def integration_name(self) -> str:
             raise self._integration_name
         return self._integration_name
 
-    def _get_enabled_features(
-        self, *, warnings: list[str]
-    ) -> frozenset[ConnectorFeature]:
-        if self.workspace._has_context_layer_api():
-            details = self._context_layer_inspect(warnings=warnings)
-            if details is not None:
-                warnings.extend(details.warnings)
-        return self.enabled_features
-
     def is_feature_enabled(self, feature: ConnectorFeature) -> bool:
         return feature in self.enabled_features
 
@@ -2260,11 +2247,14 @@ def test_describe_cloud_connector_probe_failure_marks_unknown() -> None:
     """A feature-probe failure marks `enabled_features` `"unknown"` with a warning."""
 
     class _FailingFeaturesConnector(_DescribedConnector):
-        def _get_enabled_features(
-            self, *, warnings: list[str]
-        ) -> frozenset[ConnectorFeature]:
+        @property
+        def enabled_features(self) -> frozenset[ConnectorFeature]:
             raise AirbyteCloudApiError(status_code=504)
 
+        @enabled_features.setter
+        def enabled_features(self, value: frozenset[ConnectorFeature]) -> None:
+            pass
+
     result = _describe(_FailingFeaturesConnector())
 
     assert result.enabled_features == cloud_mcp.FEATURES_UNKNOWN

From 97996d90c2ef320472d83be48f964bc027b29eb0 Mon Sep 17 00:00:00 2001
From: Piotr Kozikowski <piotr.kozikowski@gmail.com>
Date: Mon, 28 Sep 2026 16:53:29 -0700
Subject: [PATCH 10/10] test(cloud): align docs warning assertion with
 unavailable notice

---
 tests/unit_tests/test_cloud_direct_execute.py | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py
index 514e2e283..53ff35c03 100644
--- a/tests/unit_tests/test_cloud_direct_execute.py
+++ b/tests/unit_tests/test_cloud_direct_execute.py
@@ -1492,11 +1492,11 @@ def fail(**kwargs: Any) -> Any:  # noqa: ANN401
 
 
 @pytest.mark.parametrize("status_code", [403, 404])
-def test_docs_probe_not_enabled_handling_unchanged(
+def test_docs_probe_unavailable_emits_neutral_warning(
     monkeypatch: pytest.MonkeyPatch,
     status_code: int,
 ) -> None:
-    """The skills-docs probe still reads a 403 or 404 as "not enabled", with a warning."""
+    """Unavailable docs produce a warning without asserting disabled access."""
     workspace = _make_workspace(monkeypatch)
 
     def fail(**_: Any) -> Any:  # noqa: ANN401
@@ -1507,5 +1507,6 @@ def fail(**_: Any) -> Any:  # noqa: ANN401
     warnings: list[str] = []
 
     assert source._context_layer_inspect(warnings=warnings) is None  # noqa: SLF001
-    assert len(warnings) == 1
-    assert "docs lookup failed" in warnings[0]
+    assert warnings == [
+        "Connector direct-access docs are unavailable (access denied or not found)."
+    ]