From b1fb644e981abff4e7703a5e6be6531d97bdaedb Mon Sep 17 00:00:00 2001 From: Caleb Evans Date: Mon, 12 Oct 2026 19:31:16 -0700 Subject: [PATCH 1/8] refactor(mcp): relay Cloud problem detail instead of fixed messages --- AGENTS.md | 10 +- CONTRIBUTING.md | 13 +- airbyte/_direct_connectors/api_util.py | 40 +- airbyte/_util/api_util.py | 98 +-- airbyte/_util/cloud_errors.py | 173 +--- airbyte/_util/cloud_errors.yaml | 288 ------ airbyte/mcp/_datadog.py | 4 +- airbyte/mcp/_error_handling.py | 75 +- tests/unit_tests/test_cloud_api_util.py | 821 +++++------------- tests/unit_tests/test_cloud_direct_execute.py | 61 +- tests/unit_tests/test_mcp_datadog.py | 65 +- tests/unit_tests/test_mcp_tool_errors.py | 175 +++- 12 files changed, 605 insertions(+), 1218 deletions(-) delete mode 100644 airbyte/_util/cloud_errors.yaml diff --git a/AGENTS.md b/AGENTS.md index e48736ca9..a27c3e255 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -22,8 +22,14 @@ Key conventions: ## Agent-facing Cloud Error Messages -Cloud API error text shown to agents comes from `airbyte/_util/cloud_errors.yaml`. Read its -header and the matching section in `CONTRIBUTING.md` before you change it. +Cloud API error text comes from the response's nonempty `detail`, falling back to `title`. +PyAirbyte preserves the selected string; fix poor Cloud wording in the platform. +Only actor-not-ready setup instructions and 403 resource/access guidance are added. +Bodies must be JSON objects of at most 65,536 UTF-8 bytes; invalid or non-problem responses +use a status-only fallback. `message`, `data.message`, and `errorId` are not displayed. +Agent text still passes through credential redaction. Native Datadog uses the selected, +unredacted Cloud message without guidance; this does not add it to LLM Obs metadata. + Non-Cloud MCP trace messages are the first line of the exception class's docstring (trace-only, exported to Datadog as `error.message`), so keep those first lines fixed text with no IDs, values or URLs. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 39619bd8b..b3e1c1a9f 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -27,12 +27,13 @@ as a "presentation" layer on top of the core modules.) ## Agent-facing Cloud Error Messages -When an Airbyte Cloud API call fails, the text the agent sees comes from the fixed table in -`airbyte/_util/cloud_errors.yaml`, keyed by Cloud's problem type. Never build that text from -Cloud's `detail`, `data.message` or other response fields, because they can quote secrets or -server internals. To support a new problem type, add a `message` and a `guidance` entry -to the YAML file and add the type to `test_every_published_problem_type_matches_a_fixed_message`. -The header of the YAML file lists the remaining rules. +Cloud API error text comes from the response's nonempty `detail`, falling back to `title`. +PyAirbyte preserves the selected string; fix poor Cloud wording in the platform. +Only actor-not-ready setup instructions and 403 resource/access guidance are added. +Bodies must be JSON objects of at most 65,536 UTF-8 bytes; invalid or non-problem responses +use a status-only fallback. `message`, `data.message`, and `errorId` are not displayed. +Agent text still passes through credential redaction. Native Datadog uses the selected, +unredacted Cloud message without guidance; this does not add it to LLM Obs metadata. Non-Cloud MCP trace messages are the first line of the exception class's docstring (trace-only, exported to Datadog as `error.message`), so keep those first lines fixed text diff --git a/airbyte/_direct_connectors/api_util.py b/airbyte/_direct_connectors/api_util.py index f999fca03..8cf7514bf 100644 --- a/airbyte/_direct_connectors/api_util.py +++ b/airbyte/_direct_connectors/api_util.py @@ -26,11 +26,14 @@ from airbyte._util.api_util import ( AIRBYTE_CLOUD_ANALYTIC_SOURCE_HEADER, CLOUD_API_ROOT, + cloud_error_guidance, get_bearer_token, get_cloud_api_analytic_source, get_config_api_root, status_ok, ) +from airbyte._util.cloud_errors import parse_cloud_error +from airbyte._util.meta import is_mcp_mode from airbyte.exceptions import ( AirbyteAgentsUnavailableError, AirbyteCloudApiError, @@ -71,38 +74,6 @@ def _resolve_bearer_token(credentials: _AirbyteCredentials) -> str: ) -def _error_message(*, response: requests.Response, full_url: str) -> str: - """Build an error message for a failed Cloud API request.""" - message = f"Airbyte Cloud API request failed with status {response.status_code}" - if response.status_code == HTTPStatus.UNAUTHORIZED: - return f"{message} (Unauthorized) when accessing: {full_url}." - if response.status_code == HTTPStatus.FORBIDDEN: - return f"{message} (Forbidden) when accessing: {full_url}." - return f"{message} when accessing: {full_url}." - - -def _error_guidance(*, response: requests.Response) -> str | None: - """Return actionable guidance for a failed Cloud API request, if any applies.""" - if response.status_code == HTTPStatus.UNAUTHORIZED: - return "Check that the Airbyte Cloud credentials are valid." - if response.status_code == HTTPStatus.FORBIDDEN: - return ( - "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 - - def is_not_enabled_error(error: AirbyteCloudError) -> bool: """Return whether `error` reports the connector is not enabled for a Fusion feature. @@ -169,10 +140,11 @@ def make_cloud_agent_request( timeout=_REQUEST_TIMEOUT_SECONDS, ) if not status_ok(response.status_code): + problem = parse_cloud_error(response.status_code, response.text) raise AirbyteCloudApiError( status_code=response.status_code, - message=_error_message(response=response, full_url=full_url), - guidance=_error_guidance(response=response), + message=problem.message, + guidance=cloud_error_guidance(problem, for_mcp=is_mcp_mode()), context={ "full_url": full_url, "path": path, diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py index 8002227fe..e2a87f397 100644 --- a/airbyte/_util/api_util.py +++ b/airbyte/_util/api_util.py @@ -26,7 +26,7 @@ from airbyte_api import api, models from airbyte_api.errors import SDKError -from airbyte._util.cloud_errors import describe_cloud_error, parse_cloud_error +from airbyte._util.cloud_errors import CloudErrorInfo, parse_cloud_error from airbyte._util.meta import ( AIRBYTE_CLOUD_ANALYTIC_SOURCE_HEADER, get_cloud_api_analytic_source, @@ -64,15 +64,8 @@ PAGE_SIZE = 100 JWT_PART_COUNT = 3 FORBIDDEN_RESOURCE_GUIDANCE = ( - "Airbyte Cloud returns 403 both for IDs that don't exist and for resources outside your " - "access. Check the ID, and that it belongs to the workspace you passed." -) -ACTOR_NOT_READY_ERROR_MESSAGE = ( - "The source or destination is still a draft: a person has not finished its setup in " - "Airbyte Cloud (HTTP 409 actor-not-ready)." -) -FORBIDDEN_RESOURCE_MESSAGE = ( - "The requested resource was not found, or these credentials can't access it (HTTP 403)." + "The resource may be missing or inaccessible in this workspace. " + "Check the resource ID, workspace, and access permissions." ) ACTOR_NOT_READY_MCP_GUIDANCE = ( "This is expected right after creating a connector with `defer_credentials=True`. " @@ -184,30 +177,21 @@ def sdk_error_response(error: BaseException) -> tuple[int | None, str | None] | def _is_actor_not_ready_error(error: SDKError) -> bool: """Check whether an SDKError contains Airbyte Cloud's actor-not-ready problem.""" - if error.status_code != HTTPStatus.CONFLICT: - return False + return parse_cloud_error(error.status_code, error.body).is_actor_not_ready - try: - body = json.loads(error.body) - except (json.JSONDecodeError, TypeError): - return False - if not isinstance(body, dict): - return False - error_type = body.get("type") - return body.get("title") == "actor-not-ready" or ( - isinstance(error_type, str) and error_type.endswith("409-actor-not-ready") - ) +def sdk_error_message(error: SDKError) -> str: + """Return the selected Cloud problem message or transport status fallback.""" + return parse_cloud_error(error.status_code, error.body).message -def sdk_error_message(error: SDKError) -> str: - """Return the fixed message an SDKError is reported with; never Cloud body text.""" - if _is_actor_not_ready_error(error): - return ACTOR_NOT_READY_ERROR_MESSAGE - if error.status_code == HTTPStatus.FORBIDDEN: - return FORBIDDEN_RESOURCE_MESSAGE - message, _ = describe_cloud_error(parse_cloud_error(error.status_code, error.body)) - return message +def cloud_error_guidance(problem: CloudErrorInfo, *, for_mcp: bool) -> str | None: + """Add only connector setup instructions or general resource/access guidance.""" + if problem.is_actor_not_ready: + return ACTOR_NOT_READY_MCP_GUIDANCE if for_mcp else ACTOR_NOT_READY_PYTHON_GUIDANCE + if problem.status_code == HTTPStatus.FORBIDDEN: + return FORBIDDEN_RESOURCE_GUIDANCE + return None def _wrap_sdk_error( @@ -225,26 +209,15 @@ def _wrap_sdk_error( **sdk_context, "problem_type": problem.slug, } - if _is_actor_not_ready_error(error): - return AirbyteConnectorNotReadyError( - message=ACTOR_NOT_READY_ERROR_MESSAGE, - guidance=( - ACTOR_NOT_READY_MCP_GUIDANCE if is_mcp_mode() else ACTOR_NOT_READY_PYTHON_GUIDANCE - ), - context=merged_context, - ) - - status_code = sdk_context.get("status_code") - is_forbidden = status_code == HTTPStatus.FORBIDDEN - _, guidance = describe_cloud_error(problem) - error_type = ( - AirbyteMissingResourceError - if is_forbidden or status_code == HTTPStatus.NOT_FOUND - else AirbyteCloudError - ) + if problem.is_actor_not_ready: + error_type = AirbyteConnectorNotReadyError + elif error.status_code in {HTTPStatus.FORBIDDEN, HTTPStatus.NOT_FOUND}: + error_type = AirbyteMissingResourceError + else: + error_type = AirbyteCloudError return error_type( - message=sdk_error_message(error), - guidance=FORBIDDEN_RESOURCE_GUIDANCE if is_forbidden else guidance, + message=problem.message, + guidance=cloud_error_guidance(problem, for_mcp=is_mcp_mode()), context=merged_context, ) @@ -2017,13 +1990,8 @@ def create_connection( # noqa: PLR0913 # Too many arguments "selected_stream_names": selected_stream_names, } if e.status_code == HTTPStatus.BAD_REQUEST: - wrapped_message, _ = describe_cloud_error(parse_cloud_error(e.status_code, e.body)) raise AirbyteLibInputError( - message=wrapped_message, - guidance=( - "Check that every name in `selected_streams` is a stream of the " - "source, and that `source_id` and `destination_id` are correct." - ), + message=parse_cloud_error(e.status_code, e.body).message, context={**base_context, **_get_sdk_error_context(e)}, ) from e raise _wrap_sdk_error(e, base_context) from e @@ -2367,7 +2335,8 @@ def _make_config_api_request( response.raise_for_status() except requests.HTTPError as ex: problem = parse_cloud_error(response.status_code, response.text) - message, guidance = describe_cloud_error(problem) + message = problem.message + guidance = cloud_error_guidance(problem, for_mcp=is_mcp_mode()) error_context = { "full_url": full_url, "config_api_root": config_api_root, @@ -2378,11 +2347,8 @@ def _make_config_api_request( } if response.status_code == HTTPStatus.FORBIDDEN: raise AirbyteMissingResourceError( - message=( - "The requested resource was not found, or these credentials can't " - "access it (HTTP 403)." - ), - guidance=FORBIDDEN_RESOURCE_GUIDANCE, + message=message, + guidance=guidance, context=error_context, ) from ex raise AirbyteCloudError( @@ -2454,7 +2420,8 @@ def _get_public_api_json( response.raise_for_status() except requests.HTTPError as ex: problem = parse_cloud_error(response.status_code, response.text) - message, guidance = describe_cloud_error(problem) + message = problem.message + guidance = cloud_error_guidance(problem, for_mcp=is_mcp_mode()) error_context = { "full_url": full_url, "path": path, @@ -2470,11 +2437,8 @@ def _get_public_api_json( ) from ex if response.status_code == HTTPStatus.FORBIDDEN: raise AirbyteMissingResourceError( - message=( - "The requested resource was not found, or these credentials can't " - "access it (HTTP 403)." - ), - guidance=FORBIDDEN_RESOURCE_GUIDANCE, + message=message, + guidance=guidance, context=error_context, ) from ex raise AirbyteCloudError( diff --git a/airbyte/_util/cloud_errors.py b/airbyte/_util/cloud_errors.py index 3a13088a2..174ee7c28 100644 --- a/airbyte/_util/cloud_errors.py +++ b/airbyte/_util/cloud_errors.py @@ -1,25 +1,14 @@ # Copyright (c) 2026 Airbyte, Inc., all rights reserved. -"""Fixed agent-facing messages for Airbyte Cloud problem (RFC 9457) error bodies. - -`parse_cloud_error` reads only the allow-listed fields of a Cloud error body, -`describe_cloud_error` maps it to the fixed `(message, guidance)` text stored in -`cloud_errors.yaml`, and `cloud_errors.yaml` is the single source of those -strings. `detail`, `data.message` and every other body field can quote caller -values or server internals, so they are never read for message text. -""" +"""Bounded parsing of Cloud problem messages and validated reason labels.""" from __future__ import annotations -import functools -import importlib.resources import json import re from dataclasses import dataclass from http import HTTPStatus from typing import Any -import yaml - _PROBLEM_SLUG = re.compile(r"[a-z0-9][a-z0-9:._-]{0,99}") # The last path segment of the generic problem `type` URL; `title` is the slug then. @@ -27,22 +16,6 @@ # A slug names a kind of problem; one that is or holds an ID names a resource. _ID_LIKE = re.compile(r"\d+|.*[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}.*") _MAX_PROBLEM_BODY = 65536 -_RESOURCE_TYPES = frozenset( - { - "source", - "destination", - "connection", - "workspace", - "organization", - "job", - "user", - "permission", - "source_definition", - "destination_definition", - } -) -_UUID = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}") -_PROBLEM_TYPE_PREFIX = "https://reference.airbyte.com/reference/errors" def is_valid_problem_slug(value: object) -> bool: @@ -54,18 +27,12 @@ def is_valid_problem_slug(value: object) -> bool: @dataclass(frozen=True) class CloudErrorInfo: - """The allow-listed fields of an Airbyte Cloud error body. - - `detail`, `data.message` and every other field can quote caller values or - server internals, so they are never read for message text. - """ + """Cloud-owned message, transport status, and independently validated reason.""" status_code: int | None - slug: str | None # Last `type`/`title` segment, like `mcp_tool_error_reason`. - key: str | None # Normalized `type` path used to look up the fixed message. - resource_type: str | None # `data.resourceType`, a fixed word. - limit: int | None # `data.limit`. - error_id: str | None # Top-level `errorId` of a non-problem body. + slug: str | None + message: str + is_actor_not_ready: bool def _problem_slug(problem: dict[str, Any]) -> str | None: @@ -94,111 +61,41 @@ def slug_of(value: object) -> str | None: return slug -def _problem_key(problem: dict[str, Any]) -> str | None: - """Return the normalized `type` path used to look up the fixed message. +def _problem_message(payload: dict[str, Any]) -> str | None: + for field in ("detail", "title"): + value = payload.get(field) + if isinstance(value, str) and value.strip(): + return value + return None - Cloud emits `type` as the bare errors URL (slug in `title`), as the URL - plus `#`, or as `error:`; all three normalize to the same key. - """ - value = problem.get("type") - key = value if isinstance(value, str) else "" - if key.startswith(_PROBLEM_TYPE_PREFIX): - key = key.removeprefix(_PROBLEM_TYPE_PREFIX) - if key[:1] in {"#", "/"}: - key = key[1:] - if key.startswith("error:"): - key = key.removeprefix("error:") - if key.startswith("409-"): - key = key.removeprefix("409-") - if key in {"", _GENERIC_PROBLEM_SLUG}: - title = problem.get("title") - key = title if isinstance(title, str) else "" - if _ID_LIKE.fullmatch(key): - return None - return key or None +def _status_fallback_message(status_code: int | None) -> str: + if status_code == HTTPStatus.TOO_MANY_REQUESTS: + return "Airbyte Cloud returned HTTP 429 (rate limited)." + if status_code is not None: + return f"Airbyte Cloud returned HTTP {status_code}." + return "Airbyte Cloud request failed." -def parse_cloud_error(status_code: int | None, body: str | None) -> CloudErrorInfo: - """Read only the allow-listed fields of a Cloud API error body. - Bodies over 64 KiB or non-JSON give `slug=None`. - """ - slug = key = resource_type = error_id = None - limit = None - if body is not None and len(body) <= _MAX_PROBLEM_BODY: +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 + if isinstance(body, str) and len(body) <= _MAX_PROBLEM_BODY: try: - problem = json.loads(body) - except (ValueError, RecursionError): - problem = None - if isinstance(problem, dict): - slug = _problem_slug(problem) - key = _problem_key(problem) - data = problem.get("data") - if isinstance(data, dict): - candidate = data.get("resourceType") - if isinstance(candidate, str) and candidate in _RESOURCE_TYPES: - resource_type = candidate - limit_value = data.get("limit") - if isinstance(limit_value, int) and not isinstance(limit_value, bool): - limit = limit_value - candidate = problem.get("errorId") - if isinstance(candidate, str) and _UUID.fullmatch(candidate): - error_id = candidate + 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) + except (UnicodeEncodeError, json.JSONDecodeError, RecursionError): + pass + problem = payload if isinstance(payload, dict) else {} + problem_type = problem.get("type") return CloudErrorInfo( status_code=status_code, - slug=slug, - key=key, - resource_type=resource_type, - limit=limit, - error_id=error_id, + slug=_problem_slug(problem), + message=_problem_message(problem) or _status_fallback_message(status_code), + is_actor_not_ready=status_code == HTTPStatus.CONFLICT + and ( + problem.get("title") == "actor-not-ready" + or (isinstance(problem_type, str) and problem_type.endswith("409-actor-not-ready")) + ), ) - - -@functools.cache -def _messages() -> dict[str, Any]: - """Return the fixed error text table from `cloud_errors.yaml`.""" - path = importlib.resources.files("airbyte._util").joinpath("cloud_errors.yaml") - return yaml.safe_load(path.read_text(encoding="utf-8")) - - -def _status_fallback_message(status_code: int | None) -> tuple[str, str]: - """Return (message, guidance) for a status without a known problem key.""" - fallbacks = _messages()["status_fallbacks"] - if status_code in fallbacks: - entry = fallbacks[status_code] - elif status_code is None: - entry = fallbacks["default"] - elif status_code < HTTPStatus.INTERNAL_SERVER_ERROR: - entry = fallbacks["4xx"] - else: - entry = fallbacks["5xx"] - return entry["message"], entry["guidance"] - - -def describe_cloud_error(problem: CloudErrorInfo) -> tuple[str, str]: - """Return (message, guidance) for a parsed Cloud problem.""" - entry = _messages().get(problem.key or "") - if entry is None: - message, guidance = _status_fallback_message(problem.status_code) - else: - message, guidance = entry["message"], entry["guidance"] - # `key` is raw `type`/`title` text and can quote caller input, so it is - # shown only when it matched a fixed message row. - known_key = entry is not None - if problem.key == "resource-not-found" and problem.resource_type: - message = f"The {problem.resource_type} was not found." - elif problem.key == "workspace-limit-for-organization-reached" and problem.limit is not None: - message += f" (limit: {problem.limit} workspaces)" - if problem.status_code is not None: - message += ( - f" (Cloud error: {problem.key}, HTTP {problem.status_code})" - if known_key - else f" (HTTP {problem.status_code})" - ) - if ( - problem.status_code is not None - and problem.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR - and problem.error_id - ): - guidance += f" Error ID: {problem.error_id}." - return message, guidance diff --git a/airbyte/_util/cloud_errors.yaml b/airbyte/_util/cloud_errors.yaml deleted file mode 100644 index 3bafd3deb..000000000 --- a/airbyte/_util/cloud_errors.yaml +++ /dev/null @@ -1,288 +0,0 @@ -# Agent-facing messages for Airbyte Cloud problem types, used by -# `describe_cloud_error` in cloud_errors.py. -# -# Maintaining this file: -# - Keys are normalized problem types: the bare path, like `connection/locked`, -# without the `https://reference.airbyte.com/reference/errors` URL, `#`, -# `error:` or `409-` prefix (see `_problem_key`). Keep keys sorted. -# - Each entry has exactly `message` (what went wrong) and `guidance` (one -# recovery step: retry later, fix input, ask the user, or don't retry). -# - Write fixed text only. Never copy in Cloud's `detail`, `data.message` or -# other body fields; they can quote caller secrets or server internals. -# - Problem types missing here fall back to `status_fallbacks` by HTTP status, -# and the raw type is not shown to the agent. -# - When adding a type, also add it to -# `test_every_published_problem_type_matches_a_fixed_message`. -# `test_cloud_errors_yaml_entries_have_message_and_guidance` checks the shape. -# - 403 and actor-not-ready are handled in api_util.py, not here. -auth/sso-required: - message: This email domain must sign in with SSO. - guidance: Ask the user to sign in through SSO. -auth/user-already-exists: - message: A resource with this name or membership already exists. - guidance: Use the existing one or pick another name; don't retry as is. -bad-request: - message: Airbyte Cloud rejected the request as invalid. - guidance: Check the arguments against the tool description. -basic-schedule-validation/under-one-hour-not-allowed: - message: The connection schedule is invalid. - guidance: Fix the schedule (cron expression, timezone, or frequency) and retry. -billing/insufficient-credit-balance: - message: The organization's payment status or credit balance doesn't allow this. - guidance: Ask the user to resolve billing; don't retry. -billing/insufficient-payment-status: - message: The organization's payment status or credit balance doesn't allow this. - guidance: Ask the user to resolve billing; don't retry. -billing/no-active-subscription: - message: This organization needs an active Airbyte subscription for this operation. - guidance: Ask the user to check billing; don't retry. -billing/no-cancelable-subscription: - message: There is no subscription change of this kind to act on. - guidance: Don't retry; tell the user. -billing/no-scheduled-cancellation-subscription: - message: There is no subscription change of this kind to act on. - guidance: Don't retry; tell the user. -billing/no-scheduled-plan-change: - message: There is no subscription change of this kind to act on. - guidance: Don't retry; tell the user. -billing/subscription/subscription-required: - message: This organization needs an active Airbyte subscription for this operation. - guidance: Ask the user to check billing; don't retry. -connection-conflicting-destination-stream: - message: The selected streams can't be used with this connection (file transfer is unsupported, or a stream collides with another connection). - guidance: Change the selected streams or connection settings and retry. -connection-validation/file-transfer/connection-unsupported: - message: The selected streams can't be used with this connection (file transfer is unsupported, or a stream collides with another connection). - guidance: Change the selected streams or connection settings and retry. -connection-validation/file-transfer/stream-unsupported: - message: The selected streams can't be used with this connection (file transfer is unsupported, or a stream collides with another connection). - guidance: Change the selected streams or connection settings and retry. -connection/destination-catalog/invalid-additional-field: - message: The destination catalog settings for a stream are invalid. - guidance: Fix the stream's destination settings (object name, sync mode, fields, primary key) and retry. -connection/destination-catalog/invalid-operation: - message: The destination catalog settings for a stream are invalid. - guidance: Fix the stream's destination settings (object name, sync mode, fields, primary key) and retry. -connection/destination-catalog/invalid-primary-key: - message: The destination catalog settings for a stream are invalid. - guidance: Fix the stream's destination settings (object name, sync mode, fields, primary key) and retry. -connection/destination-catalog/missing-object-name: - message: The destination catalog settings for a stream are invalid. - guidance: Fix the stream's destination settings (object name, sync mode, fields, primary key) and retry. -connection/destination-catalog/missing-primary-key: - message: The destination catalog settings for a stream are invalid. - guidance: Fix the stream's destination settings (object name, sync mode, fields, primary key) and retry. -connection/destination-catalog/missing-required-field: - message: The destination catalog settings for a stream are invalid. - guidance: Fix the stream's destination settings (object name, sync mode, fields, primary key) and retry. -connection/destination-catalog/required: - message: The destination catalog settings for a stream are invalid. - guidance: Fix the stream's destination settings (object name, sync mode, fields, primary key) and retry. -connection/locked: - message: The connection is locked and can't be changed right now. - guidance: Don't retry; ask the user. -connector-rollout/invalid-request: - message: The connector rollout request was rejected. - guidance: Don't retry; tell the user. -connector-rollout/not-enough-actors: - message: The connector rollout request was rejected. - guidance: Don't retry; tell the user. -connector-rollout/rollout-percentage-reached: - message: The connector rollout request was rejected. - guidance: Don't retry; tell the user. -cron-validation/invalid-expression: - message: The connection schedule is invalid. - guidance: Fix the schedule (cron expression, timezone, or frequency) and retry. -cron-validation/invalid-timezone: - message: The connection schedule is invalid. - guidance: Fix the schedule (cron expression, timezone, or frequency) and retry. -cron-validation/missing-component: - message: The connection schedule is invalid. - guidance: Fix the schedule (cron expression, timezone, or frequency) and retry. -cron-validation/missing-cron-data: - message: The connection schedule is invalid. - guidance: Fix the schedule (cron expression, timezone, or frequency) and retry. -cron-validation/under-one-hour-not-allowed: - message: The connection schedule is invalid. - guidance: Fix the schedule (cron expression, timezone, or frequency) and retry. -cron-validation/unsupported-timezone: - message: The connection schedule is invalid. - guidance: Fix the schedule (cron expression, timezone, or frequency) and retry. -dataplane-group-name-already-exists: - message: A resource with this name or membership already exists. - guidance: Use the existing one or pick another name; don't retry as is. -dataplane-name-already-exists: - message: A resource with this name or membership already exists. - guidance: Use the existing one or pick another name; don't retry as is. -dbtcloud/access-denied: - message: dbt Cloud rejected the integration credentials, requires a paid plan, or the call failed. - guidance: Ask the user to check the dbt Cloud integration. -dbtcloud/generic: - message: dbt Cloud rejected the integration credentials, requires a paid plan, or the call failed. - guidance: Ask the user to check the dbt Cloud integration. -dbtcloud/paid-plan-required: - message: dbt Cloud rejected the integration credentials, requires a paid plan, or the call failed. - guidance: Ask the user to check the dbt Cloud integration. -destination/catalog-not-found: - message: This connection has no destination catalog yet. - guidance: Set up the destination catalog before this operation; do not change the ID. -destination/discover-not-supported: - message: The destination catalog settings for a stream are invalid. - guidance: Fix the stream's destination settings (object name, sync mode, fields, primary key) and retry. -embedded/endpoint-moved: - message: This operation isn't available at this Airbyte endpoint. - guidance: Don't retry. -entitlement-service/error-adding-organization: - message: Airbyte Cloud is temporarily unavailable. - guidance: Wait and retry once. -entitlement-service/invalid-organization-state: - message: Airbyte Cloud failed to complete this operation. - guidance: Don't retry; tell the user. -failed-precondition: - message: A required condition for this operation isn't met. - guidance: Check the resource's state; ask the user if unclear. -generate-contribution: - message: Airbyte Cloud failed to complete this operation. - guidance: Don't retry; tell the user. -generate-contribution/connector-image-name-in-use: - message: A resource with this name or membership already exists. - guidance: Use the existing one or pick another name; don't retry as is. -generate-contribution/insufficient-github-token-permissions: - message: The GitHub token is invalid or lacks repo write permission. - guidance: Ask the user for a valid token. -generate-contribution/invalid-github-token: - message: The GitHub token is invalid or lacks repo write permission. - guidance: Ask the user for a valid token. -group-already-exists: - message: A resource with this name or membership already exists. - guidance: Use the existing one or pick another name; don't retry as is. -group-managed-by-scim: - message: This group is managed by SCIM and can't be changed here. - guidance: Don't retry; tell the user. -group-member-already-exists: - message: A resource with this name or membership already exists. - guidance: Use the existing one or pick another name; don't retry as is. -group-permission-already-exists: - message: A resource with this name or membership already exists. - guidance: Use the existing one or pick another name; don't retry as is. -implementation/not-implemented-in-oss: - message: This operation isn't available at this Airbyte endpoint. - guidance: Don't retry. -invalid-api-key: - message: Airbyte Cloud rejected the credentials. - guidance: Ask the user to check the client ID/secret or token; don't retry. -invalid-consent-url: - message: Airbyte Cloud failed to complete this operation. - guidance: Don't retry; tell the user. -invalid-redirect-url: - message: The redirect URL is not a valid HTTPS URL. - guidance: Fix the configuration and retry. -license/entitlement: - message: The organization's plan doesn't allow this. - guidance: Ask the user; don't retry. -mapper-validation: - message: The mapper configuration is invalid. - guidance: Fix the mapper configuration and retry. If a mapper secret is missing, resend the secret values. -mapper-validation/invalid-config: - message: The mapper configuration is invalid. - guidance: Fix the mapper configuration and retry. If a mapper secret is missing, resend the secret values. -mapper-validation/missing-required-param: - message: The mapper configuration is invalid. - guidance: Fix the mapper configuration and retry. If a mapper secret is missing, resend the secret values. -mapper-validation/runtime-secrets-manager-required: - message: This needs a runtime secrets manager configured. - guidance: Ask the user. -mapper-validation/secret-not-found: - message: The mapper configuration is invalid. - guidance: Fix the mapper configuration and retry. If a mapper secret is missing, resend the secret values. -notification/config/missing-url: - message: Airbyte Cloud rejected a value in the request. - guidance: Fix the input and retry. -notification/config/required: - message: Airbyte Cloud rejected a value in the request. - guidance: Fix the input and retry. -oauth-callback-failure: - message: Airbyte Cloud failed to complete this operation. - guidance: Don't retry; tell the user. -request-timeout-exceeded: - message: Airbyte Cloud timed out. The change may or may not have been applied. - guidance: Check the resource's current state before retrying; don't blindly retry a create. -resource-not-found: - message: The resource was not found. - guidance: Check the ID; list the resources to find the right one. -service-unavailable: - message: Airbyte Cloud is temporarily unavailable. - guidance: Wait and retry once. -sso-activation: - message: Airbyte Cloud failed to complete this operation. - guidance: Don't retry; tell the user. -sso-config-retrieval: - message: Airbyte Cloud failed to complete this operation. - guidance: Don't retry; tell the user. -sso-credential-update: - message: Airbyte Cloud failed to complete this operation. - guidance: Don't retry; tell the user. -sso-deletion: - message: Airbyte Cloud failed to complete this operation. - guidance: Don't retry; tell the user. -sso-setup: - message: Airbyte Cloud failed to complete this operation. - guidance: Don't retry; tell the user. -sso-token-validation: - message: The SSO access token is invalid or expired. - guidance: Ask the user to sign in again. -state-conflict: - message: The resource is in a state that doesn't allow this operation (for example, a job is running or already finished). - guidance: Check the resource's current status before retrying. -tag-already-exists: - message: A resource with this name or membership already exists. - guidance: Use the existing one or pick another name; don't retry as is. -tag-invalid-hex-color: - message: Airbyte Cloud rejected a value in the request. - guidance: Fix the input and retry. -tag-limit-for-workspace-reached: - message: The workspace has reached its tag limit. - guidance: Delete a tag first or ask the user. -tag-name-too-long: - message: Airbyte Cloud rejected a value in the request. - guidance: Fix the input and retry. -try-again-later: - message: Airbyte Cloud is busy with a conflicting change. - guidance: Wait and retry once. -unexpected-problem: - message: Airbyte Cloud hit an unexpected error. This is often caused by an invalid argument value. - guidance: Check the arguments (ranges, IDs, formats). If they look right, don't retry more than once; tell the user. -unprocessable-entity: - message: The request body or connector configuration failed validation. - guidance: Fix the configuration and retry. -value-not-found: - message: Airbyte doesn't recognize the connector type name. - guidance: Check the name against the connector list (e.g. `source-postgres`). -workspace-limit-for-organization-reached: - message: The organization's plan doesn't allow this. - guidance: Ask the user; don't retry. -status_fallbacks: - 401: - message: Airbyte Cloud rejected the credentials. - guidance: Ask the user to check the client ID/secret or token; don't retry. - 404: - message: The resource was not found. - guidance: Check the ID; list the resources to find the right one. - 408: - message: Airbyte Cloud timed out. The change may or may not have been applied. - guidance: Check the resource's current state before retrying; don't blindly retry a create. - 409: - message: The resource is in a state that doesn't allow this operation (for example, a job is running or already finished). - guidance: Check the resource's current status before retrying. - 429: - message: Airbyte Cloud is refusing requests from these credentials (rate limited). - guidance: Stop calling Airbyte Cloud tools and tell the user; don't retry. - 4xx: - message: Airbyte Cloud rejected the request as invalid. - guidance: Check the arguments against the tool description. - 5xx: - message: Airbyte Cloud hit an unexpected error. This is often caused by an invalid argument value. - guidance: Check the arguments (ranges, IDs, formats). If they look right, don't retry more than once; tell the user. - default: - message: Airbyte Cloud returned an error. - guidance: Retry at most once; tell the user if it persists. diff --git a/airbyte/mcp/_datadog.py b/airbyte/mcp/_datadog.py index 15332e43a..0a30b735d 100644 --- a/airbyte/mcp/_datadog.py +++ b/airbyte/mcp/_datadog.py @@ -212,7 +212,7 @@ def _exception_attributes(error: BaseException) -> dict[str, str]: if message := tool_error_trace_message(cause): attrs["error.message"] = message except Exception: - logger.debug("Datadog fixed error message unavailable") + logger.debug("Datadog error message unavailable") return attrs @@ -258,7 +258,7 @@ def _tool_error_message(span: Span) -> str: def _raised_tool_error_message(error: BaseException, span: Span) -> str: - """Return a fixed message when available, falling back to the span summary.""" + """Return a selected error message when available, falling back to the span summary.""" try: message = tool_error_trace_message(error) except Exception: diff --git a/airbyte/mcp/_error_handling.py b/airbyte/mcp/_error_handling.py index 22bcc28d9..5febdf2d3 100644 --- a/airbyte/mcp/_error_handling.py +++ b/airbyte/mcp/_error_handling.py @@ -16,12 +16,11 @@ from airbyte._util.api_util import ( SDKError, + cloud_error_guidance, error_response_body, - sdk_error_message, - sdk_error_response, ) from airbyte._util.cloud_errors import ( - describe_cloud_error, + CloudErrorInfo, is_valid_problem_slug, parse_cloud_error, ) @@ -97,36 +96,39 @@ """Expected errors returned to MCP clients as concise message and guidance text.""" -def cloud_error_trace_message(error: BaseException) -> str | None: - """Return the fixed Cloud error message for a failed call's span, if any. - - Only the message half: guidance can carry Cloud's error ID. - """ - cloud_error: AirbyteCloudApiError | None = None +def _cloud_error_info(error: BaseException) -> CloudErrorInfo | None: + """Resolve known Cloud carriers, preferring a concrete body to a bodyless wrapper.""" + fallback = None for cause in _chain(error): - if ( + if isinstance(cause, SDKError): + return parse_cloud_error(cause.status_code, cause.body) + if isinstance(cause, AirbyteCloudApiError): + body = error_response_body(cause) + problem = parse_cloud_error(cause.status_code, body) + if body is not None: + return problem + fallback = problem + elif ( isinstance(cause, AirbyteLibError) and isinstance(cause.__cause__, requests.HTTPError) and "problem_type" in (cause.context or {}) ): - return cause.message - if isinstance(cause, AirbyteCloudApiError): - body = error_response_body(cause) - if body is not None: - message, _ = describe_cloud_error(parse_cloud_error(cause.status_code, body)) - return message - cloud_error = cause - elif isinstance(cause, SDKError): - return sdk_error_message(cause) - - if cloud_error is not None: - message, _ = describe_cloud_error(parse_cloud_error(cloud_error.status_code, None)) - return message - return None + response = cause.__cause__.response + if response is not None: + return parse_cloud_error(response.status_code, response.text) + status = (cause.context or {}).get("status_code") + return parse_cloud_error(status if isinstance(status, int) else None, None) + return fallback + + +def cloud_error_trace_message(error: BaseException) -> str | None: + """Return the selected Cloud message without workflow guidance for a failed span.""" + problem = _cloud_error_info(error) + return problem.message if problem is not None else None def tool_error_trace_message(error: BaseException) -> str | None: - """Return the fixed trace message for an exception, if any.""" + """Return the Cloud message or a fixed non-Cloud trace message, if any.""" if (message := cloud_error_trace_message(error)) is not None: return message for cause in _chain(error): @@ -186,19 +188,15 @@ def cloud_error_route(error: BaseException) -> tuple[str, str] | None: return None -def _cloud_api_error_text(error: AirbyteCloudApiError) -> str: - """Return fixed Cloud error text, retaining direct-access guidance when present.""" - message, table_guidance = describe_cloud_error( - parse_cloud_error(error.status_code, error_response_body(error)) - ) - guidance = error.guidance if error.guidance is not None else table_guidance - return f"{message} {guidance}".rstrip() +def _cloud_problem_text(problem: CloudErrorInfo) -> str: + guidance = cloud_error_guidance(problem, for_mcp=True) + return f"{problem.message} {guidance}".rstrip() if guidance else problem.message def format_user_facing_error(error: BaseException) -> str: """Return the error message followed by its guidance, when present.""" - if isinstance(error, AirbyteCloudApiError): - return _cloud_api_error_text(error) + if (problem := _cloud_error_info(error)) is not None: + return _cloud_problem_text(problem) if not isinstance(error, AirbyteLibError): return str(error) text = error.get_message() @@ -243,16 +241,11 @@ def mcp_tool_error_reason(error: BaseException) -> str | None: def agent_error_text(error: ToolError) -> str: """Return the text an agent sees for a tool error before redaction.""" + if (problem := _cloud_error_info(error)) is not None: + return _cloud_problem_text(problem) cause = error.__cause__ - if isinstance(cause, AirbyteCloudApiError): - return _cloud_api_error_text(cause) if isinstance(cause, AirbyteLibError): return format_user_facing_error(cause) - sdk_response = sdk_error_response(cause) if cause is not None else None - if sdk_response is not None: - status_code, body = sdk_response - message, guidance = describe_cloud_error(parse_cloud_error(status_code, body)) - return f"{message} {guidance}" return str(error) diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index 947c29836..0a4f96786 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/tests/unit_tests/test_cloud_api_util.py @@ -270,25 +270,22 @@ def _list_workspaces_response( pytest.param( 403, AirbyteMissingResourceError, - "The requested resource was not found, or these credentials can't access it " - "(HTTP 403).", + "Airbyte Cloud returned HTTP 403.", api_util.FORBIDDEN_RESOURCE_GUIDANCE, id="forbidden", ), pytest.param( 404, AirbyteMissingResourceError, - "The resource was not found. (HTTP 404)", - "Check the ID; list the resources to find the right one.", + "Airbyte Cloud returned HTTP 404.", + None, id="not_found", ), pytest.param( 500, AirbyteCloudError, - "Airbyte Cloud hit an unexpected error. This is often caused by an " - "invalid argument value. (HTTP 500)", - "Check the arguments (ranges, IDs, formats). If they look right, " - "don't retry more than once; tell the user.", + "Airbyte Cloud returned HTTP 500.", + None, id="server_error", ), ], @@ -1270,9 +1267,7 @@ def test_get_connection_fallback_404_raises_missing_resource_error() -> None: assert len(responses.calls) == 2 assert exc_info.value.context["status_code"] == 404 - assert exc_info.value.guidance == ( - "Check the ID; list the resources to find the right one." - ) + assert exc_info.value.guidance is None @responses.activate @@ -2095,18 +2090,15 @@ def test_update_connector_builder_project_payload( pytest.param( 403, AirbyteMissingResourceError, - "The requested resource was not found, or these credentials can't access it " - "(HTTP 403).", + "Airbyte Cloud returned HTTP 403.", api_util.FORBIDDEN_RESOURCE_GUIDANCE, id="forbidden", ), pytest.param( 500, AirbyteCloudError, - "Airbyte Cloud hit an unexpected error. This is often caused by an " - "invalid argument value. (HTTP 500)", - "Check the arguments (ranges, IDs, formats). If they look right, " - "don't retry more than once; tell the user.", + "Airbyte Cloud returned HTTP 500.", + None, id="server-error", ), ], @@ -2335,9 +2327,9 @@ def test_create_connection_400_raises_input_error_without_api_message( _call_create_connection() assert exc_info.value.get_message() == ( - "Airbyte Cloud rejected the request as invalid. " - "(Cloud error: bad-request, HTTP 400)" + "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.context["source_id"] == "source-id" assert exc_info.value.context["destination_id"] == "destination-id" @@ -2372,9 +2364,7 @@ def test_create_connection_400_with_non_json_body_uses_fallback_message( with pytest.raises(AirbyteLibInputError) as exc_info: _call_create_connection() - assert exc_info.value.get_message() == ( - "Airbyte Cloud rejected the request as invalid. (HTTP 400)" - ) + assert exc_info.value.get_message() == ("Airbyte Cloud returned HTTP 400.") assert exc_info.value.__cause__ is sdk_error @@ -2728,7 +2718,7 @@ def _problem_body( return json.dumps({ "type": problem_type, "title": title, - "detail": _INTERNAL_MARKER, + "detail": "Cloud detail", "data": data, **extra, }) @@ -2737,205 +2727,53 @@ def _problem_body( @pytest.mark.parametrize( ("status_code", "body", "expected_message", "expected_guidance", "problem_type"), [ - pytest.param( - 400, - _problem_body( - "https://reference.airbyte.com/reference/errors#bad-request", - data={"message": _INTERNAL_MARKER}, - ), - "Airbyte Cloud rejected the request as invalid. " - "(Cloud error: bad-request, HTTP 400)", - "Check the arguments against the tool description.", - "bad-request", - id="bad-request", - ), - pytest.param( - 404, - _problem_body( - "https://reference.airbyte.com/reference/errors#resource-not-found", - data={"resourceType": "source", "resourceId": _INTERNAL_MARKER}, - ), - "The source was not found. (Cloud error: resource-not-found, HTTP 404)", - "Check the ID; list the resources to find the right one.", - "resource-not-found", - id="resource-not-found-with-type", - ), - pytest.param( - 401, - _problem_body( - "https://reference.airbyte.com/reference/errors#invalid-api-key" - ), - "Airbyte Cloud rejected the credentials. " - "(Cloud error: invalid-api-key, HTTP 401)", - "Ask the user to check the client ID/secret or token; don't retry.", - "invalid-api-key", - id="invalid-api-key", - ), - pytest.param( - 400, - _problem_body("error:cron-validation/invalid-expression"), - "The connection schedule is invalid. " - "(Cloud error: cron-validation/invalid-expression, HTTP 400)", - "Fix the schedule (cron expression, timezone, or frequency) and retry.", - "invalid-expression", - id="schedule-invalid", - ), - pytest.param( - 400, - _problem_body("error:mapper-validation/secret-not-found"), - "The mapper configuration is invalid. " - "(Cloud error: mapper-validation/secret-not-found, HTTP 400)", - "Fix the mapper configuration and retry. " - "If a mapper secret is missing, resend the secret values.", - "secret-not-found", - id="mapper-secret-not-found", - ), - pytest.param( - 409, - _problem_body("error:tag-already-exists"), - "A resource with this name or membership already exists. " - "(Cloud error: tag-already-exists, HTTP 409)", - "Use the existing one or pick another name; don't retry as is.", - "error:tag-already-exists", - id="already-exists", + ( + 422, + _problem_body("error:new-problem"), + "Cloud detail", + None, + "error:new-problem", ), - pytest.param( - 503, - _problem_body( - "https://reference.airbyte.com/reference/errors", - title="service-unavailable", - ), - "Airbyte Cloud is temporarily unavailable. " - "(Cloud error: service-unavailable, HTTP 503)", - "Wait and retry once.", - "service-unavailable", - id="service-unavailable", + ( + 403, + _problem_body("error:forbidden"), + "Cloud detail", + api_util.FORBIDDEN_RESOURCE_GUIDANCE, + "error:forbidden", ), - pytest.param( + (404, _problem_body("error:missing"), "Cloud detail", None, "error:missing"), + ( 429, '{"message":"You are being rate limited."}', - "Airbyte Cloud is refusing requests from these credentials " - "(rate limited). (HTTP 429)", - "Stop calling Airbyte Cloud tools and tell the user; don't retry.", - None, - id="load-shed-429", - ), - pytest.param( - 401, - '{"message":"Unauthorized"}', - "Airbyte Cloud rejected the credentials. (HTTP 401)", - "Ask the user to check the client ID/secret or token; don't retry.", + "Airbyte Cloud returned HTTP 429 (rate limited).", None, - id="non-problem-401", - ), - pytest.param( - 408, - "not json", - "Airbyte Cloud timed out. The change may or may not have been " - "applied. (HTTP 408)", - "Check the resource's current state before retrying; don't blindly " - "retry a create.", None, - id="timeout-fallback", ), - pytest.param( - 409, - "not json", - "The resource is in a state that doesn't allow this operation " - "(for example, a job is running or already finished). (HTTP 409)", - "Check the resource's current status before retrying.", - None, - id="conflict-fallback", - ), - pytest.param( - 422, - "not json", - "Airbyte Cloud rejected the request as invalid. (HTTP 422)", - "Check the arguments against the tool description.", - None, - id="other-4xx-fallback", - ), - pytest.param( - 500, - _problem_body( - "https://reference.airbyte.com/reference/errors", - title="unexpected-problem", - data={"message": _INTERNAL_MARKER}, - ), - "Airbyte Cloud hit an unexpected error. This is often caused by an " - "invalid argument value. (Cloud error: unexpected-problem, HTTP 500)", - "Check the arguments (ranges, IDs, formats). If they look right, " - "don't retry more than once; tell the user.", - "unexpected-problem", - id="unexpected-problem-hides-message", - ), - pytest.param( - 500, - '{"errorId":"123e4567-e89b-42d3-a456-426614174000"}', - "Airbyte Cloud hit an unexpected error. This is often caused by an " - "invalid argument value. (HTTP 500)", - "Check the arguments (ranges, IDs, formats). If they look right, " - "don't retry more than once; tell the user. " - "Error ID: 123e4567-e89b-42d3-a456-426614174000.", - None, - id="error-id-relayed-for-5xx", - ), - pytest.param( + ( 500, - '{"errorId":"not-a-uuid"}', - "Airbyte Cloud hit an unexpected error. This is often caused by an " - "invalid argument value. (HTTP 500)", - "Check the arguments (ranges, IDs, formats). If they look right, " - "don't retry more than once; tell the user.", + '{"errorId":"123e4567-e89b-42d3-a456-426614174000","message":"internal failure"}', + "Airbyte Cloud returned HTTP 500.", None, - id="error-id-ignored-unless-uuid", - ), - pytest.param( - 500, - json.dumps({"type": "too-big", "x": "y" * 70000}), - "Airbyte Cloud hit an unexpected error. This is often caused by an " - "invalid argument value. (HTTP 500)", - "Check the arguments (ranges, IDs, formats). If they look right, " - "don't retry more than once; tell the user.", None, - id="oversized-body-falls-back", ), ], ) -def test_wrap_sdk_error_describes_cloud_problem( +def test_wrap_sdk_error_relays_cloud_problem( status_code: int, body: str, expected_message: str, - expected_guidance: str, + expected_guidance: str | None, problem_type: str | None, ) -> None: - """Each Cloud problem maps to its fixed message; `data.message` never relays.""" - wrapped = api_util._wrap_sdk_error(_sdk_error(status_code, body)) - - assert wrapped.get_message() == expected_message + """Relay only the selected problem message, with two workflow guidance cases.""" + error = _sdk_error(status_code, body) + wrapped = api_util._wrap_sdk_error(error) + assert wrapped.message == expected_message + assert api_util.sdk_error_message(error) == expected_message assert wrapped.guidance == expected_guidance assert wrapped.context["problem_type"] == problem_type assert wrapped.context["status_code"] == status_code - assert _INTERNAL_MARKER not in wrapped.get_message() - assert _INTERNAL_MARKER not in (wrapped.guidance or "") - - -def test_wrap_sdk_error_403_keeps_fixed_forbidden_text() -> None: - error = _sdk_error( - 403, - _problem_body( - "https://reference.airbyte.com/reference/errors#forbidden", - data={"message": _INTERNAL_MARKER}, - ), - ) - wrapped = api_util._wrap_sdk_error(error) - - assert type(wrapped) is AirbyteMissingResourceError - assert wrapped.get_message() == api_util.FORBIDDEN_RESOURCE_MESSAGE - assert api_util.sdk_error_message(error) == api_util.FORBIDDEN_RESOURCE_MESSAGE - assert wrapped.guidance == api_util.FORBIDDEN_RESOURCE_GUIDANCE - assert wrapped.context["problem_type"] == "forbidden" - assert _INTERNAL_MARKER not in str(wrapped) + assert _INTERNAL_MARKER not in wrapped.message def test_public_api_json_error_context_hides_body_and_response( @@ -2972,313 +2810,8 @@ def test_public_api_json_error_context_hides_body_and_response( assert _INTERNAL_MARKER not in str(error) -# Every (status, type, title) default published in problems-api's -# api-problems.yaml, hard-coded so the test does not read the platform repo. -_PROBLEM_TYPE_DEFAULTS: list[tuple[int, str, str]] = [ - ( - 401, - "https://reference.airbyte.com/reference/errors#invalid-api-key", - "invalid-api-key", - ), - (403, "https://reference.airbyte.com/reference/errors#forbidden", "forbidden"), - (403, "error:auth/sso-required", "SSO Sign-in Required"), - (301, "error:embedded/endpoint-moved", "Airbyte Embedded Endpoint Moved"), - (403, "error:license/entitlement", "License Entitlement Error"), - (409, "error:auth/user-already-exists", "User already exists"), - (412, "error:failed-precondition", "Failed Precondition"), - ( - 500, - "https://reference.airbyte.com/reference/errors#oauth-callback-failure", - "oauth-callback-failure", - ), - ( - 500, - "https://reference.airbyte.com/reference/errors#invalid-consent-url", - "invalid-consent-url", - ), - ( - 422, - "https://reference.airbyte.com/reference/errors#invalid-redirect-url", - "invalid-redirect-url", - ), - ( - 422, - "https://reference.airbyte.com/reference/errors#unprocessable-entity", - "unprocessable-entity", - ), - (400, "https://reference.airbyte.com/reference/errors", "value-not-found"), - ( - 409, - "https://reference.airbyte.com/reference/errors#409-state-conflict", - "state-conflict", - ), - ( - 409, - "https://reference.airbyte.com/reference/errors#409-actor-not-ready", - "actor-not-ready", - ), - (409, "error:connection/locked", "Connection is locked"), - ( - 409, - "https://reference.airbyte.com/reference/errors#try-again-later", - "try-again-later", - ), - (500, "https://reference.airbyte.com/reference/errors", "unexpected-problem"), - (400, "https://reference.airbyte.com/reference/errors#bad-request", "bad-request"), - ( - 501, - "error:implementation/not-implemented-in-oss", - "API not implemented in Airbyte OSS", - ), - (400, "error:dbtcloud/access-denied", "Incorrect integration credentials"), - ( - 401, - "error:dbtcloud/paid-plan-required", - "Unable to access dbt Cloud integration", - ), - (401, "error:dbtcloud/generic", "Unable to access dbt Cloud integration"), - (400, "error:cron-validation/missing-cron-data", "Cron data is missing"), - (400, "error:cron-validation/missing-component", "Cron is missing a component"), - (400, "error:cron-validation/unsupported-timezone", "Unsupported cron timezone"), - (400, "error:cron-validation/invalid-expression", "Invalid cron expression"), - (400, "error:cron-validation/invalid-timezone", "Invalid cron timezone"), - ( - 400, - "error:cron-validation/under-one-hour-not-allowed", - "Cron sync schedules more frequent than once per hour are not allowed", - ), - ( - 400, - "error:basic-schedule-validation/under-one-hour-not-allowed", - "Basic sync schedules more frequent than once per hour are not allowed", - ), - (400, "error:mapper-validation", "Mapper validation failed"), - ( - 400, - "error:mapper-validation/missing-required-param", - "Mapper configuration missing required parameter", - ), - (400, "error:mapper-validation/invalid-config", "Invalid Mapper Configuration"), - (400, "error:mapper-validation/secret-not-found", "Mapper secret not found"), - ( - 400, - "error:connection-validation/file-transfer/connection-unsupported", - "Connection does not support file transfers", - ), - ( - 400, - "error:connection-validation/file-transfer/stream-unsupported", - "Stream does not support file transfers", - ), - ( - 400, - "error:connection-conflicting-destination-stream", - "Connection contains conflicting stream(s).", - ), - ( - 409, - "error:mapper-validation/runtime-secrets-manager-required", - "Runtime Secrets Manager Required", - ), - (503, "https://reference.airbyte.com/reference/errors", "service-unavailable"), - ( - 404, - "https://reference.airbyte.com/reference/errors#resource-not-found", - "resource-not-found", - ), - ( - 409, - "error:generate-contribution/connector-image-name-in-use", - "The image name provided is already in use.", - ), - ( - 401, - "error:generate-contribution/invalid-github-token", - "Invalid GitHub token provided.", - ), - ( - 403, - "error:generate-contribution/insufficient-github-token-permissions", - "Failed to create fork of Airbyte repository.", - ), - ( - 500, - "error:generate-contribution", - "An unexpected error occurred when creating your GitHub contribution.", - ), - ( - 404, - "error:billing/subscription/subscription-required", - "A subscription is required for this operation to succeed.", - ), - ( - 422, - "error:billing/insufficient-payment-status", - "The payment status of the associated Organization is insufficient.", - ), - ( - 422, - "error:billing/insufficient-credit-balance", - "The credit balance of the associated Workspace or Organization is insufficient.", - ), - ( - 404, - "error:billing/no-active-subscription", - "The organization doesn't have an active subscription.", - ), - ( - 400, - "error:billing/no-cancelable-subscription", - "The organization doesn't have any active cancelable subscription.", - ), - ( - 400, - "error:billing/no-scheduled-cancellation-subscription", - "The organization doesn't have any cancelable subscription with a scheduled cancellation.", - ), - ( - 400, - "error:billing/no-scheduled-plan-change", - "The organization doesn't have any self-serve subscription with a scheduled plan change.", - ), - ( - 400, - "error:connector-rollout/invalid-request", - "Invalid Connector Rollout request", - ), - ( - 400, - "error:connector-rollout/rollout-percentage-reached", - "Max rollout percentage already reached", - ), - ( - 400, - "error:connector-rollout/not-enough-actors", - "Not Enough Actors for Connector Rollout", - ), - ( - 409, - "error:dataplane-group-name-already-exists", - "Data plane group name already exists", - ), - (409, "error:dataplane-name-already-exists", "Data plane name already exists"), - (409, "error:tag-already-exists", "Tag already exists"), - (409, "error:group-already-exists", "Group already exists"), - (409, "error:group-permission-already-exists", "Group permission already exists"), - (409, "error:group-managed-by-scim", "Group managed by SCIM"), - (409, "error:group-member-already-exists", "Group member already exists"), - (400, "error:tag-invalid-hex-color", "Invalid hex color"), - (400, "error:tag-name-too-long", "Tag name too long"), - (400, "error:tag-limit-for-workspace-reached", "Tag limit for workspace reached"), - ( - 403, - "error:workspace-limit-for-organization-reached", - "Workspace limit for organization reached", - ), - (408, "error:request-timeout-exceeded", "Request timeout exceeded"), - (400, "error:notification/config/required", "Notification required"), - (400, "error:notification/config/missing-url", "Notification missing URL"), - ( - 400, - "error:destination/discover-not-supported", - "Destination does not support discover", - ), - (404, "error:destination/catalog-not-found", "Destination catalog not found"), - ( - 400, - "error:connection/destination-catalog/missing-object-name", - "Configured stream missing destination object name", - ), - ( - 400, - "error:connection/destination-catalog/invalid-operation", - "Invalid destination operation configuration", - ), - ( - 400, - "error:connection/destination-catalog/missing-required-field", - "Configured stream missing required field", - ), - ( - 400, - "error:connection/destination-catalog/invalid-additional-field", - "Invalid additional field in configured stream", - ), - ( - 400, - "error:connection/destination-catalog/required", - "Destination catalog is required", - ), - ( - 400, - "error:connection/destination-catalog/missing-primary-key", - "Primary key required when matching keys are defined", - ), - ( - 400, - "error:connection/destination-catalog/invalid-primary-key", - "Primary key must match one of the matching keys", - ), - (500, "error:sso-config-retrieval", "SSO config retrieval error"), - (500, "error:sso-setup", "SSO setup configuration error"), - (500, "error:sso-deletion", "SSO deletion failed"), - (500, "error:sso-credential-update", "SSO credential update failed"), - (500, "error:sso-activation", "SSO Activation Error"), - (401, "error:sso-token-validation", "SSO Token Validation Failed"), - ( - 503, - "error:entitlement-service/error-adding-organization", - "entitlement-service-error-adding-organization", - ), - ( - 500, - "error:entitlement-service/invalid-organization-state", - "entitlement-service-invalid-organization-state", - ), -] - - -@pytest.mark.parametrize( - ("status_code", "problem_type", "title"), - [pytest.param(*row, id=row[1]) for row in _PROBLEM_TYPE_DEFAULTS], -) -def test_every_published_problem_type_matches_a_fixed_message( - status_code: int, - problem_type: str, - title: str, -) -> None: - """Each published `type`/`title` pair hits a `cloud_errors.yaml` row.""" - body = _problem_body(problem_type, title) - wrapped = api_util._wrap_sdk_error(_sdk_error(status_code, body)) - - if problem_type.endswith("409-actor-not-ready"): - assert type(wrapped) is AirbyteConnectorNotReadyError - return - - if status_code == 403: - assert type(wrapped) is AirbyteMissingResourceError - assert wrapped.get_message() == ( - "The requested resource was not found, or these credentials can't " - "access it (HTTP 403)." - ) - assert wrapped.guidance == api_util.FORBIDDEN_RESOURCE_GUIDANCE - return - - problem = cloud_errors.parse_cloud_error(status_code, body) - assert problem.key in cloud_errors._messages() - entry = cloud_errors._messages()[problem.key] - expected_message, expected_guidance = entry["message"], entry["guidance"] - # A table hit can never be the status fallback; a few rows (bad-request, - # state-conflict, ...) intentionally share the fallback wording anyway. - assert wrapped.get_message() == ( - f"{expected_message} (Cloud error: {problem.key}, HTTP {status_code})" - ) - assert wrapped.guidance == expected_guidance - assert _INTERNAL_MARKER not in wrapped.get_message() - - @pytest.mark.parametrize( - ("problem", "expected_slug", "expected_key"), + ("problem", "expected_slug"), [ pytest.param( { @@ -3289,7 +2822,6 @@ def test_every_published_problem_type_matches_a_fixed_message( "title": "resource-not-found", }, None, - None, id="id-like-type-never-reads-title", ), pytest.param( @@ -3298,25 +2830,21 @@ def test_every_published_problem_type_matches_a_fixed_message( "title": "unexpected-problem", }, "unexpected-problem", - "unexpected-problem", id="generic-type-reads-title", ), pytest.param( {"title": "unexpected-problem"}, None, - "unexpected-problem", id="missing-type-does-not-set-slug", ), pytest.param( {"title": "password=hunter2"}, None, - "password=hunter2", id="missing-type-does-not-set-untrusted-slug", ), pytest.param( {"type": 123, "title": "try-again-later"}, None, - "try-again-later", id="non-string-type-does-not-set-slug", ), ], @@ -3324,102 +2852,221 @@ def test_every_published_problem_type_matches_a_fixed_message( def test_parse_cloud_error_title_fallback_only_for_generic_type( problem: dict[str, Any], expected_slug: str | None, - expected_key: str | None, ) -> None: """Only a generic string `type` permits a title slug.""" parsed = cloud_errors.parse_cloud_error(None, json.dumps(problem)) assert parsed.slug == expected_slug - assert parsed.key == expected_key -def test_resource_type_in_agent_text_must_be_allowlisted() -> None: - body = _problem_body( - "https://reference.airbyte.com/reference/errors#resource-not-found", - data={"resourceType": "secretvalue"}, +@pytest.mark.parametrize("detail", [None, 0, False, {}, [], "", " \n\t"]) +def test_problem_message_falls_back_to_exact_title(detail: object) -> None: + title = " Title é\nmessage " + problem = cloud_errors.parse_cloud_error( + 500, json.dumps({"detail": detail, "title": title}) ) - problem = cloud_errors.parse_cloud_error(404, body) + assert problem.message == title - message, _ = cloud_errors.describe_cloud_error(problem) - assert problem.resource_type is None - assert ( - message - == "The resource was not found. (Cloud error: resource-not-found, HTTP 404)" +@pytest.mark.parametrize("title", [None, 0, False, {}, [], "", " \n\t"]) +def test_problem_message_unusable_fields_use_transport_status(title: object) -> None: + problem = cloud_errors.parse_cloud_error( + 502, json.dumps({"detail": [], "title": title, "status": 400}) ) - assert "secretvalue" not in message + assert problem.message == "Airbyte Cloud returned HTTP 502." + assert problem.status_code == 502 -def test_cloud_errors_yaml_entries_have_message_and_guidance() -> None: - """Every YAML entry is a mapping with only `message` and `guidance` strings.""" - messages = cloud_errors._messages() +@pytest.mark.parametrize( + ("problem_type", "slug"), + [ + (None, None), + ("about:blank", "about:blank"), + ("https://h/errors#future-problem", "future-problem"), + ("https://h/errors/path-problem", "path-problem"), + ("urn:problem:future", "urn:problem:future"), + ("error:future", "error:future"), + ("error:connection/locked", "locked"), + ("https://h/errors#409-actor-not-ready", "409-actor-not-ready"), + ("1234", None), + ("x" * 101, None), + ("invalid space", None), + ("ws-97953b90-8f3a-4c1e-9d2b-0a1b2c3d4e5f", None), + ], +) +def test_unknown_type_relays_exact_detail_independently_of_slug( + problem_type: str | None, slug: str | None +) -> None: + message = " Cursor é\nwas not found. " + problem = cloud_errors.parse_cloud_error( + 422, + json.dumps({"type": problem_type, "detail": message, "title": "ignored-title"}), + ) + assert problem.message == message + assert problem.slug == slug - assert set(messages) > {"status_fallbacks"} - for key, entry in messages.items(): - if key == "status_fallbacks": - for status, fallback in entry.items(): - assert set(fallback) == {"message", "guidance"} - assert isinstance(fallback["message"], str) and fallback["message"] - assert isinstance(fallback["guidance"], str) and fallback["guidance"] - continue - assert set(entry) == {"message", "guidance"} - assert isinstance(entry["message"], str) and entry["message"] - assert isinstance(entry["guidance"], str) and entry["guidance"] + +@pytest.mark.parametrize("size", [65535, 65536, 65537]) +@pytest.mark.parametrize("character", ["a", "é"]) +def test_problem_body_byte_boundary(size: int, character: str) -> None: + prefix, suffix = '{"detail":"', '"}' + count, remainder = divmod(size - len(prefix + suffix), len(character.encode())) + message = character * count + "x" * remainder + body = prefix + message + suffix + assert len(body.encode()) == size + problem = cloud_errors.parse_cloud_error(500, body) + assert problem.message == ( + message if size <= 65536 else "Airbyte Cloud returned HTTP 500." + ) @pytest.mark.parametrize( - ("status_code", "body", "absent", "suffix"), + "body", [ - pytest.param( - 500, - _problem_body( - "https://reference.airbyte.com/reference/errors", - title="password=hunter2-secret-value", - ), - ("hunter2", "password"), - "(HTTP 500)", - id="unknown-title-never-relays", - ), - pytest.param( - 500, - _problem_body("error:some/unknown-thing contains sk_live_abc"), - ("sk_live", "unknown-thing"), - "(HTTP 500)", - id="unknown-type-never-relays", + None, + "", + "bad", + "{", + "null", + "[]", + "1", + '"text"', + '{"detail":"\ud800"}', + '{"detail":"early", "padding":"' + "x" * 65536 + '"}', + ], +) +def test_problem_invalid_body_falls_back(body: str | None) -> None: + assert ( + cloud_errors.parse_cloud_error(400, body).message + == "Airbyte Cloud returned HTTP 400." + ) + + +def test_problem_excessive_nesting_falls_back(monkeypatch: pytest.MonkeyPatch) -> None: + # Inject the decoder failure without risking a native stack overflow on Windows. + monkeypatch.setattr(cloud_errors.json, "loads", Mock(side_effect=RecursionError)) + assert ( + cloud_errors.parse_cloud_error(500, '{"detail":"x"}').message + == "Airbyte Cloud returned HTTP 500." + ) + + +@pytest.mark.parametrize( + ("status", "body", "message"), + [ + (429, None, "Airbyte Cloud returned HTTP 429 (rate limited)."), + ( + 429, + '{"message":"arbitrary secret"}', + "Airbyte Cloud returned HTTP 429 (rate limited).", ), - pytest.param( + (429, '{"title":"Slow down"}', "Slow down"), + (429, '{"detail":"Wait please"}', "Wait please"), + ( 500, - _problem_body( - "https://reference.airbyte.com/reference/errors", - title="unexpected-problem", - ), - (), - "(Cloud error: unexpected-problem, HTTP 500)", - id="known-key-shown", - ), - pytest.param( - None, - _problem_body("error:some/unknown-thing contains sk_live_abc"), - ("sk_live", "unknown-thing"), - None, - id="unknown-key-no-status-no-suffix", + '{"errorId":"invalid","message":"secret","data":{"message":"SQL secret"}}', + "Airbyte Cloud returned HTTP 500.", ), + (503, None, "Airbyte Cloud returned HTTP 503."), + (504, None, "Airbyte Cloud returned HTTP 504."), + (None, '{"message":"secret"}', "Airbyte Cloud request failed."), ], ) -def test_describe_cloud_error_suffix_only_for_known_keys( - status_code: int | None, - body: str, - absent: tuple[str, ...], - suffix: str | None, +def test_status_only_fallbacks( + status: int | None, body: str | None, message: str ) -> None: - """An unmatched `type`/`title` is never echoed into agent-facing text.""" - problem = cloud_errors.parse_cloud_error(status_code, body) - message, guidance = cloud_errors.describe_cloud_error(problem) - - for needle in absent: - assert needle not in message - assert needle not in guidance - if suffix is None: - assert "(Cloud error:" not in message and "(HTTP" not in message - else: - assert message.endswith(suffix) + assert cloud_errors.parse_cloud_error(status, body).message == message + + +@pytest.mark.parametrize( + ("status", "payload", "ready"), + [ + (409, {"title": "actor-not-ready"}, True), + (409, {"type": "https://h/errors#409-actor-not-ready"}, True), + (400, {"title": "actor-not-ready"}, False), + (409, {"title": " actor-not-ready"}, False), + (409, {"type": "error:actor-not-ready"}, False), + (409, {"type": "409-actor-not-ready-extra"}, False), + ], +) +@pytest.mark.parametrize("for_mcp", [False, True]) +def test_readiness_predicate_and_guidance( + status: int, payload: dict[str, str], ready: bool, for_mcp: bool +) -> None: + payload["detail"] = "Finish authentication first." + body = json.dumps(payload) + problem = cloud_errors.parse_cloud_error(status, body) + assert problem.message == payload["detail"] + assert problem.is_actor_not_ready is ready + assert api_util._is_actor_not_ready_error(_sdk_error(status, body)) is ready + expected = ( + ( + api_util.ACTOR_NOT_READY_MCP_GUIDANCE + if for_mcp + else api_util.ACTOR_NOT_READY_PYTHON_GUIDANCE + ) + if ready + else None + ) + assert api_util.cloud_error_guidance(problem, for_mcp=for_mcp) == expected + + +@pytest.mark.parametrize("transport", ["config", "public"]) +@pytest.mark.parametrize("status", [400, 403, 404, 409, 500]) +def test_raw_http_relay_preserves_class_cause_and_context( + monkeypatch: pytest.MonkeyPatch, + transport: str, + status: int, +) -> None: + response = requests.Response() + response.status_code = status + response.url = "https://config.example/test" + response.request = requests.Request("POST", response.url).prepare() + response._content = json.dumps({ + "detail": " Cloud message\n", + "type": "https://h/errors#409-actor-not-ready", + "title": "actor-not-ready", + }).encode() + monkeypatch.setattr(api_util.requests, "request", lambda **_: response) + monkeypatch.setattr(api_util.requests, "get", lambda *_, **__: response) + monkeypatch.setattr(api_util, "is_mcp_mode", lambda: False) + kwargs = dict( + api_root="https://api.airbyte.com/v1", + path="/test", + client_id=None, + client_secret=None, + bearer_token=SecretString("token"), + ) + with pytest.raises(AirbyteCloudError) as caught: + if transport == "config": + api_util._make_config_api_request(**kwargs, json={}) + else: + api_util._get_public_api_json(**kwargs, params=None) + error = caught.value + missing = status == 403 or (transport == "public" and status == 404) + assert type(error) is ( + AirbyteMissingResourceError if missing else AirbyteCloudError + ) + assert error.message == " Cloud message\n" + assert error.guidance == ( + api_util.ACTOR_NOT_READY_PYTHON_GUIDANCE + if status == 409 + else api_util.FORBIDDEN_RESOURCE_GUIDANCE + if status == 403 + else None + ) + assert error.context["status_code"] == status + assert error.context["problem_type"] == "409-actor-not-ready" + assert "response" not in error.context + assert "response_text" not in error.context + assert isinstance(error.__cause__, requests.HTTPError) + assert error.__cause__.response is response + + +@pytest.mark.parametrize("field", ["detail", "title", "data"]) +def test_problem_large_integer_does_not_replace_upstream_error(field: str) -> None: + body = '{"title":"Title fallback", "' + field + '":' + "9" * 5000 + "}" + expected = ( + "Airbyte Cloud returned HTTP 403." if field == "title" else "Title fallback" + ) + assert cloud_errors.parse_cloud_error(403, body).message == expected diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py index 70acb891f..e30c044c1 100644 --- a/tests/unit_tests/test_cloud_direct_execute.py +++ b/tests/unit_tests/test_cloud_direct_execute.py @@ -736,40 +736,43 @@ def test_untyped_connector_execute_resolves_kind_for_routing( assert probes == ["source"] +@pytest.mark.parametrize("status_code", [401, 403, 404, 409, 502, 503, 504]) @pytest.mark.parametrize( - ("status_code", "expected_phrase", "expects_upstream_guidance"), - [ - pytest.param(502, None, True, id="bad_gateway"), - pytest.param(503, None, True, id="service_unavailable"), - pytest.param(504, None, True, id="gateway_timeout"), - pytest.param(401, "Unauthorized", False, id="unauthorized"), - pytest.param(403, "Forbidden", False, id="forbidden"), - ], + "body", ["", '{"detail":"Cloud detail"}', '{"title":"actor-not-ready"}'] ) -def test_agent_request_error_message_and_guidance( +def test_agent_request_problem_message_and_guidance( + monkeypatch: pytest.MonkeyPatch, status_code: int, - expected_phrase: str | None, - expects_upstream_guidance: bool, + body: str, ) -> None: - """Upstream 5xx failures get non-credentials guidance and no status phrase.""" - raw_response = requests.Response() - raw_response.status_code = status_code - raw_response.url = "https://cloud.airbyte.com/api/v1/sources/source-1/execute" - - message = agents_api_util._error_message( # noqa: SLF001 - response=raw_response, full_url=raw_response.url + """Direct access uses the same message and two-case guidance contract.""" + _patch_context_layer(monkeypatch) + response = requests.Response() + response.status_code = status_code + response._content = body.encode() + monkeypatch.setattr(agents_api_util.requests, "request", lambda **_: response) + credentials = _AirbyteCredentials.from_auth(bearer_token="token") + with pytest.raises(AirbyteCloudApiError) as caught: + agents_api_util.make_cloud_agent_request( + method="GET", path="/test", credentials=credentials + ) + error = caught.value + expected = ( + "Cloud detail" + if "detail" in body + else "actor-not-ready" + if body + else f"Airbyte Cloud returned HTTP {status_code}." ) - guidance = agents_api_util._error_guidance(response=raw_response) # noqa: SLF001 - - if expected_phrase is None: - assert f"{status_code} when accessing" in message - else: - assert f"({expected_phrase})" in message - if expects_upstream_guidance: - assert guidance is not None - assert "upstream" in guidance + assert error.message == expected + assert error.status_code == status_code + assert error.context["response_text"] == body + if status_code == 403: + assert error.guidance == api_util.FORBIDDEN_RESOURCE_GUIDANCE + elif status_code == 409 and "actor-not-ready" in body: + assert error.guidance == api_util.ACTOR_NOT_READY_PYTHON_GUIDANCE else: - assert guidance is None or "upstream" not in guidance + assert error.guidance is None def test_agent_request_non_2xx_raises_cloud_api_error( @@ -797,7 +800,7 @@ def test_agent_request_non_2xx_raises_cloud_api_error( assert exc_info.value.status_code == 503 assert "(Service Unavailable)" not in exc_info.value.get_message() - assert "upstream" in (exc_info.value.guidance or "") + assert exc_info.value.guidance is None @pytest.mark.parametrize("kind", ["source", "destination"]) diff --git a/tests/unit_tests/test_mcp_datadog.py b/tests/unit_tests/test_mcp_datadog.py index 6106e23f1..d4dd4206f 100644 --- a/tests/unit_tests/test_mcp_datadog.py +++ b/tests/unit_tests/test_mcp_datadog.py @@ -291,13 +291,12 @@ def test_cloud_error_route_omits_missing_or_invalid_requests() -> None: ( 404, "resource-not-found", - "The resource was not found. (Cloud error: resource-not-found, HTTP 404)", + "SENTINEL", ), ( 403, "forbidden", - "The requested resource was not found, or these credentials can't access it " - "(HTTP 403).", + "SENTINEL", ), ], ) @@ -457,14 +456,14 @@ def test_tool_error_trace_message_uses_fixed_exception_class_messages( def test_tool_error_trace_message_keeps_cloud_precedence() -> None: error = AirbyteMissingResourceError( - message=api_util.FORBIDDEN_RESOURCE_MESSAGE, - context={"problem_type": "forbidden"}, + message="Local rewording must not reach spans", + context={"problem_type": "forbidden", "status_code": 403}, ) error.__cause__ = requests.HTTPError("private response body") assert ( _error_handling.tool_error_trace_message(error) - == api_util.FORBIDDEN_RESOURCE_MESSAGE + == "Airbyte Cloud returned HTTP 403." ) @@ -637,7 +636,7 @@ class InputModel(BaseModel): ), ], ) -def test_exception_attributes_classify_without_error_text( +def test_exception_attributes_keep_problem_text_only_in_message( error: BaseException, expected: dict[str, str] ) -> None: attrs = _datadog._exception_attributes(error) @@ -650,7 +649,9 @@ def test_exception_attributes_classify_without_error_text( expected_attrs["error.message"] = message assert attrs == expected_attrs - assert "SENTINEL" not in json.dumps(attrs) + assert "SENTINEL" not in json.dumps({ + key: value for key, value in attrs.items() if key != "error.message" + }) def test_exception_attributes_export_frames_without_error_text() -> None: @@ -1249,7 +1250,15 @@ def raised_cloud_sdk_error() -> str: response.status_code = 403 response.url = request_url response.request = requests.Request("GET", request_url).prepare() - raise SDKError("private response text", 403, "{}", response) + raise SDKError( + "private response text", + 403, + json.dumps({ + "type": "https://h/errors#forbidden", + "detail": "Denied Bearer FAKE_SECRET", + }), + response, + ) @app.tool() async def handle_error() -> str: @@ -1389,8 +1398,8 @@ async def run(): "meta" ]["metadata"] cloud_uuid = "123e4567-e89b-42d3-a456-426614174000" - assert cloud_span.get_tag("error.message") == ( - api_util.FORBIDDEN_RESOURCE_MESSAGE + assert ( + cloud_span.get_tag("error.message") == "Denied Bearer FAKE_SECRET" ) assert cloud_span.get_tag("airbyte.mcp.upstream.method") == "GET" assert cloud_span.get_tag("airbyte.mcp.upstream.route") == ( @@ -1652,7 +1661,31 @@ async def fail_tool(_ctx): ), ) ) - assert forbidden_message == api_util.FORBIDDEN_RESOURCE_MESSAGE + assert forbidden_message == "SENTINEL detail" + + for status, payload, expected in [ + ( + 500, + {"detail": " ", "title": " Title fallback é\\n"}, + " Title fallback é\\n", + ), + ( + 429, + {"message": "Do not relay"}, + "Airbyte Cloud returned HTTP 429 (rate limited).", + ), + ( + 409, + {"title": "actor-not-ready", "detail": "Authenticate first."}, + "Authenticate first.", + ), + ]: + assert ( + await raised_tool_message( + _wrapped_sdk_error(status, json.dumps(payload)) + ) + == expected + ) upstream_uuid = "123e4567-e89b-42d3-a456-426614174000" upstream_query = "x=secret-query" @@ -1663,7 +1696,7 @@ async def fail_tool(_ctx): f"?{upstream_query}", ) ) - assert upstream_message == api_util.FORBIDDEN_RESOURCE_MESSAGE + assert upstream_message == "SENTINEL detail" upstream_span = next( span for span in reversed(spans) if span.span_type == "llm" ) @@ -1688,9 +1721,7 @@ async def fail_tool(_ctx): _wrapped_sdk_error(500, json.dumps(unexpected_problem)) ) assert unexpected_message is not None - assert unexpected_message.endswith( - "(Cloud error: unexpected-problem, HTTP 500)" - ) + assert unexpected_message == "SENTINEL detail" assert error_id not in unexpected_message assert "Error ID" not in unexpected_message @@ -1699,7 +1730,7 @@ async def fail_tool(_ctx): _wrapped_sdk_error(500, _problem(_GENERIC_PROBLEM_TYPE, hostile_title)) ) assert hostile_message is not None - assert hostile_message.endswith("(HTTP 500)") + assert hostile_message == "SENTINEL detail" assert hostile_title not in hostile_message runtime_message = await raised_tool_message( diff --git a/tests/unit_tests/test_mcp_tool_errors.py b/tests/unit_tests/test_mcp_tool_errors.py index fbe346a88..16571f97a 100644 --- a/tests/unit_tests/test_mcp_tool_errors.py +++ b/tests/unit_tests/test_mcp_tool_errors.py @@ -121,7 +121,7 @@ def test_format_user_facing_error() -> None: assert format_user_facing_error(ValueError("plain")) == "plain" -def test_direct_cloud_api_error_uses_fixed_table_message_in_agent_text() -> None: +def test_direct_cloud_api_error_relays_problem_message_in_agent_text() -> None: error = AirbyteCloudApiError( status_code=404, message="Request failed at https://config.example/connections/secret-id", @@ -129,6 +129,7 @@ def test_direct_cloud_api_error_uses_fixed_table_message_in_agent_text() -> None "full_url": "https://config.example/connections/secret-id", "response_text": json.dumps({ "type": "https://reference.airbyte.com/reference/errors#resource-not-found", + "detail": "Connection was removed.", "data": { "resourceType": "connection", "message": "SENTINEL_RESPONSE_TEXT", @@ -141,14 +142,13 @@ def test_direct_cloud_api_error_uses_fixed_table_message_in_agent_text() -> None text = agent_error_text(exc_info.value) - assert "The connection was not found." in text - assert "resource-not-found, HTTP 404" in text + assert text == "Connection was removed." assert "https://config.example" not in text assert "SENTINEL_RESPONSE_TEXT" not in text assert "response_text" not in text -def test_format_direct_cloud_api_error_keeps_fixed_forbidden_guidance() -> None: +def test_format_direct_cloud_api_error_uses_general_forbidden_guidance() -> None: guidance = ( "Authentication succeeded but access was denied; the workspace or connector " "may not be enabled for agent access in Airbyte Cloud." @@ -162,7 +162,11 @@ def test_format_direct_cloud_api_error_keeps_fixed_forbidden_guidance() -> None: text = format_user_facing_error(error) - assert guidance in text + assert ( + text + == "Airbyte Cloud returned HTTP 403. " + api_util.FORBIDDEN_RESOURCE_GUIDANCE + ) + assert guidance not in text assert "https://config.example" not in text assert "full_url" not in text @@ -340,8 +344,7 @@ def broken() -> str: text = _call_error(server, "broken") - assert "Airbyte Cloud hit an unexpected error" in text - assert "unexpected-problem, HTTP 500" in text + assert text == "unexpected-problem" assert "INTERNAL_MARKER" not in text assert "Status Code" not in text assert "api.airbyte.com" not in text @@ -433,3 +436,161 @@ def test_redact_agent_text_keeps_connection_error_id_and_url() -> None: ) assert redact_agent_text(text) == text + + +@pytest.mark.parametrize( + ("status", "payload", "message", "guidance"), + [ + (422, {"detail": " Cursor é\nmissing. "}, " Cursor é\nmissing. ", None), + ( + 500, + {"detail": " ", "title": " Temporarily unavailable\n"}, + " Temporarily unavailable\n", + None, + ), + ( + 409, + {"title": "actor-not-ready", "detail": "Finish authentication. "}, + "Finish authentication. ", + api_util.ACTOR_NOT_READY_MCP_GUIDANCE, + ), + ( + 403, + {"detail": "Resource is unavailable. "}, + "Resource is unavailable. ", + api_util.FORBIDDEN_RESOURCE_GUIDANCE, + ), + (404, {"detail": "Connection was removed."}, "Connection was removed.", None), + ( + 429, + {"message": "ignored non-problem text"}, + "Airbyte Cloud returned HTTP 429 (rate limited).", + None, + ), + ], +) +@pytest.mark.parametrize( + "carrier", ["sdk", "wrapped", "nested", "direct", "bodyless-direct", "http"] +) +def test_cloud_chain_uses_exact_message_for_agent_and_span( + status: int, + payload: dict[str, str], + message: str, + guidance: str | None, + carrier: str, +) -> None: + from airbyte.mcp._error_handling import cloud_error_trace_message + + body = json.dumps(payload) + response = requests.Response() + response.status_code = status + response._content = body.encode() + sdk = SDKError("ignored SDK message", status, body, response) + error: BaseException = sdk + if carrier in {"wrapped", "nested"}: + error = api_util._wrap_sdk_error(sdk) + error.__cause__ = sdk + if carrier == "nested": + outer = AirbyteMissingResourceError( + message="local override", guidance="local advice" + ) + outer.__cause__ = error + error = outer + elif carrier in {"direct", "bodyless-direct"}: + error = AirbyteCloudApiError( + status_code=status, + guidance="ignored advice", + context={"response_text": body} if carrier == "direct" else {}, + ) + if carrier == "bodyless-direct": + error.__cause__ = sdk + elif carrier == "http": + error = api_util.AirbyteCloudError( + message="ignored rewording", + context={"problem_type": None, "status_code": status}, + ) + error.__cause__ = requests.HTTPError(response=response) + expected = f"{message} {guidance}".rstrip() if guidance else message + assert format_user_facing_error(error) == expected + tool = ToolError("ignored wrapper") + tool.__cause__ = error + assert agent_error_text(tool) == expected + assert cloud_error_trace_message(tool) == message + + +@pytest.mark.parametrize("field", ["detail", "title"]) +def test_real_middleware_redacts_selected_cloud_message_after_nested_rewrap( + field: str, +) -> None: + server = _server_with_agent_error_text() + + @server.tool + def broken() -> str: + response = requests.Response() + response.status_code = 403 + sdk = SDKError( + "ignored", 403, json.dumps({field: "Denied Bearer FAKE_SECRET"}), response + ) + wrapped = api_util._wrap_sdk_error(sdk) + wrapped.__cause__ = sdk + raise AirbyteMissingResourceError(message="local rewording") from wrapped + + text = _call_error(server, "broken") + assert text == "Denied Bearer " + api_util.FORBIDDEN_RESOURCE_GUIDANCE + assert "FAKE_SECRET" not in text + + +def test_unrelated_http_error_is_not_a_cloud_carrier() -> None: + from airbyte.mcp._error_handling import cloud_error_trace_message + + response = requests.Response() + response.status_code = 500 + response._content = b'{"detail":"unrelated HTTP body"}' + error = AirbyteLibInputError(message="local input error") + error.__cause__ = requests.HTTPError(response=response) + assert cloud_error_trace_message(error) is None + assert format_user_facing_error(error) == "local input error" + + +@pytest.mark.parametrize("status", [None, 403, 429, 502]) +def test_bodyless_direct_error_has_only_status_fallback(status: int | None) -> None: + from airbyte.mcp._error_handling import cloud_error_trace_message + + error = AirbyteCloudApiError( + status_code=status, + message="ignored local wording", + guidance="ignored local guidance", + ) + message = ( + "Airbyte Cloud request failed." + if status is None + else "Airbyte Cloud returned HTTP 429 (rate limited)." + if status == 429 + else f"Airbyte Cloud returned HTTP {status}." + ) + expected = ( + message + " " + api_util.FORBIDDEN_RESOURCE_GUIDANCE + if status == 403 + else message + ) + assert format_user_facing_error(error) == expected + tool = ToolError("outer text") + tool.__cause__ = error + assert agent_error_text(tool) == expected + assert cloud_error_trace_message(tool) == message + + +def test_cloud_chain_remains_bounded_and_tolerates_cycles() -> None: + from airbyte.mcp._error_handling import cloud_error_trace_message + + error = ToolError("local") + error.__cause__ = error + assert cloud_error_trace_message(error) is None + assert agent_error_text(error) == "local" + cloud = AirbyteCloudApiError(status_code=500) + outer: BaseException = cloud + for _ in range(5): + wrapped = ToolError("outer") + wrapped.__cause__ = outer + outer = wrapped + assert cloud_error_trace_message(outer) is None From 830c828fb247abb0a2131833214acb4375846d4e Mon Sep 17 00:00:00 2001 From: Caleb Evans Date: Mon, 12 Oct 2026 20:07:35 -0700 Subject: [PATCH 2/8] test: keep oversized Cloud body case ID short on Windows --- tests/unit_tests/test_cloud_api_util.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index 0a4f96786..9c19eb0bf 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/tests/unit_tests/test_cloud_api_util.py @@ -2932,7 +2932,10 @@ def test_problem_body_byte_boundary(size: int, character: str) -> None: "1", '"text"', '{"detail":"\ud800"}', - '{"detail":"early", "padding":"' + "x" * 65536 + '"}', + pytest.param( + '{"detail":"early", "padding":"' + "x" * 65536 + '"}', + id="oversized-body", + ), ], ) def test_problem_invalid_body_falls_back(body: str | None) -> None: From dc174fdeca261a7497347e9c5045b99596df0cfe Mon Sep 17 00:00:00 2001 From: Caleb Evans Date: Mon, 12 Oct 2026 21:52:15 -0700 Subject: [PATCH 3/8] fix: reject non-standard JSON constants in Cloud error bodies --- airbyte/_util/cloud_errors.py | 10 ++++++++-- tests/unit_tests/test_cloud_api_util.py | 20 ++++++++++++++++++++ 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/airbyte/_util/cloud_errors.py b/airbyte/_util/cloud_errors.py index 174ee7c28..7b615f003 100644 --- a/airbyte/_util/cloud_errors.py +++ b/airbyte/_util/cloud_errors.py @@ -7,7 +7,7 @@ import re from dataclasses import dataclass from http import HTTPStatus -from typing import Any +from typing import Any, NoReturn _PROBLEM_SLUG = re.compile(r"[a-z0-9][a-z0-9:._-]{0,99}") @@ -77,6 +77,10 @@ def _status_fallback_message(status_code: int | None) -> str: return "Airbyte Cloud request failed." +def _reject_json_constant(value: str) -> NoReturn: + raise json.JSONDecodeError("Invalid JSON constant", value, 0) + + 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 @@ -84,7 +88,9 @@ def parse_cloud_error(status_code: int | None, body: str | None) -> CloudErrorIn 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) + payload = json.loads( + body, parse_int=lambda _: None, parse_constant=_reject_json_constant + ) except (UnicodeEncodeError, json.JSONDecodeError, RecursionError): pass problem = payload if isinstance(payload, dict) else {} diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index 9c19eb0bf..7d6f615e3 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/tests/unit_tests/test_cloud_api_util.py @@ -2945,6 +2945,26 @@ def test_problem_invalid_body_falls_back(body: str | None) -> None: ) +@pytest.mark.parametrize("constant", ["NaN", "Infinity", "-Infinity"]) +def test_problem_non_json_constants_fall_back(constant: str) -> None: + body = ( + '{"detail":"relay me", "title":"actor-not-ready", ' + '"type":"error:409-actor-not-ready", "data":[' + constant + "]}" + ) + assert cloud_errors.parse_cloud_error(409, body) == cloud_errors.CloudErrorInfo( + status_code=409, + slug=None, + message="Airbyte Cloud returned HTTP 409.", + is_actor_not_ready=False, + ) + + +@pytest.mark.parametrize("constant", ["NaN", "Infinity", "-Infinity"]) +def test_problem_quoted_constants_remain_valid_strings(constant: str) -> None: + body = json.dumps({"detail": constant}) + assert cloud_errors.parse_cloud_error(400, body).message == constant + + def test_problem_excessive_nesting_falls_back(monkeypatch: pytest.MonkeyPatch) -> None: # Inject the decoder failure without risking a native stack overflow on Windows. monkeypatch.setattr(cloud_errors.json, "loads", Mock(side_effect=RecursionError)) From c2f1c929d1a0de9ea7b1829cb9f68ec5861ecf81 Mon Sep 17 00:00:00 2001 From: Caleb Evans Date: Mon, 12 Oct 2026 22:21:37 -0700 Subject: [PATCH 4/8] fix: reject unpaired surrogates in Cloud problem messages --- airbyte/_util/cloud_errors.py | 9 +++++++-- tests/unit_tests/test_cloud_api_util.py | 20 ++++++++++++++++++++ tests/unit_tests/test_mcp_tool_errors.py | 16 ++++++++++++++++ 3 files changed, 43 insertions(+), 2 deletions(-) diff --git a/airbyte/_util/cloud_errors.py b/airbyte/_util/cloud_errors.py index 7b615f003..fec2e161d 100644 --- a/airbyte/_util/cloud_errors.py +++ b/airbyte/_util/cloud_errors.py @@ -65,6 +65,8 @@ def _problem_message(payload: dict[str, Any]) -> str | None: for field in ("detail", "title"): value = payload.get(field) if isinstance(value, str) and value.strip(): + # JSON escapes can decode to unpaired surrogates. + value.encode("utf-8") return value return None @@ -84,6 +86,7 @@ def _reject_json_constant(value: str) -> NoReturn: 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: @@ -91,14 +94,16 @@ def parse_cloud_error(status_code: int | None, body: str | None) -> CloudErrorIn 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): - pass + payload = None problem = payload if isinstance(payload, dict) else {} problem_type = problem.get("type") return CloudErrorInfo( status_code=status_code, slug=_problem_slug(problem), - message=_problem_message(problem) or _status_fallback_message(status_code), + message=message or _status_fallback_message(status_code), is_actor_not_ready=status_code == HTTPStatus.CONFLICT and ( problem.get("title") == "actor-not-ready" diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index 7d6f615e3..f639d3356 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/tests/unit_tests/test_cloud_api_util.py @@ -2965,6 +2965,26 @@ def test_problem_quoted_constants_remain_valid_strings(constant: str) -> None: assert cloud_errors.parse_cloud_error(400, body).message == constant +@pytest.mark.parametrize("field", ["detail", "title"]) +@pytest.mark.parametrize("surrogate", ["\ud800", "\udfff", "\udfff\ud800"]) +def test_problem_escaped_lone_surrogates_fall_back(field: str, surrogate: str) -> None: + body = json.dumps({field: surrogate, "type": "error:409-actor-not-ready"}) + assert body.isascii() + assert cloud_errors.parse_cloud_error(409, body) == cloud_errors.CloudErrorInfo( + status_code=409, + slug=None, + message="Airbyte Cloud returned HTTP 409.", + is_actor_not_ready=False, + ) + + +@pytest.mark.parametrize("field", ["detail", "title"]) +def test_problem_valid_surrogate_pairs_preserve_unicode(field: str) -> None: + body = json.dumps({field: " \U0001f680\n"}) + assert body.isascii() + assert cloud_errors.parse_cloud_error(400, body).message == " \U0001f680\n" + + def test_problem_excessive_nesting_falls_back(monkeypatch: pytest.MonkeyPatch) -> None: # Inject the decoder failure without risking a native stack overflow on Windows. monkeypatch.setattr(cloud_errors.json, "loads", Mock(side_effect=RecursionError)) diff --git a/tests/unit_tests/test_mcp_tool_errors.py b/tests/unit_tests/test_mcp_tool_errors.py index 16571f97a..7012dcbb1 100644 --- a/tests/unit_tests/test_mcp_tool_errors.py +++ b/tests/unit_tests/test_mcp_tool_errors.py @@ -540,6 +540,22 @@ def broken() -> str: assert "FAKE_SECRET" not in text +@pytest.mark.parametrize("field", ["detail", "title"]) +def test_real_middleware_serializes_fallback_for_escaped_lone_surrogate( + field: str, +) -> None: + server = _server_with_agent_error_text() + + @server.tool + def broken() -> str: + response = requests.Response() + response.status_code = 500 + sdk = SDKError("ignored", 500, json.dumps({field: "\ud800"}), response) + raise api_util._wrap_sdk_error(sdk) from sdk + + assert _call_error(server, "broken") == "Airbyte Cloud returned HTTP 500." + + def test_unrelated_http_error_is_not_a_cloud_carrier() -> None: from airbyte.mcp._error_handling import cloud_error_trace_message From 64f0795dcf570e4ad1c06c09df258bcd071b1a6d Mon Sep 17 00:00:00 2001 From: Caleb Evans Date: Mon, 12 Oct 2026 23:09:23 -0700 Subject: [PATCH 5/8] fix: strictly decode raw Cloud error responses as UTF-8 --- airbyte/_direct_connectors/api_util.py | 8 ++- airbyte/_util/api_util.py | 14 +++- airbyte/mcp/_error_handling.py | 3 +- tests/unit_tests/test_mcp_tool_errors.py | 90 ++++++++++++++++++++++++ 4 files changed, 110 insertions(+), 5 deletions(-) diff --git a/airbyte/_direct_connectors/api_util.py b/airbyte/_direct_connectors/api_util.py index 8cf7514bf..dfcaec795 100644 --- a/airbyte/_direct_connectors/api_util.py +++ b/airbyte/_direct_connectors/api_util.py @@ -27,6 +27,7 @@ AIRBYTE_CLOUD_ANALYTIC_SOURCE_HEADER, CLOUD_API_ROOT, cloud_error_guidance, + decode_cloud_error_body, get_bearer_token, get_cloud_api_analytic_source, get_config_api_root, @@ -140,7 +141,8 @@ def make_cloud_agent_request( timeout=_REQUEST_TIMEOUT_SECONDS, ) if not status_ok(response.status_code): - problem = parse_cloud_error(response.status_code, response.text) + body = decode_cloud_error_body(response) + problem = parse_cloud_error(response.status_code, body) raise AirbyteCloudApiError( status_code=response.status_code, message=problem.message, @@ -149,7 +151,9 @@ def make_cloud_agent_request( "full_url": full_url, "path": path, "status_code": response.status_code, - "response_text": response.text, + # Retain invalid response text for diagnostics, never for message relay. + "response_text": body if body is not None else response.text, + "response_body_is_utf8": body is not None, }, ) diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py index e2a87f397..ffc000d54 100644 --- a/airbyte/_util/api_util.py +++ b/airbyte/_util/api_util.py @@ -158,11 +158,21 @@ def _get_sdk_error_context(error: SDKError) -> dict[str, Any]: return context +def decode_cloud_error_body(response: requests.Response) -> str | None: + """Decode Cloud error bytes as UTF-8 without charset guessing or replacement.""" + try: + return (response.content or b"").decode("utf-8") + except UnicodeDecodeError: + return None + + def error_response_body(error: BaseException) -> str | None: """Return the raw response body an Airbyte API error carries, if any.""" if isinstance(error, SDKError): return error.body if isinstance(error, AirbyteCloudApiError): + if (error.context or {}).get("response_body_is_utf8") is False: + return None body = (error.context or {}).get("response_text") return body if isinstance(body, str) else None return None @@ -2334,7 +2344,7 @@ def _make_config_api_request( try: response.raise_for_status() except requests.HTTPError as ex: - problem = parse_cloud_error(response.status_code, response.text) + problem = parse_cloud_error(response.status_code, decode_cloud_error_body(response)) message = problem.message guidance = cloud_error_guidance(problem, for_mcp=is_mcp_mode()) error_context = { @@ -2419,7 +2429,7 @@ def _get_public_api_json( try: response.raise_for_status() except requests.HTTPError as ex: - problem = parse_cloud_error(response.status_code, response.text) + problem = parse_cloud_error(response.status_code, decode_cloud_error_body(response)) message = problem.message guidance = cloud_error_guidance(problem, for_mcp=is_mcp_mode()) error_context = { diff --git a/airbyte/mcp/_error_handling.py b/airbyte/mcp/_error_handling.py index 5febdf2d3..6d16c1551 100644 --- a/airbyte/mcp/_error_handling.py +++ b/airbyte/mcp/_error_handling.py @@ -17,6 +17,7 @@ from airbyte._util.api_util import ( SDKError, cloud_error_guidance, + decode_cloud_error_body, error_response_body, ) from airbyte._util.cloud_errors import ( @@ -115,7 +116,7 @@ def _cloud_error_info(error: BaseException) -> CloudErrorInfo | None: ): response = cause.__cause__.response if response is not None: - return parse_cloud_error(response.status_code, response.text) + return parse_cloud_error(response.status_code, decode_cloud_error_body(response)) status = (cause.context or {}).get("status_code") return parse_cloud_error(status if isinstance(status, int) else None, None) return fallback diff --git a/tests/unit_tests/test_mcp_tool_errors.py b/tests/unit_tests/test_mcp_tool_errors.py index 7012dcbb1..63ea31e45 100644 --- a/tests/unit_tests/test_mcp_tool_errors.py +++ b/tests/unit_tests/test_mcp_tool_errors.py @@ -610,3 +610,93 @@ def test_cloud_chain_remains_bounded_and_tolerates_cycles() -> None: wrapped.__cause__ = outer outer = wrapped assert cloud_error_trace_message(outer) is None + + +@pytest.mark.parametrize("transport", ["config", "public", "direct"]) +@pytest.mark.parametrize( + ("content", "encoding", "expected"), + [ + ( + b'{"detail":"bad\xff","type":"error:future"}', + None, + "Airbyte Cloud returned HTTP 500.", + ), + ( + b'{"detail":"bad\xff","type":"error:future"}', + "utf-8", + "Airbyte Cloud returned HTTP 500.", + ), + ( + b'{"detail":"caf\xe9","type":"error:future"}', + "iso-8859-1", + "Airbyte Cloud returned HTTP 500.", + ), + (b'{"detail":"caf\xc3\xa9","type":"error:future"}', "iso-8859-1", "café"), + (b'{"detail":"caf\xc3\xa9","type":"error:future"}', None, "café"), + ], +) +def test_raw_cloud_errors_validate_utf8_before_python_agent_and_trace_relay( + monkeypatch: pytest.MonkeyPatch, + transport: str, + content: bytes, + encoding: str | None, + expected: str, +) -> None: + from airbyte._direct_connectors import api_util as direct_api_util + from airbyte.cloud._credentials import _AirbyteCredentials + from airbyte.mcp._error_handling import cloud_error_trace_message + + response = requests.Response() + response.status_code = 500 + response._content = content + response.encoding = encoding + response.headers["Content-Type"] = "application/problem+json" + ( + f"; charset={encoding}" if encoding else "" + ) + response.request = requests.Request( + "GET", "https://api.airbyte.com/v1/test" + ).prepare() + monkeypatch.setattr(api_util.requests, "request", lambda **_: response) + monkeypatch.setattr(api_util.requests, "get", lambda *_, **__: response) + monkeypatch.setattr( + direct_api_util.deployment, "is_agents_api_available", lambda **_: True + ) + kwargs = dict( + api_root="https://api.airbyte.com/v1", + path="/test", + client_id=None, + client_secret=None, + bearer_token=api_util.SecretString("token"), + ) + with pytest.raises(api_util.AirbyteCloudError) as caught: + if transport == "config": + api_util._make_config_api_request(**kwargs, json={}) + elif transport == "public": + api_util._get_public_api_json(**kwargs, params=None) + else: + direct_api_util.make_cloud_agent_request( + method="GET", + path="/test", + credentials=_AirbyteCredentials.from_auth( + bearer_token="token", env_vars=False + ), + ) + error = caught.value + assert error.message == expected + assert error.guidance is None + assert format_user_facing_error(error) == expected + tool = ToolError("outer wrapper") + tool.__cause__ = error + assert agent_error_text(tool) == expected + assert cloud_error_trace_message(tool) == expected + valid_utf8 = expected == "café" + assert mcp_tool_error_reason(tool) == ("error:future" if valid_utf8 else None) + if transport == "direct": + assert error.context["response_body_is_utf8"] is valid_utf8 + assert error.context["response_text"] == ( + content.decode("utf-8") if valid_utf8 else response.text + ) + else: + assert "response_text" not in error.context + assert isinstance(error.__cause__, requests.HTTPError) + assert error.__cause__.response is response From 85054290f070571fba7c88a6075e079b647022d6 Mon Sep 17 00:00:00 2001 From: Caleb Evans Date: Mon, 12 Oct 2026 23:35:23 -0700 Subject: [PATCH 6/8] fix(mcp): redact caught Cloud errors and validate SDK bytes --- AGENTS.md | 3 +- CONTRIBUTING.md | 3 +- airbyte/_util/api_util.py | 12 ++-- airbyte/mcp/_error_handling.py | 2 +- airbyte/mcp/cloud.py | 9 ++- tests/unit_tests/test_cloud_api_util.py | 57 +++++++++++++++ tests/unit_tests/test_mcp_cloud.py | 88 ++++++++++++++++++++++++ tests/unit_tests/test_mcp_tool_errors.py | 17 +++-- 8 files changed, 176 insertions(+), 15 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index a27c3e255..3dc0d21b2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -27,7 +27,8 @@ PyAirbyte preserves the selected string; fix poor Cloud wording in the platform. Only actor-not-ready setup instructions and 403 resource/access guidance are added. Bodies must be JSON objects of at most 65,536 UTF-8 bytes; invalid or non-problem responses use a status-only fallback. `message`, `data.message`, and `errorId` are not displayed. -Agent text still passes through credential redaction. Native Datadog uses the selected, +Agent errors, caught-error billing messages, and connector warnings pass through credential +redaction. Native Datadog uses the selected, unredacted Cloud message without guidance; this does not add it to LLM Obs metadata. Non-Cloud MCP trace messages are the first line of the exception class's docstring diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index b3e1c1a9f..aab6973c3 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -32,7 +32,8 @@ PyAirbyte preserves the selected string; fix poor Cloud wording in the platform. Only actor-not-ready setup instructions and 403 resource/access guidance are added. Bodies must be JSON objects of at most 65,536 UTF-8 bytes; invalid or non-problem responses use a status-only fallback. `message`, `data.message`, and `errorId` are not displayed. -Agent text still passes through credential redaction. Native Datadog uses the selected, +Agent errors, caught-error billing messages, and connector warnings pass through credential +redaction. Native Datadog uses the selected, unredacted Cloud message without guidance; this does not add it to LLM Obs metadata. Non-Cloud MCP trace messages are the first line of the exception class's docstring diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py index ffc000d54..2be3b717f 100644 --- a/airbyte/_util/api_util.py +++ b/airbyte/_util/api_util.py @@ -169,6 +169,8 @@ def decode_cloud_error_body(response: requests.Response) -> str | None: def error_response_body(error: BaseException) -> str | None: """Return the raw response body an Airbyte API error carries, if any.""" if isinstance(error, SDKError): + if isinstance(getattr(error.raw_response, "content", None), bytes): + return decode_cloud_error_body(error.raw_response) return error.body if isinstance(error, AirbyteCloudApiError): if (error.context or {}).get("response_body_is_utf8") is False: @@ -181,18 +183,18 @@ def error_response_body(error: BaseException) -> str | None: def sdk_error_response(error: BaseException) -> tuple[int | None, str | None] | None: """Return the status code and body carried by an SDKError.""" if isinstance(error, SDKError): - return error.status_code, error.body + return error.status_code, error_response_body(error) return None def _is_actor_not_ready_error(error: SDKError) -> bool: """Check whether an SDKError contains Airbyte Cloud's actor-not-ready problem.""" - return parse_cloud_error(error.status_code, error.body).is_actor_not_ready + return parse_cloud_error(error.status_code, error_response_body(error)).is_actor_not_ready def sdk_error_message(error: SDKError) -> str: """Return the selected Cloud problem message or transport status fallback.""" - return parse_cloud_error(error.status_code, error.body).message + return parse_cloud_error(error.status_code, error_response_body(error)).message def cloud_error_guidance(problem: CloudErrorInfo, *, for_mcp: bool) -> str | None: @@ -213,7 +215,7 @@ def _wrap_sdk_error( full URL context, making it easier to debug API issues like 403 and 404 errors. """ sdk_context = _get_sdk_error_context(error) - problem = parse_cloud_error(error.status_code, error.body) + problem = parse_cloud_error(error.status_code, error_response_body(error)) merged_context = { **(base_context or {}), **sdk_context, @@ -2001,7 +2003,7 @@ def create_connection( # noqa: PLR0913 # Too many arguments } if e.status_code == HTTPStatus.BAD_REQUEST: raise AirbyteLibInputError( - message=parse_cloud_error(e.status_code, e.body).message, + message=sdk_error_message(e), context={**base_context, **_get_sdk_error_context(e)}, ) from e raise _wrap_sdk_error(e, base_context) from e diff --git a/airbyte/mcp/_error_handling.py b/airbyte/mcp/_error_handling.py index 6d16c1551..6c09d7283 100644 --- a/airbyte/mcp/_error_handling.py +++ b/airbyte/mcp/_error_handling.py @@ -102,7 +102,7 @@ def _cloud_error_info(error: BaseException) -> CloudErrorInfo | None: fallback = None for cause in _chain(error): if isinstance(cause, SDKError): - return parse_cloud_error(cause.status_code, cause.body) + return parse_cloud_error(cause.status_code, error_response_body(cause)) if isinstance(cause, AirbyteCloudApiError): body = error_response_body(cause) problem = parse_cloud_error(cause.status_code, body) diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 5ccaf0b18..31d558579 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -87,6 +87,7 @@ CloudConnectorDocsResult, render_connector_docs_result, ) +from airbyte.mcp._error_handling import redact_agent_text from airbyte.mcp._scope import record_default_workspace, record_resolved_organization from airbyte.mcp._tool_utils import ( AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET, @@ -197,7 +198,9 @@ def _feature_lookup_warning(error: Exception) -> str: "unknown. This is a permission problem: check that the credentials have " "access to this workspace and connector." ) - return f"Connector feature lookup failed; enabled features are unknown: {error}" + return redact_agent_text( + f"Connector feature lookup failed; enabled features are unknown: {error}" + ) FEATURES_NOT_CHECKED: Final = "not_checked" @@ -1461,7 +1464,7 @@ def _describe_cloud_connector( # noqa: PLR0912 # Too many branches except (AirbyteLibError, requests.RequestException) as error: warnings.append(f"Data replication docs are unavailable: {error}") - result.warnings = warnings + result.warnings = [redact_agent_text(warning) for warning in warnings] return result @@ -2939,7 +2942,7 @@ def get_cloud_organization_billing_status( organization_id=org.organization_id, organization_name=org.organization_name, billing_info_available=False, - message=f"Billing information could not be retrieved: {reason}", + message=redact_agent_text(f"Billing information could not be retrieved: {reason}"), ) return CloudOrganizationBillingStatusResult( organization_id=org.organization_id, diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index f639d3356..b9d43f15b 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/tests/unit_tests/test_cloud_api_util.py @@ -3113,3 +3113,60 @@ def test_problem_large_integer_does_not_replace_upstream_error(field: str) -> No "Airbyte Cloud returned HTTP 403." if field == "title" else "Title fallback" ) assert cloud_errors.parse_cloud_error(403, body).message == expected + + +@pytest.mark.parametrize( + ("content", "expected", "ready"), + [ + ( + b'{"title":"actor-not-ready","detail":"bad\xff"}', + "Airbyte Cloud returned HTTP 409.", + False, + ), + (b"", "Airbyte Cloud returned HTTP 409.", False), + ( + b'{"title":"actor-not-ready","detail":"Finish setup."}', + "Finish setup.", + True, + ), + ], +) +def test_sdk_raw_bytes_control_readiness( + content: bytes, expected: str, ready: bool +) -> None: + response = requests.Response() + response.status_code = 409 + response._content = content + error = SDKError( + "ignored", + 409, + '{"title":"actor-not-ready","detail":"Already decoded SDK text"}', + response, + ) + assert api_util._is_actor_not_ready_error(error) is ready + wrapped = api_util._wrap_sdk_error(error) + assert wrapped.message == expected + assert isinstance(wrapped, api_util.AirbyteConnectorNotReadyError) is ready + assert (wrapped.guidance is not None) is ready + + +def test_create_connection_rejects_invalid_sdk_response_bytes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + response = requests.Response() + response.status_code = 400 + response._content = b'{"detail":"bad\xff"}' + response.encoding = "iso-8859-1" + error = SDKError("ignored", 400, response.text, response) + monkeypatch.setattr( + api_util, + "get_airbyte_server_instance", + lambda **_: SimpleNamespace( + connections=SimpleNamespace(create_connection=Mock(side_effect=error)) + ), + ) + with pytest.raises(AirbyteLibInputError) as caught: + _call_create_connection() + assert caught.value.message == "Airbyte Cloud returned HTTP 400." + assert caught.value.guidance is None + assert caught.value.__cause__ is error diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index 296f24fa5..c03950660 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -2886,3 +2886,91 @@ def enabled_features(self, value: frozenset[ConnectorFeature]) -> None: assert result.enabled_features == cloud_mcp.FEATURES_UNKNOWN assert any("enabled features are unknown" in w for w in result.warnings) + + +@pytest.mark.parametrize("field", ["detail", "title"]) +@pytest.mark.parametrize("result_path", ["billing", "feature-warning", "description"]) +def test_successful_results_redact_caught_cloud_error_credentials( + monkeypatch: pytest.MonkeyPatch, field: str, result_path: str +) -> None: + """Caught errors remain successful results with credentials masked in both encodings.""" + import json + + from airbyte_api.errors import SDKError + from fastmcp import Client + from fastmcp_extensions import mcp_server + + from airbyte._util import api_util + from airbyte.mcp._error_handling import ( + AgentErrorTextMiddleware, + cloud_error_trace_message, + ) + + message = ( + "Denied Bearer FAKE_SECRET at " + "https://fake-user:fake-password@example.com/path?token=fake-query" + ) + response = requests.Response() + response.status_code = 500 + response._content = json.dumps({field: message}).encode() + sdk = SDKError("ignored", 500, response.text, response) + error = api_util._wrap_sdk_error(sdk) + error.__cause__ = sdk + + def fail(*_args: object, **_kwargs: object) -> None: + raise error + + organization = SimpleNamespace( + organization_id="org-id", + organization_name="Organization", + get_billing_status=fail, + ) + monkeypatch.setattr( + cloud_mcp, + "_get_cloud_client", + lambda _: SimpleNamespace(get_organization=lambda **_: organization), + ) + monkeypatch.setattr(connector_docs, "build_connection_details", fail) + connector = _DescribedConnector( + enabled_features=frozenset({ConnectorFeature.DIRECT_ACCESS}) + ) + connector.workspace = SimpleNamespace(_has_context_layer_api=lambda: True) + connector._integration_name = error + connector.inspect_error = error + connector.config = error + connector.guidance = error + connector.replication_docs = error + server = mcp_server("test", telemetry=False) + server.add_middleware(AgentErrorTextMiddleware()) + + @server.tool + def caught_error_result() -> dict[str, Any]: + if result_path == "billing": + return cloud_mcp.get_cloud_organization_billing_status( + cast(Context, object()), organization_id=None, organization_name=None + ).model_dump() + if result_path == "feature-warning": + return {"warnings": [cloud_mcp._feature_lookup_warning(error)]} + result = _describe( + connector, + with_config=True, + with_replication_details=True, + with_direct_access_guidance=True, + with_data_replication_docs=True, + ) + assert len(result.warnings) == 6 + return result.model_dump() + + async def call_tool() -> None: + async with Client(server) as client: + result = await client.call_tool("caught_error_result") + assert result.is_error is False + assert result.structured_content is not None + for rendered in (str(result.content), json.dumps(result.structured_content)): + assert "Denied Bearer " in rendered + for secret in ("FAKE_SECRET", "fake-user", "fake-password", "fake-query"): + assert secret not in rendered + + asyncio.run(call_tool()) + assert error.message == message + assert cloud_error_trace_message(error) == message diff --git a/tests/unit_tests/test_mcp_tool_errors.py b/tests/unit_tests/test_mcp_tool_errors.py index 63ea31e45..bfc8a3966 100644 --- a/tests/unit_tests/test_mcp_tool_errors.py +++ b/tests/unit_tests/test_mcp_tool_errors.py @@ -612,7 +612,7 @@ def test_cloud_chain_remains_bounded_and_tolerates_cycles() -> None: assert cloud_error_trace_message(outer) is None -@pytest.mark.parametrize("transport", ["config", "public", "direct"]) +@pytest.mark.parametrize("transport", ["config", "public", "direct", "sdk"]) @pytest.mark.parametrize( ("content", "encoding", "expected"), [ @@ -669,7 +669,12 @@ def test_raw_cloud_errors_validate_utf8_before_python_agent_and_trace_relay( bearer_token=api_util.SecretString("token"), ) with pytest.raises(api_util.AirbyteCloudError) as caught: - if transport == "config": + if transport == "sdk": + sdk = SDKError("ignored SDK wording", 500, response.text, response) + assert api_util.sdk_error_message(sdk) == expected + assert cloud_error_trace_message(sdk) == expected + raise api_util._wrap_sdk_error(sdk) from sdk + elif transport == "config": api_util._make_config_api_request(**kwargs, json={}) elif transport == "public": api_util._get_public_api_json(**kwargs, params=None) @@ -698,5 +703,9 @@ def test_raw_cloud_errors_validate_utf8_before_python_agent_and_trace_relay( ) else: assert "response_text" not in error.context - assert isinstance(error.__cause__, requests.HTTPError) - assert error.__cause__.response is response + if transport == "sdk": + assert isinstance(error.__cause__, SDKError) + assert error.__cause__.raw_response is response + else: + assert isinstance(error.__cause__, requests.HTTPError) + assert error.__cause__.response is response From 49d7324e03b7d273471eb36189bd89a73691ce8d Mon Sep 17 00:00:00 2001 From: Caleb Evans Date: Mon, 12 Oct 2026 23:52:52 -0700 Subject: [PATCH 7/8] fix(mcp): format caught Cloud errors before building results --- airbyte/cloud/connectors.py | 8 +- airbyte/mcp/cloud.py | 37 +++---- tests/unit_tests/test_cloud_credentials.py | 26 ++++- tests/unit_tests/test_mcp_cloud.py | 107 ++++++++++++++++++--- 4 files changed, 139 insertions(+), 39 deletions(-) diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py index 2f931220a..e1ae7693f 100644 --- a/airbyte/cloud/connectors.py +++ b/airbyte/cloud/connectors.py @@ -101,7 +101,7 @@ if TYPE_CHECKING: - from collections.abc import Sequence + from collections.abc import Callable, Sequence from airbyte.cloud.connections import CloudConnection from airbyte.cloud.workspaces import CloudWorkspace @@ -754,6 +754,7 @@ def _context_layer_inspect( *, warnings: list[str], force_refresh: bool = False, + format_error: Callable[[exc.AirbyteCloudError], str] = str, ) -> _DirectConnectorInspectResult | None: """Fetch and cache the Context layer connector details. @@ -761,7 +762,8 @@ def _context_layer_inspect( Context layer knows. A 403 or 404 `AirbyteCloudError` from the docs read means the connector is not enabled for agent access, so a warning is appended to `warnings` and `None` is returned. Any other failure (auth, server, malformed - response, transport) is raised to the caller. + response, transport) is raised to the caller. `format_error` lets presentation + callers omit diagnostic context from the warning. """ if self._context_layer_details is not None and not force_refresh: return self._context_layer_details @@ -781,7 +783,7 @@ def _context_layer_inspect( except exc.AirbyteCloudError as error: if not agents_api_util.is_not_enabled_error(error): raise - warnings.append(f"Connector direct-access docs lookup failed: {error}") + warnings.append(f"Connector direct-access docs lookup failed: {format_error(error)}") return None parsed = _DirectConnectorInspectResult( connector_id=self.connector_id, diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 31d558579..e9bfb4ffd 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -87,7 +87,7 @@ CloudConnectorDocsResult, render_connector_docs_result, ) -from airbyte.mcp._error_handling import redact_agent_text +from airbyte.mcp._error_handling import format_user_facing_error, redact_agent_text from airbyte.mcp._scope import record_default_workspace, record_resolved_organization from airbyte.mcp._tool_utils import ( AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET, @@ -190,16 +190,16 @@ def _get_connector_check_message(check_result: CheckResult) -> str | None: ) +def _caught_error_text(error: Exception) -> str: + """Format Cloud failures without diagnostic context; retain other exception wording.""" + return format_user_facing_error(error) if isinstance(error, AirbyteCloudError) else str(error) + + def _feature_lookup_warning(error: Exception) -> str: - """Describe a failed connector feature lookup, calling out access denials.""" - if isinstance(error, AirbyteCloudApiError) and error.status_code == HTTPStatus.FORBIDDEN: - return ( - "Connector feature lookup was denied (403 Forbidden); enabled features are " - "unknown. This is a permission problem: check that the credentials have " - "access to this workspace and connector." - ) + """Describe a failed connector feature lookup with Cloud workflow guidance.""" return redact_agent_text( - f"Connector feature lookup failed; enabled features are unknown: {error}" + "Connector feature lookup failed; enabled features are unknown: " + f"{_caught_error_text(error)}" ) @@ -1402,7 +1402,7 @@ def _describe_cloud_connector( # noqa: PLR0912 # Too many branches try: integration_name = connector.integration_name except AirbyteCloudError as error: - warnings.append(f"Integration name lookup failed: {error}") + warnings.append(f"Integration name lookup failed: {_caught_error_text(error)}") integration_name = None result = CloudConnectorDetailsResult( @@ -1428,9 +1428,12 @@ def _describe_cloud_connector( # noqa: PLR0912 # Too many branches try: context_layer = connector._context_layer_inspect( # noqa: SLF001 warnings=warnings, + format_error=format_user_facing_error, ) except (AirbyteCloudError, requests.RequestException) as error: - warnings.append(f"Connector direct-access docs lookup failed: {error}") + warnings.append( + f"Connector direct-access docs lookup failed: {_caught_error_text(error)}" + ) else: if context_layer is not None: warnings.extend(str(warning) for warning in context_layer.warnings) @@ -1442,19 +1445,19 @@ def _describe_cloud_connector( # noqa: PLR0912 # Too many branches elif connector_type == ConnectorType.DESTINATION: result.config = connector.as_cloud_destination().configuration except AirbyteCloudError as error: - warnings.append(f"Connector configuration lookup failed: {error}") + warnings.append(f"Connector configuration lookup failed: {_caught_error_text(error)}") if with_replication_details: try: result.replication_details = connector_docs.build_connection_details(connector) except AirbyteCloudError as error: - warnings.append(f"Connection listing failed: {error}") + warnings.append(f"Connection listing failed: {_caught_error_text(error)}") if with_direct_access_guidance: try: docs = connector.get_direct_access_guidance() except (AirbyteLibError, requests.RequestException) as error: - warnings.append(f"Direct access docs are unavailable: {error}") + warnings.append(f"Direct access docs are unavailable: {_caught_error_text(error)}") else: result.direct_access_guidance = render_connector_docs_result(docs) @@ -1462,7 +1465,7 @@ def _describe_cloud_connector( # noqa: PLR0912 # Too many branches try: result.data_replication_docs = connector.get_data_replication_docs() except (AirbyteLibError, requests.RequestException) as error: - warnings.append(f"Data replication docs are unavailable: {error}") + warnings.append(f"Data replication docs are unavailable: {_caught_error_text(error)}") result.warnings = [redact_agent_text(warning) for warning in warnings] return result @@ -2935,9 +2938,7 @@ def get_cloud_organization_billing_status( try: info = org.get_billing_status() except (AirbyteCloudError, NotImplementedError) as error: - reason = ( - error.message if isinstance(error, AirbyteCloudError) and error.message else str(error) - ) + reason = _caught_error_text(error) return CloudOrganizationBillingStatusResult( organization_id=org.organization_id, organization_name=org.organization_name, diff --git a/tests/unit_tests/test_cloud_credentials.py b/tests/unit_tests/test_cloud_credentials.py index d949039f4..28b4d7bc4 100644 --- a/tests/unit_tests/test_cloud_credentials.py +++ b/tests/unit_tests/test_cloud_credentials.py @@ -2554,7 +2554,10 @@ def fail_for_source_2(**kwargs: object) -> ConnectorEnablement: ] assert any("enabled features are unknown" in w for w in results[1].warnings) is_forbidden = isinstance(error, AirbyteCloudApiError) and error.status_code == 403 - assert any("permission problem" in w for w in results[1].warnings) is is_forbidden + assert ( + any(api_util.FORBIDDEN_RESOURCE_GUIDANCE in w for w in results[1].warnings) + is is_forbidden + ) # A 502/503/504 or transport failure stops further enablement lookups; a 403 does not. assert calls["enablement"] == (2 if workspace_wide else 3) @@ -2593,12 +2596,17 @@ def fail(**_: object) -> ConnectorEnablement: assert result.integration_name == "GitHub" assert any("enabled features are unknown" in w for w in result.warnings) is_forbidden = isinstance(error, AirbyteCloudApiError) and error.status_code == 403 - assert any("permission problem" in w for w in result.warnings) is is_forbidden + assert ( + any(api_util.FORBIDDEN_RESOURCE_GUIDANCE in w for w in result.warnings) + is is_forbidden + ) assert calls["list"] == 0 +@pytest.mark.parametrize("status", [403, 404, 500]) def test_mcp_describe_cloud_connector_docs_failure_keeps_features( monkeypatch: pytest.MonkeyPatch, + status: int, ) -> None: """A docs lookup failure is a warning; features still come from enablement.""" workspace = _make_workspace( @@ -2610,7 +2618,14 @@ def test_mcp_describe_cloud_connector_docs_failure_keeps_features( monkeypatch.setattr( cloud_workspaces.agents_api_util, "read_cloud_skill_docs", - lambda **_: (_ for _ in ()).throw(AirbyteCloudApiError(status_code=500)), + lambda **_: (_ for _ in ()).throw( + AirbyteCloudApiError( + status_code=status, + context={ + "response_text": '{"detail":"Safe detail", "data":{"message":"PRIVATE_DATA"}}' + }, + ) + ), ) source = _seed_source(workspace, "source-1", "GitHub Issues") source._connector_definition = SimpleNamespace( # noqa: SLF001 @@ -2629,7 +2644,10 @@ def test_mcp_describe_cloud_connector_docs_failure_keeps_features( ConnectorFeature.DIRECT_ACCESS, ConnectorFeature.DIRECT_API_QUERY, ] - assert any("docs lookup failed" in w for w in result.warnings) + expected = "Connector direct-access docs lookup failed: Safe detail" + if status == 403: + expected += " " + api_util.FORBIDDEN_RESOURCE_GUIDANCE + assert result.warnings == [expected] @pytest.mark.parametrize( diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index c03950660..a9cadbc4c 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -2326,10 +2326,16 @@ def is_feature_enabled(self, feature: ConnectorFeature) -> bool: return feature in self.enabled_features def _context_layer_inspect( - self, *, warnings: list[str], **_kwargs: object + self, + *, + warnings: list[str], + format_error: Callable[[AirbyteCloudError], str] = str, + **_kwargs: object, ) -> object: if self.inspect_error is not None: - warnings.append(f"Connector inspect failed: {self.inspect_error}") + warnings.append( + f"Connector inspect failed: {format_error(self.inspect_error)}" + ) return None return self.inspect_result @@ -2888,10 +2894,25 @@ def enabled_features(self, value: frozenset[ConnectorFeature]) -> None: assert any("enabled features are unknown" in w for w in result.warnings) -@pytest.mark.parametrize("field", ["detail", "title"]) +@pytest.mark.parametrize( + ("status", "body_kind"), + [ + (403, "detail"), + (409, "ready"), + (500, "title"), + (500, "non-problem"), + (500, "malformed"), + (500, "oversized"), + ], +) +@pytest.mark.parametrize("carrier", ["sdk", "direct", "config"]) @pytest.mark.parametrize("result_path", ["billing", "feature-warning", "description"]) def test_successful_results_redact_caught_cloud_error_credentials( - monkeypatch: pytest.MonkeyPatch, field: str, result_path: str + monkeypatch: pytest.MonkeyPatch, + status: int, + body_kind: str, + carrier: str, + result_path: str, ) -> None: """Caught errors remain successful results with credentials masked in both encodings.""" import json @@ -2899,6 +2920,7 @@ def test_successful_results_redact_caught_cloud_error_credentials( from airbyte_api.errors import SDKError from fastmcp import Client from fastmcp_extensions import mcp_server + from mcp.types import TextContent from airbyte._util import api_util from airbyte.mcp._error_handling import ( @@ -2911,11 +2933,50 @@ def test_successful_results_redact_caught_cloud_error_credentials( "https://fake-user:fake-password@example.com/path?token=fake-query" ) response = requests.Response() - response.status_code = 500 - response._content = json.dumps({field: message}).encode() - sdk = SDKError("ignored", 500, response.text, response) - error = api_util._wrap_sdk_error(sdk) - error.__cause__ = sdk + payload = { + "message": "PRIVATE_TOP_LEVEL", + "data": {"message": "PRIVATE_DATA_MESSAGE"}, + "errorId": "PRIVATE_ERROR_ID", + } + if body_kind in {"detail", "title", "ready", "oversized"}: + payload["title" if body_kind == "title" else "detail"] = message + if body_kind == "ready": + payload["title"] = "actor-not-ready" + body = json.dumps(payload) + if body_kind == "malformed": + body = "PRIVATE_RAW_BODY" + elif body_kind == "oversized": + body += " " * 65536 + selected = ( + message + if body_kind in {"detail", "title", "ready"} + else (f"Airbyte Cloud returned HTTP {status}.") + ) + response.status_code = status + response._content = body.encode() + response.request = requests.Request( + "POST", "https://api.airbyte.com/test" + ).prepare() + if carrier == "sdk": + sdk = SDKError("ignored", status, response.text, response) + error = api_util._wrap_sdk_error(sdk) + error.__cause__ = sdk + elif carrier == "direct": + error = AirbyteCloudApiError( + status_code=status, message=selected, context={"response_text": body} + ) + else: + monkeypatch.setattr(api_util.requests, "request", lambda **_: response) + with pytest.raises(AirbyteCloudError) as caught: + api_util._make_config_api_request( + api_root="https://api.airbyte.com/api/public/v1", + path="/test", + json={}, + client_id=None, + client_secret=None, + bearer_token=api_util.SecretString("fake"), + ) + error = caught.value def fail(*_args: object, **_kwargs: object) -> None: raise error @@ -2966,11 +3027,29 @@ async def call_tool() -> None: result = await client.call_tool("caught_error_result") assert result.is_error is False assert result.structured_content is not None - for rendered in (str(result.content), json.dumps(result.structured_content)): - assert "Denied Bearer " in rendered - for secret in ("FAKE_SECRET", "fake-user", "fake-password", "fake-query"): + for rendered in ( + "\n".join( + block.text for block in result.content if isinstance(block, TextContent) + ), + json.dumps(result.structured_content), + ): + assert cloud_mcp.redact_agent_text(selected) in rendered + for secret in ( + "FAKE_SECRET", + "fake-user", + "fake-password", + "fake-query", + "PRIVATE_TOP_LEVEL", + "PRIVATE_DATA_MESSAGE", + "PRIVATE_ERROR_ID", + "PRIVATE_RAW_BODY", + ): assert secret not in rendered + if status == 403: + assert api_util.FORBIDDEN_RESOURCE_GUIDANCE in rendered + elif body_kind == "ready": + assert api_util.ACTOR_NOT_READY_MCP_GUIDANCE in rendered asyncio.run(call_tool()) - assert error.message == message - assert cloud_error_trace_message(error) == message + assert error.message == selected + assert cloud_error_trace_message(error) == selected From ed101574d22f9cfbd2c149c670467d7bea3a27f0 Mon Sep 17 00:00:00 2001 From: Caleb Evans Date: Tue, 13 Oct 2026 08:14:24 -0700 Subject: [PATCH 8/8] fix(mcp): redact bearer credentials case-insensitively --- airbyte/mcp/_error_handling.py | 2 +- tests/unit_tests/test_mcp_cloud.py | 3 ++- tests/unit_tests/test_mcp_tool_errors.py | 7 ++++++- 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/airbyte/mcp/_error_handling.py b/airbyte/mcp/_error_handling.py index 6c09d7283..0360fece0 100644 --- a/airbyte/mcp/_error_handling.py +++ b/airbyte/mcp/_error_handling.py @@ -272,7 +272,7 @@ def format_validation_error(error: ValidationError) -> str: _INPUT_VALUE = re.compile(r"input_value=(?:'[^']*'|\"[^\"]*\"|[^\s,\])]+)") -_BEARER_TOKEN = re.compile(r"Bearer\s+[^\s'\")]+") +_BEARER_TOKEN = re.compile(r"Bearer\s+[^\s'\")]+", re.IGNORECASE) _URL_CREDENTIALS = re.compile(r"([a-zA-Z][a-zA-Z0-9+.-]*://)[^\s/?#'\")]+@") _URL_QUERY = re.compile(r"([a-zA-Z][a-zA-Z0-9+.-]*://[^\s?#'\")]*)\?[^\s'\")]*") diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index a9cadbc4c..62e425a22 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -2929,7 +2929,7 @@ def test_successful_results_redact_caught_cloud_error_credentials( ) message = ( - "Denied Bearer FAKE_SECRET at " + "Denied authorization: bearer FAKE_SECRET and bEaReR MIXED_SECRET at " "https://fake-user:fake-password@example.com/path?token=fake-query" ) response = requests.Response() @@ -3036,6 +3036,7 @@ async def call_tool() -> None: assert cloud_mcp.redact_agent_text(selected) in rendered for secret in ( "FAKE_SECRET", + "MIXED_SECRET", "fake-user", "fake-password", "fake-query", diff --git a/tests/unit_tests/test_mcp_tool_errors.py b/tests/unit_tests/test_mcp_tool_errors.py index bfc8a3966..5a1aa2799 100644 --- a/tests/unit_tests/test_mcp_tool_errors.py +++ b/tests/unit_tests/test_mcp_tool_errors.py @@ -519,8 +519,10 @@ def test_cloud_chain_uses_exact_message_for_agent_and_span( @pytest.mark.parametrize("field", ["detail", "title"]) +@pytest.mark.parametrize("scheme", ["Bearer", "bearer", "BEARER", "bEaReR"]) def test_real_middleware_redacts_selected_cloud_message_after_nested_rewrap( field: str, + scheme: str, ) -> None: server = _server_with_agent_error_text() @@ -529,7 +531,10 @@ def broken() -> str: response = requests.Response() response.status_code = 403 sdk = SDKError( - "ignored", 403, json.dumps({field: "Denied Bearer FAKE_SECRET"}), response + "ignored", + 403, + json.dumps({field: f"Denied {scheme} FAKE_SECRET"}), + response, ) wrapped = api_util._wrap_sdk_error(sdk) wrapped.__cause__ = sdk