feat(cloud): name conflicting destination streams and point to table_prefix on connection create

airbyte/_util/api_util.py

@@ -26,7 +26,11 @@
 from airbyte_api import api, models
 from airbyte_api.errors import SDKError
 
-from airbyte._util.cloud_errors import CloudErrorInfo, parse_cloud_error
+from airbyte._util.cloud_errors import (
+    CloudErrorInfo,
+    parse_cloud_error,
+    parse_problem_payload,
+)
 from airbyte._util.meta import (
     AIRBYTE_CLOUD_ANALYTIC_SOURCE_HEADER,
     get_cloud_api_analytic_source,
@@ -1989,6 +1993,80 @@ def build_connection_schedule(
     )
 
 
+_MAX_CONFLICTING_DESTINATION_STREAMS = 10
+
+
+def _format_conflicting_destination_stream(stream: object) -> str | None:
+    if not isinstance(stream, dict):
+        return None
+    name = stream.get("streamName")
+    namespace = stream.get("streamNamespace", "")
+    connection_ids = stream.get("connectionIds")
+    if not isinstance(name, str) or not name:
+        return None
+    if namespace is None:
+        namespace = ""
+    if not isinstance(namespace, str):
+        return None
+    if not isinstance(connection_ids, list) or not connection_ids:
+        return None
+    if any(not isinstance(connection_id, str) for connection_id in connection_ids):
+        return None
+
+    stream_name = f"{namespace}.{name}" if namespace else name
+    rendered_ids = ", ".join(f"`{connection_id}`" for connection_id in connection_ids)
+    return f"`{stream_name}` (connection {rendered_ids})"
+
+
+def _format_conflicting_destination_streams(body: str) -> str | None:
+    payload = parse_problem_payload(body)
+    if payload is None:
+        return None
+    data = payload.get("data")
+    streams = data.get("streams") if isinstance(data, dict) else None
+    if not isinstance(streams, list) or not streams:
+        return None
+
+    formatted_streams: list[str] = []
+    for stream in streams:
+        formatted_stream = _format_conflicting_destination_stream(stream)
+        if formatted_stream is not None:
+            formatted_streams.append(formatted_stream)
+    if not formatted_streams:
+        return None
+
+    stream_list = ", ".join(formatted_streams[:_MAX_CONFLICTING_DESTINATION_STREAMS])
+    if len(formatted_streams) > _MAX_CONFLICTING_DESTINATION_STREAMS:
+        remaining_stream_count = len(formatted_streams) - _MAX_CONFLICTING_DESTINATION_STREAMS
+        stream_list = f"{stream_list}, and {remaining_stream_count} more"
+    return stream_list
+
+
+def _conflicting_destination_stream_guidance(error: SDKError) -> str | None:
+    """Explain how to avoid destination table collisions for a known Cloud problem."""
+    body = error_response_body(error)
+    problem = parse_cloud_error(error.status_code, body)
+    if problem.slug != "error:connection-conflicting-destination-stream":
+        return None
+
+    generic_guidance = (
+        "Another connection on this destination may already write to the same tables. "
+        "Set `table_prefix` (for example `<source>_`) so this connection writes to "
+        "different table names."
+    )
+    if body is None:
+        return generic_guidance
+    stream_list = _format_conflicting_destination_streams(body)
+    if stream_list is None:
+        return generic_guidance
+
+    return (
+        "Another connection on this destination already writes to these tables: "
+        f"{stream_list}. Set `table_prefix` (for example `<source>_`) so this "
+        "connection writes to different table names."
+    )
+
+
 def create_connection(  # noqa: PLR0913  # Too many arguments
     name: str,
     *,
@@ -2029,6 +2107,7 @@ def create_connection(  # noqa: PLR0913  # Too many arguments
         if e.status_code == HTTPStatus.BAD_REQUEST:
             raise AirbyteLibInputError(
                 message=sdk_error_message(e),
+                guidance=_conflicting_destination_stream_guidance(e),
                 context={**base_context, **_get_sdk_error_context(e)},
             ) from e
         raise _wrap_sdk_error(e, base_context) from e

airbyte/_util/cloud_errors.py

@@ -83,22 +83,32 @@ def _reject_json_constant(value: str) -> NoReturn:
     raise json.JSONDecodeError("Invalid JSON constant", value, 0)
 
 
+def parse_problem_payload(body: str | None) -> dict[str, Any] | None:
+    """Parse a JSON problem body no larger than 64 KiB."""
+    if not isinstance(body, str) or len(body) > _MAX_PROBLEM_BODY:
+        return None
+    try:
+        if len(body.encode("utf-8")) > _MAX_PROBLEM_BODY:
+            return None
+        # Numeric fields are unused; avoid Python's integer-conversion size limit.
+        payload = json.loads(
+            body,
+            parse_int=lambda _: None,
+            parse_constant=_reject_json_constant,
+        )
+        if isinstance(payload, dict):
+            # Surface unpaired-surrogate failures here, not in callers.
+            _problem_message(payload)
+    except (UnicodeEncodeError, json.JSONDecodeError, RecursionError):
+        return None
+    return payload if isinstance(payload, dict) else None
+
+
 def parse_cloud_error(status_code: int | None, body: str | None) -> CloudErrorInfo:
     """Select detail/title from a JSON object of at most 64 KiB, or a status fallback."""
-    payload: Any = None
-    message = None
-    if isinstance(body, str) and len(body) <= _MAX_PROBLEM_BODY:
-        try:
-            if len(body.encode("utf-8")) <= _MAX_PROBLEM_BODY:
-                # Numeric fields are unused; avoid Python's integer-conversion size limit.
-                payload = json.loads(
-                    body, parse_int=lambda _: None, parse_constant=_reject_json_constant
-                )
-                if isinstance(payload, dict):
-                    message = _problem_message(payload)
-        except (UnicodeEncodeError, json.JSONDecodeError, RecursionError):
-            payload = None
-    problem = payload if isinstance(payload, dict) else {}
+    payload = parse_problem_payload(body)
+    message = _problem_message(payload) if payload is not None else None
+    problem = payload if payload is not None else {}
     problem_type = problem.get("type")
     return CloudErrorInfo(
         status_code=status_code,

airbyte/cloud/workspaces.py

@@ -997,7 +997,9 @@ def deploy_connection(
             source: The deployed source. You can pass a source ID or a CloudSource object.
             destination: The deployed destination. You can pass a destination ID or a
                 CloudDestination object.
-            table_prefix: Optional. The table prefix to use when syncing to the destination.
+            table_prefix: Optional prefix for destination table names. Use it to keep
+                connections to the same destination from writing to the same table
+                (namespace + name).
             selected_streams: The selected stream names to sync within the connection.
         """
         if not selected_streams:

airbyte/mcp/cloud.py

@@ -908,7 +908,9 @@ def create_connection_on_cloud(
     table_prefix: Annotated[
         str | None,
         Field(
-            description="Optional table prefix to use when syncing to the destination.",
+            description="Optional prefix for destination table names. Two connections to "
+            "the same destination cannot write to the same table (namespace + name); set "
+            "a prefix such as `<source>_` to keep them separate.",
             default=None,
         ),
         TraceArg.FINGERPRINT,
@@ -922,6 +924,9 @@ def create_connection_on_cloud(
     When `selected_streams` is not provided, the connection selects the source
     connector's suggested streams from the connector registry. If the connector has
     no suggested streams, the call fails with guidance to pass streams explicitly.
+
+    Two connections to the same destination cannot write to the same table (namespace +
+    name). Use `table_prefix` (for example `<source>_`) to keep them separate.
     """
     resolved_streams_list: list[str] = resolve_list_of_strings(selected_streams) or []
     workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)

tests/unit_tests/test_cloud_api_util.py

@@ -27,6 +27,7 @@
     AirbyteWorkspaceNotEmptyError,
     AirbyteLibInputError,
 )
+from airbyte.mcp._error_handling import format_user_facing_error
 from airbyte.registry import ConnectorType
 from airbyte.secrets.base import SecretString
 from airbyte_api import api, models, utils
@@ -2291,23 +2292,113 @@ def _call_create_connection() -> None:
     )
 
 
-def test_create_connection_400_raises_input_error_without_api_message(
+@pytest.mark.parametrize(
+    ("problem_type", "data", "expected_guidance"),
+    [
+        pytest.param(
+            "https://reference.airbyte.com/reference/errors"
+            "#error:connection-conflicting-destination-stream",
+            {
+                "streams": [
+                    {
+                        "streamName": "users",
+                        "connectionIds": ["users-connection"],
+                    },
+                    {
+                        "streamName": "orders",
+                        "streamNamespace": "",
+                        "connectionIds": ["orders-connection"],
+                    },
+                ]
+            },
+            "Another connection on this destination already writes to these tables: "
+            "`users` (connection `users-connection`), `orders` "
+            "(connection `orders-connection`). Set `table_prefix` (for example "
+            "`<source>_`) so this connection writes to different table names.",
+            id="conflicting-streams",
+        ),
+        pytest.param(
+            "https://reference.airbyte.com/reference/errors"
+            "#error:connection-conflicting-destination-stream",
+            {
+                "streams": [
+                    {
+                        "streamName": f"stream-{index}",
+                        "streamNamespace": "",
+                        "connectionIds": [f"connection-{index}"],
+                    }
+                    for index in range(12)
+                ]
+            },
+            (
+                "Another connection on this destination already writes to these tables: "
+                + ", ".join(
+                    f"`stream-{index}` (connection `connection-{index}`)"
+                    for index in range(10)
+                )
+                + ", and 2 more. Set `table_prefix` (for example `<source>_`) so this "
+                "connection writes to different table names."
+            ),
+            id="stream-list-capped",
+        ),
+        pytest.param(
+            "https://reference.airbyte.com/reference/errors"
+            "#error:connection-conflicting-destination-stream",
+            {
+                "streams": [
+                    {
+                        "streamName": "users",
+                        "streamNamespace": "analytics",
+                        "connectionIds": ["users-connection"],
+                    },
+                    {"streamName": 42},
+                    {
+                        "streamName": "orders",
+                        "streamNamespace": None,
+                        "connectionIds": ["orders-connection"],
+                    },
+                ]
+            },
+            "Another connection on this destination already writes to these tables: "
+            "`analytics.users` (connection `users-connection`), `orders` "
+            "(connection `orders-connection`). Set `table_prefix` (for example "
+            "`<source>_`) so this connection writes to different table names.",
+            id="malformed-stream-item-skipped",
+        ),
+        pytest.param(
+            "https://reference.airbyte.com/reference/errors"
+            "#error:connection-conflicting-destination-stream",
+            {"streams": [{"streamName": ""}, {"streamName": "orders"}]},
+            "Another connection on this destination may already write to the same tables. "
+            "Set `table_prefix` (for example `<source>_`) so this connection writes to "
+            "different table names.",
+            id="all-stream-items-malformed",
+        ),
+        pytest.param(
+            "https://reference.airbyte.com/reference/errors#bad-request",
+            {"message": "Internal API message"},
+            None,
+            id="non-conflict-400",
+        ),
+    ],
+)
+def test_create_connection_400_guidance(
     monkeypatch: pytest.MonkeyPatch,
+    problem_type: str,
+    data: Any,
+    expected_guidance: str | None,
 ) -> None:
-    """A 400 raises `AirbyteLibInputError`; `data.message` is never relayed."""
-    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."
+    """Use conflicting-stream data for guidance without exposing other API details."""
+    detail = (
+        "The request could not be understood by the server due to malformed syntax."
     )
     body = json.dumps({
         "status": 400,
-        "type": "https://reference.airbyte.com/reference/errors#bad-request",
+        "type": problem_type,
         "title": "bad-request",
-        "detail": "The request could not be understood by the server "
-        "due to malformed syntax.",
+        "detail": detail,
         "documentationUrl": None,
-        "data": {"message": api_message},
+        "data": data,
     })
     raw_response = requests.Response()
     raw_response.status_code = 400
@@ -2326,15 +2417,15 @@ def test_create_connection_400_raises_input_error_without_api_message(
     with pytest.raises(AirbyteLibInputError) as exc_info:
         _call_create_connection()
 
-    assert exc_info.value.get_message() == (
-        "The request could not be understood by the server due to malformed syntax."
-    )
-    assert exc_info.value.guidance is None
-    assert api_message not in str(exc_info.value)
+    assert exc_info.value.get_message() == detail
+    assert exc_info.value.guidance == expected_guidance
+    assert "Internal API message" not in str(exc_info.value)
     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
+    expected_text = f"{detail} {expected_guidance}" if expected_guidance else detail
+    assert format_user_facing_error(exc_info.value) == expected_text
 
 
 def test_create_connection_400_with_non_json_body_uses_fallback_message(