From 92571071dc07078eaf3f4dfffb4591f10da84ebb Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:27:23 +0000 Subject: [PATCH 1/5] style(test): move MCP error helper import to module scope Co-Authored-By: AJ Steers --- tests/unit_tests/test_mcp_tool_errors.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/unit_tests/test_mcp_tool_errors.py b/tests/unit_tests/test_mcp_tool_errors.py index 1164e851a..e030d1d12 100644 --- a/tests/unit_tests/test_mcp_tool_errors.py +++ b/tests/unit_tests/test_mcp_tool_errors.py @@ -28,6 +28,7 @@ MCP_TOOL_USER_FACING_ERRORS, AgentErrorTextMiddleware, agent_error_text, + cloud_error_trace_message, redact_agent_text, classify_mcp_tool_error, format_user_facing_error, @@ -579,8 +580,6 @@ def test_local_guidance_is_preserved_with_cloud_error( tool_error.__cause__ = error assert agent_error_text(tool_error) == expected - from airbyte.mcp._error_handling import cloud_error_trace_message - assert cloud_error_trace_message(error) == cloud_message From 023f1ea3f05dd108d5420099c0615e0afcfc64be Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:32:02 +0000 Subject: [PATCH 2/5] fix(mcp): preserve local guidance with Cloud carriers Co-Authored-By: AJ Steers --- airbyte/mcp/_error_handling.py | 10 ++++------ tests/unit_tests/test_mcp_tool_errors.py | 4 +++- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/airbyte/mcp/_error_handling.py b/airbyte/mcp/_error_handling.py index a870d6f3b..f6cfaa0fd 100644 --- a/airbyte/mcp/_error_handling.py +++ b/airbyte/mcp/_error_handling.py @@ -28,6 +28,7 @@ from airbyte.exceptions import ( AirbyteAgentsUnavailableError, AirbyteCloudApiError, + AirbyteCloudError, AirbyteConnectionSyncActiveError, AirbyteConnectionSyncError, AirbyteConnectionSyncTimeoutError, @@ -208,6 +209,9 @@ def _format_cloud_error_with_local_guidance(error: BaseException) -> str | None: for cause in _chain(error) if isinstance(cause, AirbyteLibError) and not isinstance(cause, AirbyteCloudApiError) + and not ( + isinstance(cause, AirbyteCloudError) and cause.get_message() == problem.message + ) ), None, ) @@ -215,12 +219,6 @@ def _format_cloud_error_with_local_guidance(error: BaseException) -> str | None: return _cloud_problem_text(problem) local_message = local_error.get_message() - if ( - isinstance(local_error, AirbyteConnectorNotReadyError) - and local_message == problem.message - ): - return _cloud_problem_text(problem) - local_text = f"{local_message} {local_error.guidance}" if problem.message == local_message or problem.message in local_text: return local_text diff --git a/tests/unit_tests/test_mcp_tool_errors.py b/tests/unit_tests/test_mcp_tool_errors.py index e030d1d12..4e7aed5a6 100644 --- a/tests/unit_tests/test_mcp_tool_errors.py +++ b/tests/unit_tests/test_mcp_tool_errors.py @@ -586,7 +586,9 @@ def test_local_guidance_is_preserved_with_cloud_error( def test_active_sync_error_is_a_user_facing_mcp_error() -> None: connection_id = "connection-id" job_id = 42 - message = f"A sync is already running for connection '{connection_id}' (job {job_id})." + message = ( + f"A sync is already running for connection '{connection_id}' (job {job_id})." + ) guidance = "Monitor it with `get_cloud_sync_status`." body = json.dumps({ "type": "https://reference.airbyte.com/reference/errors#try-again-later", From 6b69534ee4d7fb8f6a81235f9e3546c1a49d9d1b Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:32:20 +0000 Subject: [PATCH 3/5] feat(cloud): guide conflicting destination stream errors Co-Authored-By: AJ Steers --- airbyte/_util/api_util.py | 78 ++++++++++++++++++++ airbyte/cloud/workspaces.py | 3 +- airbyte/mcp/cloud.py | 9 ++- tests/unit_tests/test_cloud_api_util.py | 98 +++++++++++++++++++++---- 4 files changed, 171 insertions(+), 17 deletions(-) diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py index bb6ebc4da..7a7565bc4 100644 --- a/airbyte/_util/api_util.py +++ b/airbyte/_util/api_util.py @@ -1957,6 +1957,8 @@ def patch_destination( # Create and delete connections +_MAX_CONFLICTING_DESTINATION_STREAMS = 10 + def build_stream_configurations( stream_names: list[str], @@ -1989,6 +1991,81 @@ def build_connection_schedule( ) +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.strip(): + return None + 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) or not connection_id.strip() + 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: + try: + if len(body.encode("utf-8")) > 64 * 1024: + return None + payload = json.loads(body) + except (UnicodeEncodeError, json.JSONDecodeError, RecursionError): + return None + + data = payload.get("data") if isinstance(payload, dict) else None + 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 None: + return None + formatted_streams.append(formatted_stream) + + 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 `_`) 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 `_`) so this " + "connection writes to different table names." + ) + + def create_connection( # noqa: PLR0913 # Too many arguments name: str, *, @@ -2029,6 +2106,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 diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 101e7d1eb..18eb8172e 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -997,7 +997,8 @@ 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 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: diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 6c6aa69e1..b20b0f2e1 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -908,7 +908,11 @@ def create_connection_on_cloud( table_prefix: Annotated[ str | None, Field( - description="Optional table prefix to use when syncing to the destination.", + description=( + "Two connections to the same destination cannot write to the same table " + "(namespace + name); use `table_prefix`, such as `_`, to keep " + "them separate." + ), default=None, ), TraceArg.FINGERPRINT, @@ -922,6 +926,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 `_`) to keep them separate. """ resolved_streams_list: list[str] = resolve_list_of_strings(selected_streams) or [] workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index f4907260b..18adc7d25 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/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,90 @@ 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", + "streamNamespace": "analytics", + "connectionIds": ["users-connection"], + }, + { + "streamName": "orders", + "streamNamespace": "", + "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 " + "`_`) 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 `_`) 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": "malformed"}, + "Another connection on this destination may already write to the same tables. " + "Set `table_prefix` (for example `_`) so this connection writes to " + "different table names.", + id="malformed-stream-data", + ), + 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 +2394,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( From 680eb1172617fe36c9cb3255bcd55d9339d2554b Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:37:30 +0000 Subject: [PATCH 4/5] fix(cloud): improve conflicting stream guidance Co-Authored-By: AJ Steers --- airbyte/_util/api_util.py | 39 +++++++++++++------------ airbyte/_util/cloud_errors.py | 37 ++++++++++++++--------- airbyte/cloud/workspaces.py | 5 ++-- airbyte/mcp/cloud.py | 8 ++--- tests/unit_tests/test_cloud_api_util.py | 31 +++++++++++++++++--- 5 files changed, 76 insertions(+), 44 deletions(-) diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py index 7a7565bc4..84cc80b31 100644 --- a/airbyte/_util/api_util.py +++ b/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, @@ -1957,8 +1961,6 @@ def patch_destination( # Create and delete connections -_MAX_CONFLICTING_DESTINATION_STREAMS = 10 - def build_stream_configurations( stream_names: list[str], @@ -1991,22 +1993,24 @@ 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") + namespace = stream.get("streamNamespace", "") connection_ids = stream.get("connectionIds") - if not isinstance(name, str) or not name.strip(): + 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) or not connection_id.strip() - for connection_id in connection_ids - ): + if any(not isinstance(connection_id, str) for connection_id in connection_ids): return None stream_name = f"{namespace}.{name}" if namespace else name @@ -2015,14 +2019,10 @@ def _format_conflicting_destination_stream(stream: object) -> str | None: def _format_conflicting_destination_streams(body: str) -> str | None: - try: - if len(body.encode("utf-8")) > 64 * 1024: - return None - payload = json.loads(body) - except (UnicodeEncodeError, json.JSONDecodeError, RecursionError): + payload = parse_problem_payload(body) + if payload is None: return None - - data = payload.get("data") if isinstance(payload, dict) else 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 @@ -2030,9 +2030,10 @@ def _format_conflicting_destination_streams(body: str) -> str | None: formatted_streams: list[str] = [] for stream in streams: formatted_stream = _format_conflicting_destination_stream(stream) - if formatted_stream is None: - return None - formatted_streams.append(formatted_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: diff --git a/airbyte/_util/cloud_errors.py b/airbyte/_util/cloud_errors.py index fec2e161d..495ee1f60 100644 --- a/airbyte/_util/cloud_errors.py +++ b/airbyte/_util/cloud_errors.py @@ -83,22 +83,31 @@ 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): + _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, diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 18eb8172e..f0b77ca90 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -997,8 +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 prefix to keep connections to the same destination from - writing to the same table (namespace + name). + 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: diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index b20b0f2e1..b1498d70e 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -908,11 +908,9 @@ def create_connection_on_cloud( table_prefix: Annotated[ str | None, Field( - description=( - "Two connections to the same destination cannot write to the same table " - "(namespace + name); use `table_prefix`, such as `_`, to keep " - "them separate." - ), + 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 `_` to keep them separate.", default=None, ), TraceArg.FINGERPRINT, diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index 18adc7d25..3fde2e91a 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/tests/unit_tests/test_cloud_api_util.py @@ -2302,7 +2302,6 @@ def _call_create_connection() -> None: "streams": [ { "streamName": "users", - "streamNamespace": "analytics", "connectionIds": ["users-connection"], }, { @@ -2313,7 +2312,7 @@ def _call_create_connection() -> None: ] }, "Another connection on this destination already writes to these tables: " - "`analytics.users` (connection `users-connection`), `orders` " + "`users` (connection `users-connection`), `orders` " "(connection `orders-connection`). Set `table_prefix` (for example " "`_`) so this connection writes to different table names.", id="conflicting-streams", @@ -2345,11 +2344,35 @@ def _call_create_connection() -> None: pytest.param( "https://reference.airbyte.com/reference/errors" "#error:connection-conflicting-destination-stream", - {"streams": "malformed"}, + { + "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 " + "`_`) 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 `_`) so this connection writes to " "different table names.", - id="malformed-stream-data", + id="all-stream-items-malformed", ), pytest.param( "https://reference.airbyte.com/reference/errors#bad-request", From aaf4b2b2c22fca1388c780ae7b81279f004fa1dd Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:38:46 +0000 Subject: [PATCH 5/5] docs: explain problem-message pre-check in parse_problem_payload Co-Authored-By: AJ Steers --- airbyte/_util/cloud_errors.py | 1 + 1 file changed, 1 insertion(+) diff --git a/airbyte/_util/cloud_errors.py b/airbyte/_util/cloud_errors.py index 495ee1f60..3fb82ed70 100644 --- a/airbyte/_util/cloud_errors.py +++ b/airbyte/_util/cloud_errors.py @@ -97,6 +97,7 @@ def parse_problem_payload(body: str | None) -> dict[str, Any] | 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