fix(mcp): degrade gracefully when the connector feature probe fails

airbyte/_direct_connectors/api_util.py

@@ -27,6 +27,7 @@
 )
 from airbyte.exceptions import (
     AirbyteAgentsUnavailableError,
+    AirbyteCloudApiError,
     AirbyteError,
     PyAirbyteInputError,
 )
@@ -83,6 +84,16 @@ def _error_guidance(*, response: requests.Response) -> str | None:
             "Authentication succeeded but access was denied; the workspace or connector "
             "may not be enabled for agent access in Airbyte Cloud."
         )
+    if response.status_code in {
+        HTTPStatus.BAD_GATEWAY,
+        HTTPStatus.SERVICE_UNAVAILABLE,
+        HTTPStatus.GATEWAY_TIMEOUT,
+    }:
+        return (
+            "Airbyte Cloud accepted the request but its upstream direct-access "
+            "service failed or timed out. This is not a credentials or permissions "
+            "problem; retry later or contact Airbyte support if it persists."
+        )
     return None
 
 
@@ -94,7 +105,7 @@ def is_not_enabled_error(error: AirbyteError) -> bool:
     Auth, server, and malformed-response failures carry other statuses and must not be
     read as "not enabled".
     """
-    return (error.context or {}).get("status_code") in {
+    return isinstance(error, AirbyteCloudApiError) and error.status_code in {
         HTTPStatus.FORBIDDEN,
         HTTPStatus.NOT_FOUND,
     }
@@ -115,8 +126,8 @@ def make_cloud_agent_request(
     workspace, so no organization header is sent.
 
     Raises `AirbyteAgentsUnavailableError` when the credentials' API roots have no
-    Context layer API, and `AirbyteError` with the status code and response text on
-    non-2xx responses, or when the response is not a JSON object.
+    Context layer API, `AirbyteCloudApiError` with the status code and response text on
+    non-2xx responses, or `AirbyteError` when the response is not a JSON object.
     """
     if not deployment.is_agents_api_available(
         public_api_root=credentials.public_api_root,
@@ -151,7 +162,8 @@ def make_cloud_agent_request(
         timeout=_REQUEST_TIMEOUT_SECONDS,
     )
     if not status_ok(response.status_code):
-        raise AirbyteError(
+        raise AirbyteCloudApiError(
+            status_code=response.status_code,
             message=_error_message(response=response, full_url=full_url),
             guidance=_error_guidance(response=response),
             context={

airbyte/exceptions.py

@@ -585,6 +585,13 @@ class AirbyteAgentsUnavailableError(AirbyteError):
     )
 
 
+@dataclass
+class AirbyteCloudApiError(AirbyteError):
+    """The Airbyte Cloud Config API returned a non-2xx response."""
+
+    status_code: int | None = None
+
+
 @dataclass
 class AirbyteExternalAccessNotEnabledError(PyAirbyteError):
     """The connector is not enabled for external access, so it cannot execute direct actions."""

airbyte/mcp/cloud.py

@@ -66,6 +66,7 @@
 )
 from airbyte.destinations.util import get_noop_destination
 from airbyte.exceptions import (
+    AirbyteCloudApiError,
     AirbyteConnectorNotRegisteredError,
     AirbyteDeferredSetupError,
     AirbyteError,
@@ -174,11 +175,15 @@ def _get_connector_check_message(check_result: CheckResult) -> str | None:
     "query. `enabled_features` is only resolved and returned when this filter is set; "
     "use `direct_access` to find every connector with any external-access feature "
     "enabled and see its full feature list. Omit to list every connector with "
-    "`enabled_features='not_checked'` (no feature check performed)."
+    "`enabled_features='not_checked'` (no feature check performed). Connectors whose "
+    "feature lookup fails are returned with `enabled_features='unknown'` and a "
+    "`warnings` entry, so they can still be inspected or tried."
 )
 
 FEATURES_NOT_CHECKED: Final = "not_checked"
 FeaturesNotChecked = Literal["not_checked"]
+FEATURES_UNKNOWN: Final = "unknown"
+FeaturesUnknown = Literal["unknown"]
 
 
 CONNECTOR_TYPE_TIP_TEXT = (
@@ -1094,9 +1099,15 @@ class CloudConnectorResult(BaseModel):
     """The connector's display name."""
     url: str
     """The connector's page in the Airbyte Cloud UI."""
-    enabled_features: list[ConnectorFeature] | FeaturesNotChecked = FEATURES_NOT_CHECKED
+    enabled_features: list[ConnectorFeature] | FeaturesNotChecked | FeaturesUnknown = (
+        FEATURES_NOT_CHECKED
+    )
     """Features enabled for this connector. `"not_checked"` means no feature check was
-    performed (no `feature_filter`); an empty list means checked with nothing enabled."""
+    performed (no `feature_filter`); `"unknown"` means the feature lookup failed (see
+    `warnings`); an empty list means checked with nothing enabled."""
+
+    warnings: list[str] = Field(default_factory=list)
+    """Non-fatal issues encountered while resolving this connector's features."""
 
 
 @mcp_tool(
@@ -1151,27 +1162,89 @@ def list_cloud_connectors(
     matching connectors with their `enabled_features` resolved; without it,
     `enabled_features` is `"not_checked"`. A returned `"not_checked"` means no
     feature check was performed; `[]` means checked and no features enabled.
+
+    When a connector's feature lookup fails for a reason other than "not enabled",
+    it is returned with `enabled_features="unknown"` plus a `warnings` entry. A
+    502/503/504 or transport failure means the endpoint is down for the whole
+    workspace, so remaining connectors are returned as `"unknown"` without further
+    probes; other failures mark only that connector `"unknown"`. `"unknown"`
+    results can still be inspected or tried via `describe_cloud_connector` or the
+    execute tools.
     """
+    if limit is not None and limit <= 0:
+        raise PyAirbyteInputError(message="`limit` must be greater than 0.")
     workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
     connectors = workspace.list_connectors(
         connector_type=connector_type,
-        feature_filter=feature_filter,
         name_contains=name_contains,
-        limit=limit,
+        limit=None if feature_filter is not None else limit,
     )
-    # Note: name and url are guaranteed non-null from list API responses
-    return [
-        CloudConnectorResult(
-            id=connector.connector_id,
-            connector_type=connector.connector_type.value,
-            name=cast(str, connector.name),
-            url=connector.connector_url,
-            enabled_features=sorted(connector.enabled_features)
-            if feature_filter is not None
-            else FEATURES_NOT_CHECKED,
+    if feature_filter is None:
+        # Note: name and url are guaranteed non-null from list API responses
+        return [
+            CloudConnectorResult(
+                id=connector.connector_id,
+                connector_type=connector.connector_type.value,
+                name=cast(str, connector.name),
+                url=connector.connector_url,
+            )
+            for connector in connectors
+        ]
+
+    results: list[CloudConnectorResult] = []
+    probe_failure: str | None = None
+    for connector in connectors:
+        features: frozenset[ConnectorFeature] | None = None
+        if probe_failure is None:
+            try:
+                features = connector.enabled_features
+            except (AirbyteError, requests.RequestException) as error:
+                warning = f"Connector feature lookup failed; enabled features are unknown: {error}"
+                if isinstance(error, requests.RequestException) or (
+                    isinstance(error, AirbyteCloudApiError)
+                    and error.status_code
+                    in {
+                        HTTPStatus.BAD_GATEWAY,
+                        HTTPStatus.SERVICE_UNAVAILABLE,
+                        HTTPStatus.GATEWAY_TIMEOUT,
+                    }
+                ):
+                    probe_failure = warning
+                else:
+                    results.append(
+                        CloudConnectorResult(
+                            id=connector.connector_id,
+                            connector_type=connector.connector_type.value,
+                            name=cast(str, connector.name),
+                            url=connector.connector_url,
+                            enabled_features=FEATURES_UNKNOWN,
+                            warnings=[warning],
+                        )
+                    )
+                    if limit is not None and len(results) >= limit:
+                        break
+                    continue
+        if probe_failure is not None:
+            enabled_features: list[ConnectorFeature] | FeaturesUnknown = FEATURES_UNKNOWN
+            connector_warnings = [probe_failure]
+        elif features is not None and feature_filter in features:
+            enabled_features = sorted(features)
+            connector_warnings = []
+        else:
+            continue
+        results.append(
+            CloudConnectorResult(
+                id=connector.connector_id,
+                connector_type=connector.connector_type.value,
+                name=cast(str, connector.name),
+                url=connector.connector_url,
+                enabled_features=enabled_features,
+                warnings=connector_warnings,
+            )
         )
-        for connector in connectors
-    ]
+        if limit is not None and len(results) >= limit:
+            break
+    return results
 
 
 class CloudConnectorDetailsResult(BaseModel):
@@ -1200,8 +1273,9 @@ class CloudConnectorDetailsResult(BaseModel):
     integration_name: str | None = None
     """Name of the underlying integration, for example `GitHub` or `Snowflake`."""
 
-    enabled_features: list[ConnectorFeature] = Field(default_factory=list)
-    """Features enabled for this connector; see `ConnectorFeature`."""
+    enabled_features: list[ConnectorFeature] | FeaturesUnknown = Field(default_factory=list)
+    """Features enabled for this connector; see `ConnectorFeature`. `"unknown"` means
+    the feature lookup failed (see `warnings`)."""
 
     config: dict[str, Any] | None = None
     """The connector configuration, populated only by `with_config`.
@@ -1249,18 +1323,22 @@ def _describe_cloud_connector(
         connector_url=connector.connector_url,
         connector_definition_id=connector.definition_id,
         integration_name=integration_name,
-        enabled_features=sorted(connector.enabled_features),
     )
 
-    if (
-        connector.is_feature_enabled(ConnectorFeature.DIRECT_ACCESS)
-        and connector.workspace._has_context_layer_api()  # noqa: SLF001
-    ):
-        context_layer = connector._context_layer_inspect(  # noqa: SLF001
-            warnings=warnings,
-        )
-        if context_layer is not None:
-            warnings.extend(str(warning) for warning in context_layer.warnings)
+    try:
+        result.enabled_features = sorted(connector.enabled_features)
+        if (
+            connector.is_feature_enabled(ConnectorFeature.DIRECT_ACCESS)
+            and connector.workspace._has_context_layer_api()  # noqa: SLF001
+        ):
+            context_layer = connector._context_layer_inspect(  # noqa: SLF001
+                warnings=warnings,
+            )
+            if context_layer is not None:
+                warnings.extend(str(warning) for warning in context_layer.warnings)
+    except (AirbyteError, requests.RequestException) as error:
+        result.enabled_features = FEATURES_UNKNOWN
+        warnings.append(f"Connector feature lookup failed; enabled features are unknown: {error}")
 
     if with_config and connector_type == ConnectorType.DESTINATION:
         try:

tests/unit_tests/test_cloud_credentials.py

@@ -35,6 +35,7 @@
 from airbyte.cloud import workspaces as cloud_workspaces
 from airbyte.cloud.workspaces import CloudWorkspace
 from airbyte.exceptions import (
+    AirbyteCloudApiError,
     AirbyteError,
     AirbyteMissingResourceError,
     PyAirbyteInputError,
@@ -1702,7 +1703,7 @@ def fake_read_cloud_skill_docs(**kwargs: object) -> dict[str, object]:
         skill_id = str(kwargs["skill_id"])
         connector_id = skill_id.rsplit(":", 1)[-1]
         if connector_id not in enabled_ids:
-            raise AirbyteError(context={"status_code": 404})
+            raise AirbyteCloudApiError(status_code=404)
         return {
             "metadata": {
                 "id": skill_id,

tests/unit_tests/test_cloud_describe.py

@@ -33,6 +33,7 @@
 )
 from airbyte.cloud.workspaces import CloudWorkspace
 from airbyte.exceptions import (
+    AirbyteCloudApiError,
     AirbyteError,
     AirbyteExternalAccessNotEnabledError,
     PyAirbyteInputError,
@@ -259,8 +260,8 @@ def test_direct_access_guidance_id_source_without_context_layer(
 @pytest.mark.parametrize(
     "probe_error",
     [
-        pytest.param(AirbyteError(context={"status_code": 404}), id="not_found"),
-        pytest.param(AirbyteError(context={"status_code": 403}), id="forbidden"),
+        pytest.param(AirbyteCloudApiError(status_code=404), id="not_found"),
+        pytest.param(AirbyteCloudApiError(status_code=403), id="forbidden"),
     ],
 )
 def test_direct_access_guidance_id_source_probe_not_enabled(
@@ -351,8 +352,8 @@ def test_get_direct_access_guidance_non_passthrough_destination_raises(
 @pytest.mark.parametrize(
     "probe_error",
     [
-        pytest.param(AirbyteError(context={"status_code": 404}), id="not_found"),
-        pytest.param(AirbyteError(context={"status_code": 403}), id="forbidden"),
+        pytest.param(AirbyteCloudApiError(status_code=404), id="not_found"),
+        pytest.param(AirbyteCloudApiError(status_code=403), id="forbidden"),
     ],
 )
 def test_get_direct_access_guidance_destination_not_enabled_adds_notice(
@@ -705,7 +706,7 @@ def test_enabled_features_disabled_connector(
     monkeypatch.setattr(
         agents_api_util,
         "read_cloud_skill_docs",
-        lambda **_: (_ for _ in ()).throw(AirbyteError(context={"status_code": 404})),
+        lambda **_: (_ for _ in ()).throw(AirbyteCloudApiError(status_code=404)),
     )
     source = _seed_source(workspace, "source-1", "GitHub")