From e67fd981b50a686da0722dfb318d260564e678b4 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sun, 11 Oct 2026 22:57:43 +0000 Subject: [PATCH 1/3] feat(mcp): give non-Cloud tool errors fixed Datadog span messages from a manifest --- AGENTS.md | 1 + CONTRIBUTING.md | 4 + airbyte/mcp/_datadog.py | 10 +- airbyte/mcp/_error_handling.py | 26 +++ airbyte/mcp/tool_errors.yaml | 99 ++++++++++ tests/unit_tests/test_mcp_datadog.py | 273 ++++++++++++++++++++++++++- 6 files changed, 402 insertions(+), 11 deletions(-) create mode 100644 airbyte/mcp/tool_errors.yaml diff --git a/AGENTS.md b/AGENTS.md index 8677111d9..26caf416d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,3 +24,4 @@ Key conventions: 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. +Non-Cloud MCP trace messages come from `airbyte/mcp/tool_errors.yaml`; that manifest is trace-only. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 0373d1bec..c3e66cc15 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -34,6 +34,10 @@ server internals. To support a new problem type, add a `message` and a `guidance 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. +Non-Cloud MCP tool failures use fixed Datadog trace messages from +`airbyte/mcp/tool_errors.yaml`, keyed by exception class. This catalog affects traces only; +it does not change the error text returned to agents. + ## 🚀 Releasing This project uses [`semantic-pr-release-drafter`](https://github.com/airbytehq/semantic-pr-release-drafter) for automated release management. To release, simply click "`Edit`" on the latest release draft from the [releases page](https://github.com/airbytehq/PyAirbyte/releases), and then click "`Publish release`". This publish operation will trigger all necessary downstream publish operations. diff --git a/airbyte/mcp/_datadog.py b/airbyte/mcp/_datadog.py index 7dcee1f4c..15332e43a 100644 --- a/airbyte/mcp/_datadog.py +++ b/airbyte/mcp/_datadog.py @@ -37,8 +37,8 @@ MCP_TOOL_USER_FACING_ERRORS, classify_mcp_tool_error, cloud_error_route, - cloud_error_trace_message, mcp_tool_error_reason, + tool_error_trace_message, ) from airbyte.mcp._otel import _arg_key, _env, _flag from airbyte.mcp._scope import current_call_scope, scope_from_request @@ -209,10 +209,10 @@ def _exception_attributes(error: BaseException) -> dict[str, str]: if route := cloud_error_route(cause): attrs[_UPSTREAM_METHOD], attrs[_UPSTREAM_ROUTE] = route try: - if message := cloud_error_trace_message(cause): + if message := tool_error_trace_message(cause): attrs["error.message"] = message except Exception: - logger.debug("Datadog Cloud error message unavailable") + logger.debug("Datadog fixed error message unavailable") return attrs @@ -258,9 +258,9 @@ def _tool_error_message(span: Span) -> str: def _raised_tool_error_message(error: BaseException, span: Span) -> str: - """Return a fixed Cloud message when available, falling back to the span summary.""" + """Return a fixed message when available, falling back to the span summary.""" try: - message = cloud_error_trace_message(error) + message = tool_error_trace_message(error) except Exception: message = None return message if message is not None else _tool_error_message(span) diff --git a/airbyte/mcp/_error_handling.py b/airbyte/mcp/_error_handling.py index bcbbac45a..953259e7e 100644 --- a/airbyte/mcp/_error_handling.py +++ b/airbyte/mcp/_error_handling.py @@ -3,14 +3,17 @@ from __future__ import annotations +import importlib.resources import json import logging import re from collections.abc import Mapping +from functools import cache from typing import TYPE_CHECKING, Any from urllib.parse import urlsplit import requests +import yaml from fastmcp.exceptions import ToolError, ValidationError from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext from fastmcp_extensions.otel._arg_digests import classify_tool # noqa: PLC2701 @@ -118,6 +121,29 @@ def cloud_error_trace_message(error: BaseException) -> str | None: return None +@cache +def _tool_error_messages() -> dict[str, str]: + """Return the fixed trace messages keyed by exception class.""" + path = importlib.resources.files("airbyte.mcp").joinpath("tool_errors.yaml") + return { + exception: entry["message"] + for exception, entry in yaml.safe_load(path.read_text(encoding="utf-8")).items() + } + + +def tool_error_trace_message(error: BaseException) -> str | None: + """Return the fixed trace message for an exception, if any.""" + if (message := cloud_error_trace_message(error)) is not None: + return message + messages = _tool_error_messages() + for cause in _chain(error): + for error_class in type(cause).__mro__: + key = f"{error_class.__module__}.{error_class.__qualname__}" + if (message := messages.get(key)) is not None: + return message + return None + + def cloud_error_route(error: BaseException) -> tuple[str, str] | None: """Return a bounded HTTP method and sanitized path for an upstream SDK error.""" for cause in _chain(error): diff --git a/airbyte/mcp/tool_errors.yaml b/airbyte/mcp/tool_errors.yaml new file mode 100644 index 000000000..8572b210d --- /dev/null +++ b/airbyte/mcp/tool_errors.yaml @@ -0,0 +1,99 @@ +# Fixed span messages for MCP tool errors that are not Airbyte Cloud API errors, +# used by `tool_error_trace_message` in airbyte/mcp/_error_handling.py. +# Cloud API errors take their span message from airbyte/_util/cloud_errors.yaml +# first; this table covers everything else (policy refusals, bad input, +# missing context, lookups that found nothing, unknown tools). +# +# Maintaining this file: +# - Keys are fully qualified exception class names (`module.ClassName`). For +# each exception in the cause chain, outermost first, the first class in its +# MRO that has an entry wins, so a base-class entry covers its subclasses. +# Keep keys sorted. +# - Each entry has exactly `message`: one sentence saying what went wrong. +# - Write fixed text only. Never include IDs, names, argument values, URLs or +# exception text; these messages are exported to Datadog as `error.message`. +# - This text goes to traces only. What the agent sees is unchanged and still +# comes from the exception's message and guidance. +# - Exceptions with no entry here (e.g. a bare `ValueError`) keep the short +# `tool resulted in an error (...)` summary built from classification tags. +# - When adding an exception class to airbyte/exceptions.py, add an entry or +# confirm that a base-class entry fits; +# `test_every_airbyte_exception_has_a_specific_tool_error_message` checks this. +airbyte.exceptions.AirbyteAgentsUnavailableError: + message: The Airbyte Agents API isn't available on this deployment. +airbyte.exceptions.AirbyteCloudConnectionError: + message: An operation on the Airbyte Cloud connection failed. +airbyte.exceptions.AirbyteCloudError: + message: An Airbyte Cloud request failed or returned an incomplete response. +airbyte.exceptions.AirbyteConnectionSyncActiveError: + message: A sync is running on this connection, so the change was rejected. +airbyte.exceptions.AirbyteConnectionSyncError: + message: The sync job failed. +airbyte.exceptions.AirbyteConnectionSyncTimeoutError: + message: Timed out waiting for the sync job to finish. +airbyte.exceptions.AirbyteConnectorCheckFailedError: + message: The connector's connection check failed. +airbyte.exceptions.AirbyteConnectorError: + message: The connector failed. +airbyte.exceptions.AirbyteConnectorInUseError: + message: The source or destination is still used by connections, so it can't be deleted. +airbyte.exceptions.AirbyteConnectorNotReadyError: + message: The source or destination is a draft that still needs setup in Airbyte Cloud. +airbyte.exceptions.AirbyteConnectorRegistryError: + message: The connector wasn't found in the connector registry or can't be installed from it. +airbyte.exceptions.AirbyteDeferredSetupError: + message: Airbyte Cloud created the connector without acknowledging draft mode. +airbyte.exceptions.AirbyteDuplicateResourcesError: + message: A resource with this name already exists. +airbyte.exceptions.AirbyteExternalAccessDisabledError: + message: External access is disabled for this MCP server or request. +airbyte.exceptions.AirbyteExternalAccessNotEnabledError: + message: The connector isn't enabled for external access. +airbyte.exceptions.AirbyteLibError: + message: PyAirbyte raised an error. +airbyte.exceptions.AirbyteLibInputError: + message: An argument value was invalid. +airbyte.exceptions.AirbyteLibInternalError: + message: PyAirbyte hit an internal error. +airbyte.exceptions.AirbyteLibNameNormalizationError: + message: A table or column name couldn't be normalized. +airbyte.exceptions.AirbyteLibNoStreamsSelectedError: + message: No streams were selected. +airbyte.exceptions.AirbyteLibSecretNotFoundError: + message: A required secret was not found. +airbyte.exceptions.AirbyteMCPError: + message: The MCP server refused the request. +airbyte.exceptions.AirbyteMissingResourceError: + message: The requested Airbyte resource was not found. +airbyte.exceptions.AirbyteMissingWorkspaceContextError: + message: No workspace was given, and no default workspace was available. +airbyte.exceptions.AirbyteMultipleResourcesError: + message: More than one resource matched, so the lookup was ambiguous. +airbyte.exceptions.AirbyteNoCloudCredentialsError: + message: No Airbyte Cloud credentials were provided. +airbyte.exceptions.AirbytePipelineChangesDisabledError: + message: Pipeline changes are disabled for this MCP server or request. +airbyte.exceptions.AirbyteSafeModeError: + message: Safe mode blocked a destructive operation on a resource this session didn't create. +airbyte.exceptions.AirbyteStateNotFoundError: + message: The requested stream state was not found. +airbyte.exceptions.AirbyteStreamNotFoundError: + message: The requested stream was not found. +airbyte.exceptions.AirbyteSubprocessError: + message: A connector subprocess failed. +airbyte.exceptions.AirbyteTrustedExecutionRequiredError: + message: This operation needs trusted execution, which is disabled on this server. +airbyte.exceptions.AirbyteWorkspaceMismatchError: + message: The resource belongs to a different workspace than the one passed. +airbyte.exceptions.AirbyteWorkspaceNotEmptyError: + message: The workspace still has connections, so it can't be deleted. +asyncio.exceptions.CancelledError: + message: The tool call was cancelled. +builtins.NotImplementedError: + message: This operation isn't supported for this resource type yet. +fastmcp.exceptions.DisabledError: + message: The requested tool is disabled on this server. +fastmcp.exceptions.NotFoundError: + message: The requested tool doesn't exist on this server. +fastmcp.exceptions.ValidationError: + message: The tool arguments failed validation. diff --git a/tests/unit_tests/test_mcp_datadog.py b/tests/unit_tests/test_mcp_datadog.py index 3494a978c..672943029 100644 --- a/tests/unit_tests/test_mcp_datadog.py +++ b/tests/unit_tests/test_mcp_datadog.py @@ -2,8 +2,12 @@ """Native Datadog payload policy and real HTTP/LLM span contracts.""" import asyncio +import importlib +import importlib.resources +import inspect import json import os +import re import subprocess import sys from pathlib import Path @@ -12,18 +16,24 @@ import pytest import requests +import yaml from airbyte_api.errors import SDKError from fastmcp.exceptions import NotFoundError, ToolError, ValidationError from fastmcp_extensions.tool_filters import ToolUnavailableError from airbyte._util import api_util, meta +import airbyte.exceptions from airbyte.exceptions import ( AirbyteCloudApiError, AirbyteConnectionSyncError, AirbyteConnectionSyncTimeoutError, + AirbyteLibError, + AirbyteLibInputError, + AirbyteMissingResourceError, + AirbyteMissingWorkspaceContextError, AirbyteSafeModeError, ) -from airbyte.mcp import _datadog, _telemetry +from airbyte.mcp import _datadog, _error_handling, _telemetry from airbyte.version import get_version @@ -320,7 +330,7 @@ def test_cloud_error_trace_message_for_config_api_errors( ) assert isinstance(exc_info.value.__cause__, requests.HTTPError) - assert _datadog.cloud_error_trace_message(exc_info.value) == expected_message + assert _error_handling.cloud_error_trace_message(exc_info.value) == expected_message def test_cloud_error_trace_message_ignores_missing_resource_without_http_error() -> ( @@ -328,7 +338,123 @@ def test_cloud_error_trace_message_ignores_missing_resource_without_http_error() ): error = api_util.AirbyteMissingResourceError(message="resource SENTINEL") - assert _datadog.cloud_error_trace_message(error) is None + assert _error_handling.cloud_error_trace_message(error) is None + + +def test_tool_error_manifest_shape_and_messages_are_safe() -> None: + path = importlib.resources.files("airbyte.mcp").joinpath("tool_errors.yaml") + entries = yaml.safe_load(path.read_text(encoding="utf-8")) + + assert len(entries) == 39 + assert list(entries) == sorted(entries) + for key, entry in entries.items(): + assert set(entry) == {"message"} + message = entry["message"] + assert isinstance(message, str) and message.strip() + module_name, _, class_name = key.rpartition(".") + error_class = getattr(importlib.import_module(module_name), class_name) + assert inspect.isclass(error_class) + assert issubclass(error_class, BaseException) + assert "{" not in message + assert re.search(r"\d+", message) is None + assert "http" not in message.lower() + assert '"' not in message + + +def test_every_airbyte_exception_has_a_specific_tool_error_message() -> None: + messages = _error_handling._tool_error_messages() + + for name, error_class in inspect.getmembers(airbyte.exceptions, inspect.isclass): + if not issubclass(error_class, AirbyteLibError) or error_class in { + AirbyteLibError, + AirbyteCloudApiError, + }: + continue + mro_keys = [ + f"{parent.__module__}.{parent.__qualname__}" + for parent in error_class.__mro__ + if parent is not AirbyteLibError + ] + assert any(key in messages for key in mro_keys), name + + +@pytest.mark.parametrize( + ("error", "expected_class"), + [ + pytest.param( + AirbyteSafeModeError(message="Blocked for 'abc-secret-id-123'."), + AirbyteSafeModeError, + id="safe-mode", + ), + pytest.param( + AirbyteMissingWorkspaceContextError(), + AirbyteMissingWorkspaceContextError, + id="missing-workspace", + ), + pytest.param( + AirbyteLibInputError(message="bad value xyz-123"), + AirbyteLibInputError, + id="library-input", + ), + pytest.param( + ToolError("outer wrapper"), + AirbyteSafeModeError, + id="wrapped-safe-mode", + ), + pytest.param( + NotFoundError("Unknown tool: foo"), + NotFoundError, + id="unknown-tool", + ), + pytest.param( + ValidationError("invalid argument"), + ValidationError, + id="fastmcp-validation", + ), + ], +) +def test_tool_error_trace_message_uses_exception_manifest( + error: BaseException, expected_class: type[BaseException] +) -> None: + if isinstance(error, ToolError): + error.__cause__ = AirbyteSafeModeError(message="wrapped secret abc-123") + expected = _error_handling._tool_error_messages()[ + f"{expected_class.__module__}.{expected_class.__qualname__}" + ] + message = _error_handling.tool_error_trace_message(error) + + assert message == expected + attributes = _datadog._exception_attributes(error) + assert attributes["error.message"] == expected + for cause in _error_handling._chain(error): + assert str(cause) not in message + assert str(cause) not in json.dumps(attributes) + + +def test_tool_error_trace_message_keeps_cloud_precedence() -> None: + error = AirbyteMissingResourceError( + message=api_util.FORBIDDEN_RESOURCE_MESSAGE, + context={"problem_type": "forbidden"}, + ) + error.__cause__ = requests.HTTPError("private response body") + + assert ( + _error_handling.tool_error_trace_message(error) + == api_util.FORBIDDEN_RESOURCE_MESSAGE + ) + + +def test_tool_error_trace_message_ignores_unlisted_exceptions() -> None: + from pydantic import BaseModel, ValidationError as PydanticValidationError + + class InputModel(BaseModel): + value: int + + with pytest.raises(PydanticValidationError) as exc_info: + InputModel.model_validate({"value": "private-pydantic-value"}) + + assert _error_handling.tool_error_trace_message(ValueError("private value")) is None + assert _error_handling.tool_error_trace_message(exc_info.value) is None @pytest.mark.parametrize( @@ -496,7 +622,7 @@ def test_exception_attributes_classify_without_error_text( "error.type": expected["airbyte.mcp.error_type"], **expected, } - if message := _datadog.cloud_error_trace_message(error): + if message := _error_handling.tool_error_trace_message(error): expected_attrs["error.message"] = message assert attrs == expected_attrs @@ -536,13 +662,18 @@ def test_fastmcp_rejections_are_classified_like_the_otel_backend() -> None: "invalid_arguments", ) - attrs = _datadog._exception_attributes(NotFoundError("SENTINEL")) + not_found = NotFoundError("SENTINEL") + assert _datadog.unwrap_tool_error(not_found) is not_found + attrs = _datadog._exception_attributes(not_found) assert attrs == { "airbyte.mcp.outcome": "unknown_tool", "airbyte.mcp.error_type": "NotFoundError", "error.type": "NotFoundError", "airbyte.mcp.error.category": "unknown_tool", "airbyte.mcp.error.fault": "caller", + "error.message": _error_handling._tool_error_messages()[ + "fastmcp.exceptions.NotFoundError" + ], } @@ -937,7 +1068,16 @@ def _native_http_contract(): InMemorySpanExporter, ) from airbyte.mcp import _otel, _scope, _telemetry - from fastmcp_extensions import ToolCallTelemetryMiddleware + from airbyte.mcp import _tool_utils + from airbyte.mcp._error_handling import ( + AgentErrorTextMiddleware, + MCP_TOOL_USER_FACING_ERRORS, + format_user_facing_error, + ) + from fastmcp_extensions import ( + ToolCallTelemetryMiddleware, + UserFacingErrorMiddleware, + ) from unittest.mock import AsyncMock # Assert every socket is loopback; writers are captured and never started. @@ -982,7 +1122,23 @@ def log_message(self, *_): upstream = HTTPServer(("127.0.0.1", 0), Handler) threading.Thread(target=upstream.serve_forever, daemon=True).start() app = FastMCP("native-mcp-contract") + app.add_middleware(AgentErrorTextMiddleware()) + app.add_middleware( + UserFacingErrorMiddleware( + MCP_TOOL_USER_FACING_ERRORS, + formatter=format_user_facing_error, + ) + ) records = [] + exception_texts = {} + original_exception_attributes = _datadog._exception_attributes + + def capture_exception_text(error): + cause = _datadog.unwrap_tool_error(error) + if isinstance(cause, (AirbyteSafeModeError, ValidationError, NotFoundError)): + exception_texts.setdefault(type(cause).__name__, str(cause)) + return original_exception_attributes(error) + app.add_middleware(_scope.CallScopeMiddleware()) telemetry = ToolCallTelemetryMiddleware( extra_properties=_scope.call_scope_properties @@ -1037,6 +1193,15 @@ def execute( def run_sql_query() -> str: return "private synthetic result" + @app.tool() + def safe_mode_blocker(guid: str) -> str: + _tool_utils.check_guid_created_in_session(guid) + return "unreachable" + + @app.tool() + def strict_integer(value: int) -> str: + return str(value) + @app.tool() def fail_tool() -> str: response = SimpleNamespace(status_code=500) @@ -1125,6 +1290,20 @@ async def run(): ], ("tools/call", {"name": "run_sql_query", "arguments": {}}), ("tools/call", {"name": "unknown_tool", "arguments": {}}), + ( + "tools/call", + { + "name": "safe_mode_blocker", + "arguments": {"guid": "abc-secret-id-123"}, + }, + ), + ( + "tools/call", + { + "name": "strict_integer", + "arguments": {"value": "private-invalid-type"}, + }, + ), ] for index, (method, params) in enumerate(sequence): response = await client.post( @@ -1187,6 +1366,8 @@ async def run(): assert cloud_span.get_tag("airbyte.mcp.upstream.route") == ( "/v1/connections/{id}" ) + assert "private response text" not in json.dumps(cloud_span.get_tags()) + assert "private response text" not in json.dumps(cloud_metadata) assert cloud_uuid not in json.dumps(cloud_span.get_tags()) assert "x=1" not in json.dumps(cloud_span.get_tags()) assert cloud_uuid not in json.dumps(cloud_metadata) @@ -1201,6 +1382,80 @@ async def run(): assert returned_error_span.get_tag("error.message").startswith( "tool resulted in an error (" ) + value_error_span = next( + span for span in spans if span.get_tag("error.type") == "ValueError" + ) + unknown_tool_span = next( + span + for span in spans + if span.get_tag("error.type") == "NotFoundError" + ) + assert value_error_span.get_tag("error.message") == ( + "tool resulted in an error (unclassified)" + ) + assert ( + unknown_tool_span.get_tag("error.message") + == ( + _error_handling._tool_error_messages()[ + "fastmcp.exceptions.NotFoundError" + ] + ) + ) + safe_span = next( + span + for span in spans + if span.get_tag("error.type") == "AirbyteSafeModeError" + ) + validation_span = next( + span + for span in spans + if span.get_tag("error.type") == "ValidationError" + ) + assert ( + safe_span.get_tag("error.message") + == ( + _error_handling._tool_error_messages()[ + "airbyte.exceptions.AirbyteSafeModeError" + ] + ) + ) + assert ( + validation_span.get_tag("error.message") + == ( + _error_handling._tool_error_messages()[ + "fastmcp.exceptions.ValidationError" + ] + ) + ) + for span, exception_name in ( + (safe_span, "AirbyteSafeModeError"), + (validation_span, "ValidationError"), + (unknown_tool_span, "NotFoundError"), + ): + exception_text = exception_texts[exception_name] + assert all( + exception_text not in str(value) + for value in span.get_tags().values() + ) + for span, private_text in ( + (safe_span, "abc-secret-id-123"), + (validation_span, "private-invalid-type"), + ): + assert private_text not in json.dumps(span.get_tags()) + safe_mode_text = responses[8][2]["result"]["content"][0]["text"] + assert safe_mode_text == ( + "Cannot perform destructive operation on '': " + "Object was not created in this session. " + "AIRBYTE_CLOUD_MCP_SAFE_MODE is set to '1'." + ) + assert "abc-secret-id-123" not in safe_mode_text + assert str(ValueError(private_payload)) not in json.dumps( + value_error_span.get_tags() + ) + assert ( + "private-invalid-type" + not in responses[9][2]["result"]["content"][0]["text"] + ) # Telemetry failures must not abort dispatch or replace its result. for target in ("tool", "annotate"): before = len(calls) @@ -1537,6 +1792,12 @@ async def dispatch(_ctx): "resolve_call_workspace_organization_id", new=AsyncMock(return_value="44444444-4444-4444-4444-444444444444"), ), + patch.object(_tool_utils, "AIRBYTE_CLOUD_MCP_SAFE_MODE", True), + patch.object( + _datadog, + "_exception_attributes", + side_effect=capture_exception_text, + ), ): asyncio.run(run()) assert client_properties == original_properties From 6e6892f5d8366fe9d3b31921062b018438f078ca Mon Sep 17 00:00:00 2001 From: "caleb.evans" Date: Sun, 11 Oct 2026 22:59:23 +0000 Subject: [PATCH 2/3] test(mcp): drop fixed entry count from tool error manifest test --- tests/unit_tests/test_mcp_datadog.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/unit_tests/test_mcp_datadog.py b/tests/unit_tests/test_mcp_datadog.py index 672943029..660e7d8e3 100644 --- a/tests/unit_tests/test_mcp_datadog.py +++ b/tests/unit_tests/test_mcp_datadog.py @@ -345,7 +345,6 @@ def test_tool_error_manifest_shape_and_messages_are_safe() -> None: path = importlib.resources.files("airbyte.mcp").joinpath("tool_errors.yaml") entries = yaml.safe_load(path.read_text(encoding="utf-8")) - assert len(entries) == 39 assert list(entries) == sorted(entries) for key, entry in entries.items(): assert set(entry) == {"message"} From e36f997f7e8aa5e02a9e40ff6139e3bc2e768e67 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 12 Oct 2026 18:50:12 +0000 Subject: [PATCH 3/3] refactor(mcp): derive non-Cloud trace messages from exception docstrings --- AGENTS.md | 4 +- CONTRIBUTING.md | 6 +- airbyte/exceptions.py | 4 +- airbyte/mcp/_error_handling.py | 36 +++---- airbyte/mcp/tool_errors.yaml | 99 ------------------- tests/unit_tests/test_mcp_datadog.py | 139 +++++++++++++++------------ 6 files changed, 101 insertions(+), 187 deletions(-) delete mode 100644 airbyte/mcp/tool_errors.yaml diff --git a/AGENTS.md b/AGENTS.md index 26caf416d..e48736ca9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,4 +24,6 @@ Key conventions: 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. -Non-Cloud MCP trace messages come from `airbyte/mcp/tool_errors.yaml`; that manifest is trace-only. +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 c3e66cc15..39619bd8b 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -34,9 +34,9 @@ server internals. To support a new problem type, add a `message` and a `guidance 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. -Non-Cloud MCP tool failures use fixed Datadog trace messages from -`airbyte/mcp/tool_errors.yaml`, keyed by exception class. This catalog affects traces only; -it does not change the error text returned to agents. +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. ## 🚀 Releasing diff --git a/airbyte/exceptions.py b/airbyte/exceptions.py index c45575f73..e77abb429 100644 --- a/airbyte/exceptions.py +++ b/airbyte/exceptions.py @@ -707,7 +707,7 @@ class AirbyteConnectorInUseError(AirbyteCloudError): @dataclass class AirbyteConnectionSyncTimeoutError(AirbyteConnectionSyncError): - """An timeout occurred while waiting for the remote Airbyte job to complete.""" + """A timeout occurred while waiting for the remote Airbyte job to complete.""" timeout: int | None = None """The timeout in seconds that was reached.""" @@ -718,7 +718,7 @@ class AirbyteConnectionSyncTimeoutError(AirbyteConnectionSyncError): @dataclass class AirbyteMissingResourceError(AirbyteCloudError): - """Remote Airbyte resources does not exist.""" + """The requested Airbyte resource does not exist.""" resource_type: str | None = None resource_name_or_id: str | None = None diff --git a/airbyte/mcp/_error_handling.py b/airbyte/mcp/_error_handling.py index 953259e7e..cb2acedc4 100644 --- a/airbyte/mcp/_error_handling.py +++ b/airbyte/mcp/_error_handling.py @@ -3,18 +3,16 @@ from __future__ import annotations -import importlib.resources +import asyncio import json import logging import re from collections.abc import Mapping -from functools import cache from typing import TYPE_CHECKING, Any from urllib.parse import urlsplit import requests -import yaml -from fastmcp.exceptions import ToolError, ValidationError +from fastmcp.exceptions import DisabledError, NotFoundError, ToolError, ValidationError from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext from fastmcp_extensions.otel._arg_digests import classify_tool # noqa: PLC2701 from fastmcp_extensions.otel._extras import _chain, declared_parameters # noqa: PLC2701 @@ -78,6 +76,14 @@ ) _MAX_UPSTREAM_ROUTE_LENGTH = 200 +_NON_AIRBYTE_TOOL_ERROR_MESSAGES: dict[type[BaseException], str] = { + asyncio.CancelledError: "The tool call was cancelled.", + NotImplementedError: "This operation isn't supported for this resource type yet.", + DisabledError: "The requested tool is disabled on this server.", + NotFoundError: "The requested tool doesn't exist on this server.", + ValidationError: "The tool arguments failed validation.", +} + MCP_TOOL_USER_FACING_ERRORS: tuple[type[AirbyteLibError], ...] = ( AirbyteLibInputError, @@ -121,26 +127,20 @@ def cloud_error_trace_message(error: BaseException) -> str | None: return None -@cache -def _tool_error_messages() -> dict[str, str]: - """Return the fixed trace messages keyed by exception class.""" - path = importlib.resources.files("airbyte.mcp").joinpath("tool_errors.yaml") - return { - exception: entry["message"] - for exception, entry in yaml.safe_load(path.read_text(encoding="utf-8")).items() - } - - def tool_error_trace_message(error: BaseException) -> str | None: """Return the fixed trace message for an exception, if any.""" if (message := cloud_error_trace_message(error)) is not None: return message - messages = _tool_error_messages() for cause in _chain(error): for error_class in type(cause).__mro__: - key = f"{error_class.__module__}.{error_class.__qualname__}" - if (message := messages.get(key)) is not None: - return message + if ( + error_class is not AirbyteLibError + and issubclass(error_class, AirbyteLibError) + and error_class.__doc__ + ): + return error_class.__doc__.split("\n")[0].strip() + if error_class in _NON_AIRBYTE_TOOL_ERROR_MESSAGES: + return _NON_AIRBYTE_TOOL_ERROR_MESSAGES[error_class] return None diff --git a/airbyte/mcp/tool_errors.yaml b/airbyte/mcp/tool_errors.yaml deleted file mode 100644 index 8572b210d..000000000 --- a/airbyte/mcp/tool_errors.yaml +++ /dev/null @@ -1,99 +0,0 @@ -# Fixed span messages for MCP tool errors that are not Airbyte Cloud API errors, -# used by `tool_error_trace_message` in airbyte/mcp/_error_handling.py. -# Cloud API errors take their span message from airbyte/_util/cloud_errors.yaml -# first; this table covers everything else (policy refusals, bad input, -# missing context, lookups that found nothing, unknown tools). -# -# Maintaining this file: -# - Keys are fully qualified exception class names (`module.ClassName`). For -# each exception in the cause chain, outermost first, the first class in its -# MRO that has an entry wins, so a base-class entry covers its subclasses. -# Keep keys sorted. -# - Each entry has exactly `message`: one sentence saying what went wrong. -# - Write fixed text only. Never include IDs, names, argument values, URLs or -# exception text; these messages are exported to Datadog as `error.message`. -# - This text goes to traces only. What the agent sees is unchanged and still -# comes from the exception's message and guidance. -# - Exceptions with no entry here (e.g. a bare `ValueError`) keep the short -# `tool resulted in an error (...)` summary built from classification tags. -# - When adding an exception class to airbyte/exceptions.py, add an entry or -# confirm that a base-class entry fits; -# `test_every_airbyte_exception_has_a_specific_tool_error_message` checks this. -airbyte.exceptions.AirbyteAgentsUnavailableError: - message: The Airbyte Agents API isn't available on this deployment. -airbyte.exceptions.AirbyteCloudConnectionError: - message: An operation on the Airbyte Cloud connection failed. -airbyte.exceptions.AirbyteCloudError: - message: An Airbyte Cloud request failed or returned an incomplete response. -airbyte.exceptions.AirbyteConnectionSyncActiveError: - message: A sync is running on this connection, so the change was rejected. -airbyte.exceptions.AirbyteConnectionSyncError: - message: The sync job failed. -airbyte.exceptions.AirbyteConnectionSyncTimeoutError: - message: Timed out waiting for the sync job to finish. -airbyte.exceptions.AirbyteConnectorCheckFailedError: - message: The connector's connection check failed. -airbyte.exceptions.AirbyteConnectorError: - message: The connector failed. -airbyte.exceptions.AirbyteConnectorInUseError: - message: The source or destination is still used by connections, so it can't be deleted. -airbyte.exceptions.AirbyteConnectorNotReadyError: - message: The source or destination is a draft that still needs setup in Airbyte Cloud. -airbyte.exceptions.AirbyteConnectorRegistryError: - message: The connector wasn't found in the connector registry or can't be installed from it. -airbyte.exceptions.AirbyteDeferredSetupError: - message: Airbyte Cloud created the connector without acknowledging draft mode. -airbyte.exceptions.AirbyteDuplicateResourcesError: - message: A resource with this name already exists. -airbyte.exceptions.AirbyteExternalAccessDisabledError: - message: External access is disabled for this MCP server or request. -airbyte.exceptions.AirbyteExternalAccessNotEnabledError: - message: The connector isn't enabled for external access. -airbyte.exceptions.AirbyteLibError: - message: PyAirbyte raised an error. -airbyte.exceptions.AirbyteLibInputError: - message: An argument value was invalid. -airbyte.exceptions.AirbyteLibInternalError: - message: PyAirbyte hit an internal error. -airbyte.exceptions.AirbyteLibNameNormalizationError: - message: A table or column name couldn't be normalized. -airbyte.exceptions.AirbyteLibNoStreamsSelectedError: - message: No streams were selected. -airbyte.exceptions.AirbyteLibSecretNotFoundError: - message: A required secret was not found. -airbyte.exceptions.AirbyteMCPError: - message: The MCP server refused the request. -airbyte.exceptions.AirbyteMissingResourceError: - message: The requested Airbyte resource was not found. -airbyte.exceptions.AirbyteMissingWorkspaceContextError: - message: No workspace was given, and no default workspace was available. -airbyte.exceptions.AirbyteMultipleResourcesError: - message: More than one resource matched, so the lookup was ambiguous. -airbyte.exceptions.AirbyteNoCloudCredentialsError: - message: No Airbyte Cloud credentials were provided. -airbyte.exceptions.AirbytePipelineChangesDisabledError: - message: Pipeline changes are disabled for this MCP server or request. -airbyte.exceptions.AirbyteSafeModeError: - message: Safe mode blocked a destructive operation on a resource this session didn't create. -airbyte.exceptions.AirbyteStateNotFoundError: - message: The requested stream state was not found. -airbyte.exceptions.AirbyteStreamNotFoundError: - message: The requested stream was not found. -airbyte.exceptions.AirbyteSubprocessError: - message: A connector subprocess failed. -airbyte.exceptions.AirbyteTrustedExecutionRequiredError: - message: This operation needs trusted execution, which is disabled on this server. -airbyte.exceptions.AirbyteWorkspaceMismatchError: - message: The resource belongs to a different workspace than the one passed. -airbyte.exceptions.AirbyteWorkspaceNotEmptyError: - message: The workspace still has connections, so it can't be deleted. -asyncio.exceptions.CancelledError: - message: The tool call was cancelled. -builtins.NotImplementedError: - message: This operation isn't supported for this resource type yet. -fastmcp.exceptions.DisabledError: - message: The requested tool is disabled on this server. -fastmcp.exceptions.NotFoundError: - message: The requested tool doesn't exist on this server. -fastmcp.exceptions.ValidationError: - message: The tool arguments failed validation. diff --git a/tests/unit_tests/test_mcp_datadog.py b/tests/unit_tests/test_mcp_datadog.py index 660e7d8e3..159036347 100644 --- a/tests/unit_tests/test_mcp_datadog.py +++ b/tests/unit_tests/test_mcp_datadog.py @@ -2,12 +2,9 @@ """Native Datadog payload policy and real HTTP/LLM span contracts.""" import asyncio -import importlib -import importlib.resources import inspect import json import os -import re import subprocess import sys from pathlib import Path @@ -16,9 +13,8 @@ import pytest import requests -import yaml from airbyte_api.errors import SDKError -from fastmcp.exceptions import NotFoundError, ToolError, ValidationError +from fastmcp.exceptions import DisabledError, NotFoundError, ToolError, ValidationError from fastmcp_extensions.tool_filters import ToolUnavailableError from airbyte._util import api_util, meta @@ -341,85 +337,114 @@ def test_cloud_error_trace_message_ignores_missing_resource_without_http_error() assert _error_handling.cloud_error_trace_message(error) is None -def test_tool_error_manifest_shape_and_messages_are_safe() -> None: - path = importlib.resources.files("airbyte.mcp").joinpath("tool_errors.yaml") - entries = yaml.safe_load(path.read_text(encoding="utf-8")) - - assert list(entries) == sorted(entries) - for key, entry in entries.items(): - assert set(entry) == {"message"} - message = entry["message"] - assert isinstance(message, str) and message.strip() - module_name, _, class_name = key.rpartition(".") - error_class = getattr(importlib.import_module(module_name), class_name) - assert inspect.isclass(error_class) - assert issubclass(error_class, BaseException) - assert "{" not in message - assert re.search(r"\d+", message) is None - assert "http" not in message.lower() - assert '"' not in message +def test_airbyte_exception_docstrings_have_safe_first_lines() -> None: + for name, error_class in inspect.getmembers(airbyte.exceptions, inspect.isclass): + if ( + error_class.__module__ != airbyte.exceptions.__name__ + or not issubclass(error_class, AirbyteLibError) + or error_class is AirbyteLibError + ): + continue + first_line = ( + error_class.__doc__.split("\n")[0].strip() if error_class.__doc__ else "" + ) + assert first_line, name + assert "{" not in first_line, name + assert "}" not in first_line, name + assert "://" not in first_line, name def test_every_airbyte_exception_has_a_specific_tool_error_message() -> None: - messages = _error_handling._tool_error_messages() - for name, error_class in inspect.getmembers(airbyte.exceptions, inspect.isclass): - if not issubclass(error_class, AirbyteLibError) or error_class in { - AirbyteLibError, - AirbyteCloudApiError, - }: + if ( + error_class.__module__ != airbyte.exceptions.__name__ + or not issubclass(error_class, AirbyteLibError) + or error_class in {AirbyteLibError, AirbyteCloudApiError} + ): continue - mro_keys = [ - f"{parent.__module__}.{parent.__qualname__}" - for parent in error_class.__mro__ - if parent is not AirbyteLibError - ] - assert any(key in messages for key in mro_keys), name + expected = error_class.__doc__.split("\n")[0].strip() + assert _error_handling.tool_error_trace_message(error_class()) == expected, name @pytest.mark.parametrize( - ("error", "expected_class"), + ("error", "expected"), + [ + pytest.param( + asyncio.CancelledError("private cancellation"), + "The tool call was cancelled.", + id="cancelled", + ), + pytest.param( + NotImplementedError("private operation"), + "This operation isn't supported for this resource type yet.", + id="not-implemented", + ), + pytest.param( + DisabledError("private disabled tool"), + "The requested tool is disabled on this server.", + id="disabled-tool", + ), + pytest.param( + NotFoundError("Unknown tool: private name"), + "The requested tool doesn't exist on this server.", + id="unknown-tool", + ), + pytest.param( + ValidationError("private invalid argument"), + "The tool arguments failed validation.", + id="validation", + ), + ], +) +def test_non_airbyte_exception_classes_use_fixed_trace_messages( + error: BaseException, expected: str +) -> None: + message = _error_handling.tool_error_trace_message(error) + + assert message == expected + assert str(error) not in message + + +@pytest.mark.parametrize( + ("error", "expected"), [ pytest.param( AirbyteSafeModeError(message="Blocked for 'abc-secret-id-123'."), - AirbyteSafeModeError, + "A destructive operation was blocked by MCP safe mode.", id="safe-mode", ), pytest.param( AirbyteMissingWorkspaceContextError(), - AirbyteMissingWorkspaceContextError, + "Workspace ID is required but not provided.", id="missing-workspace", ), pytest.param( AirbyteLibInputError(message="bad value xyz-123"), - AirbyteLibInputError, + "The input provided to PyAirbyte did not match expected validation rules.", id="library-input", ), pytest.param( ToolError("outer wrapper"), - AirbyteSafeModeError, + "A destructive operation was blocked by MCP safe mode.", id="wrapped-safe-mode", ), pytest.param( NotFoundError("Unknown tool: foo"), - NotFoundError, + "The requested tool doesn't exist on this server.", id="unknown-tool", ), pytest.param( ValidationError("invalid argument"), - ValidationError, + "The tool arguments failed validation.", id="fastmcp-validation", ), ], ) -def test_tool_error_trace_message_uses_exception_manifest( - error: BaseException, expected_class: type[BaseException] +def test_tool_error_trace_message_uses_fixed_exception_class_messages( + error: BaseException, expected: str ) -> None: if isinstance(error, ToolError): error.__cause__ = AirbyteSafeModeError(message="wrapped secret abc-123") - expected = _error_handling._tool_error_messages()[ - f"{expected_class.__module__}.{expected_class.__qualname__}" - ] message = _error_handling.tool_error_trace_message(error) assert message == expected @@ -670,9 +695,7 @@ def test_fastmcp_rejections_are_classified_like_the_otel_backend() -> None: "error.type": "NotFoundError", "airbyte.mcp.error.category": "unknown_tool", "airbyte.mcp.error.fault": "caller", - "error.message": _error_handling._tool_error_messages()[ - "fastmcp.exceptions.NotFoundError" - ], + "error.message": "The requested tool doesn't exist on this server.", } @@ -1394,11 +1417,7 @@ async def run(): ) assert ( unknown_tool_span.get_tag("error.message") - == ( - _error_handling._tool_error_messages()[ - "fastmcp.exceptions.NotFoundError" - ] - ) + == "The requested tool doesn't exist on this server." ) safe_span = next( span @@ -1412,19 +1431,11 @@ async def run(): ) assert ( safe_span.get_tag("error.message") - == ( - _error_handling._tool_error_messages()[ - "airbyte.exceptions.AirbyteSafeModeError" - ] - ) + == "A destructive operation was blocked by MCP safe mode." ) assert ( validation_span.get_tag("error.message") - == ( - _error_handling._tool_error_messages()[ - "fastmcp.exceptions.ValidationError" - ] - ) + == "The tool arguments failed validation." ) for span, exception_name in ( (safe_span, "AirbyteSafeModeError"),