fix(cloud): return actionable errors when creating a connection fails

airbyte/_util/api_util.py

@@ -171,6 +171,19 @@ def _wrap_sdk_error(
     )
 
 
+def _get_sdk_error_detail(error: SDKError) -> str | None:
+    """Return the human-readable message from an Airbyte API error body, if present."""
+    try:
+        body = json.loads(error.body)
+    except (json.JSONDecodeError, TypeError):
+        return None
+    if not isinstance(body, dict):
+        return None
+    data = body.get("data")
+    candidates = (data.get("message") if isinstance(data, dict) else None, body.get("detail"))
+    return next((c for c in candidates if isinstance(c, str) and c), None)
+
+
 def _infer_config_api_root(api_root: str) -> str | None:
     """Infer the configuration API root from a public API root."""
     normalized_api_root = api_root.rstrip("/")
@@ -1906,15 +1919,33 @@ def create_connection(  # noqa: PLR0913  # Too many arguments
         api_root=api_root,
     )
     stream_configurations_obj = build_stream_configurations(selected_stream_names)
-    response = airbyte_instance.connections.create_connection(
-        models.ConnectionCreateRequest(
-            name=name,
-            source_id=source_id,
-            destination_id=destination_id,
-            configurations=stream_configurations_obj,
-            prefix=prefix,
-        ),
-    )
+    try:
+        response = airbyte_instance.connections.create_connection(
+            models.ConnectionCreateRequest(
+                name=name,
+                source_id=source_id,
+                destination_id=destination_id,
+                configurations=stream_configurations_obj,
+                prefix=prefix,
+            ),
+        )
+    except SDKError as e:
+        base_context = {
+            "source_id": source_id,
+            "destination_id": destination_id,
+            "selected_stream_names": selected_stream_names,
+        }
+        if e.status_code == HTTPStatus.BAD_REQUEST:
+            raise AirbyteLibInputError(
+                message=_get_sdk_error_detail(e)
+                or "The Airbyte API rejected the connection request (HTTP 400).",
+                guidance=(
+                    "Check that every name in `selected_streams` is a stream of the "
+                    "source, and that `source_id` and `destination_id` are correct."
+                ),
+                context={**base_context, **_get_sdk_error_context(e)},
+            ) from e
+        raise _wrap_sdk_error(e, base_context) from e
     if not status_ok(response.status_code) or response.connection_response is None:
         raise AirbyteCloudError(
             context={

airbyte/mcp/cloud.py

@@ -754,17 +754,15 @@ def _deploy_deferred_to_cloud(
 
 
 def _get_suggested_streams_for_source(
-    workspace: CloudWorkspace,
-    source_id: str,
+    source: CloudSource,
 ) -> list[str]:
     """Resolve the source connector's suggested streams from the connector registry."""
-    source = workspace.get_source(source_id)
     metadata = get_connector_metadata_by_definition_id(source.definition_id)
     suggested_streams = metadata.suggested_streams if metadata else None
     if not suggested_streams:
         raise AirbyteLibInputError(
             message=(
-                f"No `selected_streams` provided and source '{source_id}' has no "
+                f"No `selected_streams` provided and source '{source.connector_id}' has no "
                 "suggested streams in the connector registry."
             ),
             guidance=(
@@ -834,9 +832,26 @@ def create_connection_on_cloud(
     """
     resolved_streams_list: list[str] = resolve_list_of_strings(selected_streams) or []
     workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
+
+    source = workspace.get_source(source_id)
+    destination = workspace.get_destination(destination_id)
+    for arg_name, connector in (("source_id", source), ("destination_id", destination)):
+        try:
+            _ = connector.name  # Forces the connector info fetch; objects are lazy.
+        except AirbyteMissingResourceError as ex:
+            raise AirbyteMissingResourceError(
+                message=(
+                    f"`{arg_name}` '{connector.connector_id}' was not found in workspace "
+                    f"'{workspace.workspace_id}', or these credentials can't access it."
+                ),
+                guidance=api_util.FORBIDDEN_RESOURCE_GUIDANCE,
+                resource_type=arg_name.removesuffix("_id"),
+                resource_name_or_id=connector.connector_id,
+            ) from ex
+
     used_suggested_streams = False
     if not resolved_streams_list:
-        resolved_streams_list = _get_suggested_streams_for_source(workspace, source_id)
+        resolved_streams_list = _get_suggested_streams_for_source(source)
         used_suggested_streams = True
 
     deployed_connection = workspace.deploy_connection(

tests/unit_tests/test_cloud_api_util.py

@@ -2243,6 +2243,133 @@ def test_api_util_calls_wrap_sdk_errors_with_status_context(
         get_job.assert_not_called()
 
 
+def _call_create_connection() -> None:
+    """Invoke `create_connection` with canned IDs for SDKError tests."""
+    api_util.create_connection(
+        "connection-name",
+        source_id="source-id",
+        destination_id="destination-id",
+        api_root="https://api.airbyte.com/v1",
+        client_id=None,
+        client_secret=None,
+        bearer_token=SecretString("token"),
+        prefix="",
+        selected_stream_names=["no_such_stream"],
+    )
+
+
+def test_create_connection_400_raises_input_error_with_api_message(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A 400 surfaces the API's `data.message` as an `AirbyteLibInputError`."""
+    api_message = (
+        "No streams found with name [no_such_stream] and namespace [null]. "
+        "Please call the https://reference.airbyte.com/reference/getstreamproperties "
+        "endpoint to see the valid stream name and namespace combinations."
+    )
+    body = json.dumps({
+        "status": 400,
+        "type": "https://reference.airbyte.com/reference/errors#bad-request",
+        "title": "bad-request",
+        "detail": "The request could not be understood by the server "
+        "due to malformed syntax.",
+        "documentationUrl": None,
+        "data": {"message": api_message},
+    })
+    raw_response = requests.Response()
+    raw_response.status_code = 400
+    raw_response.url = "https://api.airbyte.com/v1/connections"
+    sdk_error = SDKError("API error occurred: Status 400", 400, body, raw_response)
+    create_connection = Mock(side_effect=sdk_error)
+    airbyte_instance = SimpleNamespace(
+        connections=SimpleNamespace(create_connection=create_connection),
+    )
+    monkeypatch.setattr(
+        api_util,
+        "get_airbyte_server_instance",
+        lambda **_: airbyte_instance,
+    )
+
+    with pytest.raises(AirbyteLibInputError) as exc_info:
+        _call_create_connection()
+
+    assert exc_info.value.get_message() == api_message
+    assert exc_info.value.context["source_id"] == "source-id"
+    assert exc_info.value.context["destination_id"] == "destination-id"
+    assert exc_info.value.context["status_code"] == 400
+    assert exc_info.value.__cause__ is sdk_error
+
+
+def test_create_connection_400_with_non_json_body_uses_fallback_message(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A 400 with an unparseable body falls back to a generic input error."""
+    raw_response = requests.Response()
+    raw_response.status_code = 400
+    raw_response.url = "https://api.airbyte.com/v1/connections"
+    sdk_error = SDKError(
+        "API error occurred: Status 400",
+        400,
+        "<html>Bad Request</html>",
+        raw_response,
+    )
+    airbyte_instance = SimpleNamespace(
+        connections=SimpleNamespace(
+            create_connection=Mock(side_effect=sdk_error),
+        ),
+    )
+    monkeypatch.setattr(
+        api_util,
+        "get_airbyte_server_instance",
+        lambda **_: airbyte_instance,
+    )
+
+    with pytest.raises(AirbyteLibInputError) as exc_info:
+        _call_create_connection()
+
+    assert exc_info.value.get_message() == (
+        "The Airbyte API rejected the connection request (HTTP 400)."
+    )
+    assert exc_info.value.__cause__ is sdk_error
+
+
+@pytest.mark.parametrize(
+    ("status_code", "expect_missing_resource"),
+    [
+        pytest.param(403, True, id="403-missing-resource"),
+        pytest.param(500, False, id="500-cloud-error"),
+    ],
+)
+def test_create_connection_non_400_sdk_errors_keep_status_context(
+    monkeypatch: pytest.MonkeyPatch,
+    status_code: int,
+    expect_missing_resource: bool,
+) -> None:
+    """Non-400 SDK errors go through `_wrap_sdk_error` with the create context."""
+    sdk_error = _sdk_status_error(status_code)
+    airbyte_instance = SimpleNamespace(
+        connections=SimpleNamespace(
+            create_connection=Mock(side_effect=sdk_error),
+        ),
+    )
+    monkeypatch.setattr(
+        api_util,
+        "get_airbyte_server_instance",
+        lambda **_: airbyte_instance,
+    )
+
+    with pytest.raises(AirbyteCloudError) as exc_info:
+        _call_create_connection()
+
+    assert isinstance(exc_info.value, AirbyteMissingResourceError) == (
+        expect_missing_resource
+    )
+    assert exc_info.value.context["status_code"] == status_code
+    assert exc_info.value.context["source_id"] == "source-id"
+    assert exc_info.value.context["destination_id"] == "destination-id"
+    assert exc_info.value.__cause__ is sdk_error
+
+
 @pytest.mark.parametrize(
     "source_status",
     [

tests/unit_tests/test_mcp_cloud.py

@@ -1194,17 +1194,46 @@ def test_deploy_connector_to_cloud_rejects_unknown_prefix() -> None:
         )
 
 
+class _LazyConnector:
+    """Connector double whose `name` raises `AirbyteMissingResourceError` when missing."""
+
+    def __init__(self, connector_id: str, *, missing: bool = False) -> None:
+        self.connector_id = connector_id
+        self.definition_id = "definition-id"
+        self._missing = missing
+
+    @property
+    def name(self) -> str:
+        """Return the connector name, or raise when the connector is marked missing."""
+        if self._missing:
+            raise AirbyteMissingResourceError(
+                resource_type="connector",
+                resource_name_or_id=self.connector_id,
+            )
+        return "connector-name"
+
+
 class _DeployConnectionWorkspace:
-    """Fake `CloudWorkspace` capturing `deploy_connection` and `get_source` calls."""
+    """Fake `CloudWorkspace` capturing `deploy_connection` and connector lookups."""
 
-    def __init__(self) -> None:
+    def __init__(self, missing_ids: set[str] | None = None) -> None:
         self.deploy_calls: list[dict[str, object]] = []
         self.get_source_calls: list[str] = []
+        self.get_destination_calls: list[str] = []
+        self.workspace_id = "workspace-id"
+        self._missing_ids = missing_ids or set()
 
-    def get_source(self, source_id: str) -> SimpleNamespace:
+    def get_source(self, source_id: str) -> _LazyConnector:
         """Return a source double exposing the connector definition ID."""
         self.get_source_calls.append(source_id)
-        return SimpleNamespace(definition_id="definition-id")
+        return _LazyConnector(source_id, missing=source_id in self._missing_ids)
+
+    def get_destination(self, destination_id: str) -> _LazyConnector:
+        """Return a destination double."""
+        self.get_destination_calls.append(destination_id)
+        return _LazyConnector(
+            destination_id, missing=destination_id in self._missing_ids
+        )
 
     def deploy_connection(self, **kwargs: object) -> SimpleNamespace:
         """Capture the deploy request and return a connection double."""
@@ -1217,8 +1246,9 @@ def deploy_connection(self, **kwargs: object) -> SimpleNamespace:
 
 def _patch_deploy_connection_workspace(
     monkeypatch: pytest.MonkeyPatch,
+    missing_ids: set[str] | None = None,
 ) -> _DeployConnectionWorkspace:
-    workspace = _DeployConnectionWorkspace()
+    workspace = _DeployConnectionWorkspace(missing_ids=missing_ids)
     monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_: workspace)
     monkeypatch.setattr(
         cloud_mcp, "register_guid_created_in_session", lambda *args: None
@@ -1249,7 +1279,8 @@ def fail_lookup(_definition_id: str) -> None:
         table_prefix=None,
     )
 
-    assert workspace.get_source_calls == []
+    assert workspace.get_source_calls == ["source-id"]
+    assert workspace.get_destination_calls == ["destination-id"]
     assert workspace.deploy_calls == [
         {
             "connection_name": "My Connection",
@@ -1285,6 +1316,7 @@ def test_create_connection_on_cloud_defaults_to_suggested_streams(
     )
 
     assert workspace.get_source_calls == ["source-id"]
+    assert workspace.get_destination_calls == ["destination-id"]
     assert workspace.deploy_calls[0]["selected_streams"] == ["users"]
     assert "suggested streams" in result
 
@@ -1322,6 +1354,50 @@ def test_create_connection_on_cloud_without_suggested_streams_raises(
     assert workspace.deploy_calls == []
 
 
+def test_create_connection_on_cloud_unknown_destination_names_arg(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An unknown destination raises `AirbyteMissingResourceError` naming `destination_id`."""
+    workspace = _patch_deploy_connection_workspace(
+        monkeypatch, missing_ids={"destination-id"}
+    )
+
+    with pytest.raises(AirbyteMissingResourceError, match="`destination_id`"):
+        cloud_mcp.create_connection_on_cloud(
+            cast(Context, object()),
+            connection_name="My Connection",
+            source_id="source-id",
+            destination_id="destination-id",
+            workspace_id=None,
+            selected_streams=["users"],
+            table_prefix=None,
+        )
+
+    assert workspace.deploy_calls == []
+
+
+def test_create_connection_on_cloud_unknown_source_names_arg(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An unknown source raises `AirbyteMissingResourceError` naming `source_id`."""
+    workspace = _patch_deploy_connection_workspace(
+        monkeypatch, missing_ids={"source-id"}
+    )
+
+    with pytest.raises(AirbyteMissingResourceError, match="`source_id`"):
+        cloud_mcp.create_connection_on_cloud(
+            cast(Context, object()),
+            connection_name="My Connection",
+            source_id="source-id",
+            destination_id="destination-id",
+            workspace_id=None,
+            selected_streams=["users"],
+            table_prefix=None,
+        )
+
+    assert workspace.deploy_calls == []
+
+
 class _CombinedListingWorkspace:
     """Fake `CloudWorkspace` returning one source and one destination."""