feat(cloud): return a link to enable direct access when it's disabled

airbyte/_direct_connectors/models.py

@@ -230,6 +230,9 @@ class ExternalApiExecuteResult(BaseModel):
     warning: dict[str, Any] | None = None
     """A warning reported alongside an otherwise successful result."""
 
+    errors: list[str] = Field(default_factory=list)
+    """Agent-facing error messages; when non-empty, show them to the user."""
+
     @field_validator("connector_metadata", "execution_metadata", mode="before")
     @classmethod
     def _none_to_empty(cls, value: object) -> object:

airbyte/cloud/connectors.py

@@ -626,6 +626,7 @@ def _raise_if_feature_not_enabled(
         feature: ConnectorFeature,
         *,
         guidance: str | None = None,
+        enablement_url: str | None = None,
     ) -> None:
         """Map a forbidden or not-found `error` to a clear error, when one applies.
 
@@ -668,8 +669,15 @@ def _raise_if_feature_not_enabled(
         )
         if guidance is not None:
             not_enabled.guidance = guidance
+        if enablement_url is not None:
+            not_enabled.enablement_url = enablement_url
         raise not_enabled from error
 
+    @property
+    def _direct_access_enablement_url(self) -> str:
+        """Get the web URL where the user can enable direct access for this connector."""
+        return self.connector_url
+
     def _execute_direct_action(  # noqa: PLR0913  # Explicit args mirror the public methods.
         self,
         *,
@@ -732,7 +740,11 @@ def _execute_direct_action(  # noqa: PLR0913  # Explicit args mirror the public
                 credentials=self.workspace._credentials,  # noqa: SLF001
             )
         except exc.AirbyteCloudError as error:
-            self._raise_if_feature_not_enabled(error, ConnectorFeature.DIRECT_ACCESS)
+            self._raise_if_feature_not_enabled(
+                error,
+                ConnectorFeature.DIRECT_ACCESS,
+                enablement_url=self._direct_access_enablement_url,
+            )
             raise
 
         return response

airbyte/exceptions.py

@@ -623,6 +623,8 @@ class AirbyteExternalAccessNotEnabledError(AirbyteCloudError):
         "organization's Context Layer settings. Check `enabled_features` on the connector "
         "and organization."
     )
+    enablement_url: str | None = None
+    """The web URL where the user can enable direct access."""
 
 
 @dataclass

airbyte/mcp/cloud.py

@@ -70,6 +70,7 @@
     AirbyteCloudError,
     AirbyteConnectorNotRegisteredError,
     AirbyteDeferredSetupError,
+    AirbyteExternalAccessNotEnabledError,
     AirbyteLibError,
     AirbyteLibInputError,
     AirbyteMissingResourceError,
@@ -1542,6 +1543,21 @@ def describe_cloud_connector(
     )
 
 
+def _direct_access_disabled_result(
+    error: AirbyteExternalAccessNotEnabledError,
+    connector: CloudConnector,
+) -> ExternalApiExecuteResult:
+    """Return user-facing direct-access guidance as an MCP result error."""
+    label = error.connector_name or f"{connector.connector_type} {connector.connector_id}"
+    return ExternalApiExecuteResult(
+        status="error",
+        errors=[
+            f"Show the user this message as-is: Direct access is currently disabled for "
+            f"{label}. [Click here]({error.enablement_url}) to enable it."
+        ],
+    )
+
+
 @mcp_tool(
     read_only=True,
     idempotent=True,
@@ -1646,17 +1662,22 @@ def execute_external_api_query(  # noqa: PLR0913  # Explicit args mirror the con
     connector's action docs.
     """
     connector = _get_cloud_workspace(ctx, workspace_id).get_connector(connector_id)
-    return connector.execute_api_query(
-        entity_type,
-        action,
-        resolve_api_args(api_args),
-        select_fields=resolve_list_of_strings(select_fields),
-        exclude_fields=resolve_list_of_strings(exclude_fields),
-        page_size=page_size,
-        cursor=cursor,
-        skip_truncation=skip_truncation,
-        intent=intent,
-    )
+    try:
+        return connector.execute_api_query(
+            entity_type,
+            action,
+            resolve_api_args(api_args),
+            select_fields=resolve_list_of_strings(select_fields),
+            exclude_fields=resolve_list_of_strings(exclude_fields),
+            page_size=page_size,
+            cursor=cursor,
+            skip_truncation=skip_truncation,
+            intent=intent,
+        )
+    except AirbyteExternalAccessNotEnabledError as error:
+        if error.enablement_url is None:
+            raise
+        return _direct_access_disabled_result(error, connector)
 
 
 # Not yet registered as an MCP tool: write actions are not supported by the backend.
@@ -1739,15 +1760,20 @@ def _execute_external_api_action(  # noqa: PLR0913  # Explicit args mirror the c
     differ per connector and are not guessable.
     """
     connector = _get_cloud_workspace(ctx, workspace_id).get_connector(connector_id)
-    return connector.execute_api_action(
-        entity_type,
-        action,
-        resolve_api_args(api_args),
-        select_fields=resolve_list_of_strings(select_fields),
-        exclude_fields=resolve_list_of_strings(exclude_fields),
-        skip_truncation=skip_truncation,
-        intent=intent,
-    )
+    try:
+        return connector.execute_api_action(
+            entity_type,
+            action,
+            resolve_api_args(api_args),
+            select_fields=resolve_list_of_strings(select_fields),
+            exclude_fields=resolve_list_of_strings(exclude_fields),
+            skip_truncation=skip_truncation,
+            intent=intent,
+        )
+    except AirbyteExternalAccessNotEnabledError as error:
+        if error.enablement_url is None:
+            raise
+        return _direct_access_disabled_result(error, connector)
 
 
 @mcp_tool(
@@ -1826,13 +1852,18 @@ def execute_external_sql_query(
     identifiers: double-quoting makes them case-sensitive.
     """
     connector = _get_cloud_workspace(ctx, workspace_id).get_connector(connector_id)
-    return connector.execute_sql_query(
-        sql,
-        sql_dialect=sql_dialect,
-        page_size=page_size,
-        cursor=cursor,
-        dry_run=dry_run,
-    )
+    try:
+        return connector.execute_sql_query(
+            sql,
+            sql_dialect=sql_dialect,
+            page_size=page_size,
+            cursor=cursor,
+            dry_run=dry_run,
+        )
+    except AirbyteExternalAccessNotEnabledError as error:
+        if error.enablement_url is None:
+            raise
+        return _direct_access_disabled_result(error, connector)
 
 
 @mcp_tool(

tests/unit_tests/test_cloud_direct_execute.py

@@ -342,13 +342,40 @@ def test_execute_error_handling(
         assert isinstance(exc_info.value, AirbyteExternalAccessNotEnabledError)
         assert exc_info.value.connector_id == "source-1"
         assert exc_info.value.connector_name == "GitHub Issues"
+        assert exc_info.value.enablement_url == source.connector_url
+        assert exc_info.value.guidance == AirbyteExternalAccessNotEnabledError.guidance
     else:
         assert isinstance(exc_info.value, AirbyteCloudError)
         assert not isinstance(exc_info.value, AirbyteExternalAccessNotEnabledError)
         assert isinstance(exc_info.value, AirbyteCloudApiError)
         assert exc_info.value.status_code == expected_status
 
 
+def test_execute_error_sets_enablement_url_without_connector_info(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    workspace = _make_workspace(monkeypatch)
+    _patch_context_layer(monkeypatch)
+    _patch_execute(
+        monkeypatch,
+        {},
+        error=AirbyteCloudApiError(
+            status_code=403,
+            context={"status_code": 403},
+        ),
+    )
+    source = CloudSource(workspace=workspace, connector_id="source-1")
+
+    with (
+        _patch_direct_access(enabled=False),
+        pytest.raises(AirbyteExternalAccessNotEnabledError) as exc_info,
+    ):
+        source.execute_api_query("issues")
+
+    assert exc_info.value.enablement_url == source.connector_url
+    assert exc_info.value.guidance == AirbyteExternalAccessNotEnabledError.guidance
+
+
 @pytest.mark.parametrize("method_name", ["execute_api_query", "execute_api_action"])
 def test_direct_methods_raise_without_context_layer(
     monkeypatch: pytest.MonkeyPatch,

tests/unit_tests/test_mcp_cloud.py

@@ -49,6 +49,7 @@
     AirbyteCloudApiError,
     AirbyteConnectorInUseError,
     AirbyteCloudError,
+    AirbyteExternalAccessNotEnabledError,
     AirbyteMissingResourceError,
     AirbyteLibInputError,
 )
@@ -1623,9 +1624,18 @@ def test_get_agent_skill_docs_tool_connector_id(
 class _RecordingExecuteConnector:
     """Fake connector recording `execute_api_*`/`execute_sql_query` calls."""
 
-    def __init__(self) -> None:
+    def __init__(
+        self,
+        *,
+        error: AirbyteExternalAccessNotEnabledError | None = None,
+        connector_id: str = "source-1",
+        connector_type: ConnectorType = ConnectorType.SOURCE,
+    ) -> None:
         self.calls: list[tuple[str, dict[str, object]]] = []
         self.result = ExternalApiExecuteResult(status="success", result={"ok": True})
+        self.error = error
+        self.connector_id = connector_id
+        self.connector_type = connector_type
 
     def execute_api_query(
         self, entity_type: str, action: object, api_args: object, **kwargs: object
@@ -1639,6 +1649,8 @@ def execute_api_query(
                 **kwargs,
             },
         ))
+        if self.error is not None:
+            raise self.error
         return self.result
 
     def execute_api_action(
@@ -1653,10 +1665,14 @@ def execute_api_action(
                 **kwargs,
             },
         ))
+        if self.error is not None:
+            raise self.error
         return self.result
 
     def execute_sql_query(self, sql: str, **kwargs: object) -> object:
         self.calls.append(("sql", {"sql": sql, **kwargs}))
+        if self.error is not None:
+            raise self.error
         return self.result
 
     def execute_search_query(self, prompt: str, **kwargs: object) -> object:
@@ -1728,6 +1744,86 @@ def test_execute_external_api_query_forwards_kwargs(
     assert result.status == "success"
 
 
+def test_execute_external_api_query_returns_direct_access_error_result(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The tool returns the cached connector name and enablement link in errors."""
+    error = AirbyteExternalAccessNotEnabledError(
+        connector_id="source-1",
+        connector_name="GitHub Issues",
+        enablement_url="https://cloud.airbyte.com/workspaces/workspace-1/source/source-1",
+    )
+    connector = _RecordingExecuteConnector(error=error)
+    _execute_workspace(monkeypatch, connector)
+
+    result = cloud_mcp.execute_external_api_query(
+        None,
+        connector_id="source-1",
+        entity_type="issues",
+        workspace_id=None,
+    )
+
+    assert result.status == "error"
+    assert result.errors == [
+        "Show the user this message as-is: Direct access is currently disabled for "
+        "GitHub Issues. [Click here](https://cloud.airbyte.com/workspaces/workspace-1/"
+        "source/source-1) to enable it."
+    ]
+
+
+def test_execute_external_sql_query_uses_fallback_direct_access_label(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The SQL tool falls back to connector type and ID when its name is unavailable."""
+    error = AirbyteExternalAccessNotEnabledError(
+        connector_id="destination-1",
+        enablement_url=(
+            "https://cloud.airbyte.com/workspaces/workspace-1/destination/destination-1"
+        ),
+    )
+    connector = _RecordingExecuteConnector(
+        error=error,
+        connector_id="destination-1",
+        connector_type=ConnectorType.DESTINATION,
+    )
+    _execute_workspace(monkeypatch, connector)
+
+    result = cloud_mcp.execute_external_sql_query(
+        None,
+        connector_id="destination-1",
+        sql="SHOW TABLES",
+        workspace_id=None,
+    )
+
+    assert result.status == "error"
+    assert result.errors == [
+        "Show the user this message as-is: Direct access is currently disabled for "
+        "destination destination-1. [Click here](https://cloud.airbyte.com/workspaces/"
+        "workspace-1/destination/destination-1) to enable it."
+    ]
+
+
+def test_execute_external_api_action_reraises_without_enablement_url(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The action tool re-raises errors that lack an enablement URL."""
+    error = AirbyteExternalAccessNotEnabledError(connector_id="source-1")
+    connector = _RecordingExecuteConnector(error=error)
+    _execute_workspace(monkeypatch, connector)
+
+    with pytest.raises(AirbyteExternalAccessNotEnabledError) as exc_info:
+        cloud_mcp._execute_external_api_action(
+            None,
+            connector_id="source-1",
+            entity_type="issues",
+            action=ExternalApiWriteAction.CREATE,
+            api_args={"title": "Bug"},
+            workspace_id=None,
+        )
+
+    assert exc_info.value is error
+
+
 def test_execute_external_api_query_rejects_non_object_api_args(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None: