feat(mcp): use fixed Cloud error message as Datadog span error.message

airbyte/_util/api_util.py

@@ -71,6 +71,9 @@
     "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)."
+)
 ACTOR_NOT_READY_MCP_GUIDANCE = (
     "This is expected right after creating a connector with `defer_credentials=True`. "
     "Don't retry yet. Ask the user to open the connector's settings page, complete "
@@ -197,6 +200,16 @@ def _is_actor_not_ready_error(error: SDKError) -> bool:
     )
 
 
+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 _wrap_sdk_error(
     error: SDKError, base_context: dict[str, Any] | None = None
 ) -> AirbyteCloudError:
@@ -223,19 +236,14 @@ def _wrap_sdk_error(
 
     status_code = sdk_context.get("status_code")
     is_forbidden = status_code == HTTPStatus.FORBIDDEN
-    message, guidance = describe_cloud_error(problem)
+    _, guidance = describe_cloud_error(problem)
     error_type = (
         AirbyteMissingResourceError
         if is_forbidden or status_code == HTTPStatus.NOT_FOUND
         else AirbyteCloudError
     )
     return error_type(
-        message=(
-            "The requested resource was not found, or these credentials can't access it "
-            "(HTTP 403)."
-            if is_forbidden
-            else message
-        ),
+        message=sdk_error_message(error),
         guidance=FORBIDDEN_RESOURCE_GUIDANCE if is_forbidden else guidance,
         context=merged_context,
     )

airbyte/mcp/_datadog.py

@@ -36,6 +36,8 @@
 from airbyte.mcp._error_handling import (
     MCP_TOOL_USER_FACING_ERRORS,
     classify_mcp_tool_error,
+    cloud_error_route,
+    cloud_error_trace_message,
     mcp_tool_error_reason,
 )
 from airbyte.mcp._otel import _arg_key, _env, _flag
@@ -83,6 +85,8 @@
 _ARG_TRACER = ArgTracer(_ARG_PREFIX, key=_arg_key, skip=(INTENT_ARG,))
 _TOOL_ERROR_MESSAGE = "tool resulted in an error"
 _STATUS_CODE = "airbyte.mcp.upstream.status_code"
+_UPSTREAM_METHOD = "airbyte.mcp.upstream.method"
+_UPSTREAM_ROUTE = "airbyte.mcp.upstream.route"
 
 
 def _build_tool_maps() -> None:
@@ -202,6 +206,13 @@ def _exception_attributes(error: BaseException) -> dict[str, str]:
     }
     if (stack := error_stack(cause)) is not None:
         attrs["error.stack"] = stack
+    if route := cloud_error_route(cause):
+        attrs[_UPSTREAM_METHOD], attrs[_UPSTREAM_ROUTE] = route
+    try:
+        if message := cloud_error_trace_message(cause):
+            attrs["error.message"] = message
+    except Exception:
+        logger.debug("Datadog Cloud error message unavailable")
     return attrs
 
 
@@ -246,6 +257,15 @@ def _tool_error_message(span: Span) -> str:
     return f"{_TOOL_ERROR_MESSAGE} ({details})" if details else _TOOL_ERROR_MESSAGE
 
 
+def _raised_tool_error_message(error: BaseException, span: Span) -> str:
+    """Return a fixed Cloud message when available, falling back to the span summary."""
+    try:
+        message = cloud_error_trace_message(error)
+    except Exception:
+        message = None
+    return message if message is not None else _tool_error_message(span)
+
+
 class _StripMetaTraceContextMiddleware:
     async def __call__(
         self,
@@ -520,14 +540,14 @@ async def _trace_call(
                 logger.debug("Datadog tool attributes unavailable")
 
 
-def _annotate_attributes(span: Span, source: Mapping[str, str | bool]) -> None:
+def _annotate_attributes(
+    span: Span, source: Mapping[str, str | bool], *, llmobs: bool = True
+) -> None:
     """Keep native APM attributes and LLM metadata consistent.
 
     Argument records pass the OTel backend's export validation: ints become
     metrics, and the validated records also reach LLM metadata.
     """
-    from ddtrace.llmobs import LLMObs  # noqa: PLC0415
-
     attrs = dict(source)
     arg_attrs = {key: attrs.pop(key) for key in list(attrs) if is_arg_key(_ARG_PREFIX, key)}
     tool = attrs.get("gen_ai.tool.name")
@@ -553,13 +573,14 @@ def _annotate_attributes(span: Span, source: Mapping[str, str | bool]) -> None:
     metadata.update({key.removeprefix("airbyte.mcp."): value for key, value in accepted.items()})
     if "gen_ai.tool.call.id" in attrs:
         metadata["tool_id"] = attrs["gen_ai.tool.call.id"]
-    LLMObs.annotate(span, metadata=metadata)
+    if llmobs:
+        from ddtrace.llmobs import LLMObs  # noqa: PLC0415
 
+        LLMObs.annotate(span, metadata=metadata)
 
-def _annotate_request(span: Span, ctx: ServerRequestContext[Any]) -> None:
-    """Annotate protocol identity without copying tool arguments into the SDK."""
-    from ddtrace.llmobs import LLMObs  # noqa: PLC0415
 
+def _annotate_request(span: Span, ctx: ServerRequestContext[Any], *, llmobs: bool = True) -> None:
+    """Annotate protocol identity without copying tool arguments into the SDK."""
     request_types = {
         "initialize": InitializeRequest,
         "tools/list": ListToolsRequest,
@@ -605,26 +626,31 @@ def _annotate_request(span: Span, ctx: ServerRequestContext[Any]) -> None:
             )
         except Exception:
             logger.debug("Early tool scope unavailable")
-    _annotate_attributes(span, attrs)
+    _annotate_attributes(span, attrs, llmobs=llmobs)
     session = attrs.get("airbyte.mcp.session_id")
     if session:
         tags["mcp_session_id"] = session
-    LLMObs.annotate(
-        span,
-        tags=tags,
-        metadata={"pyairbyte.version": get_version()},
-        # Seed scalar payloads so the processor can replace Input even if the
-        # tool fails before the metadata middleware runs. Never seed raw data.
-        input_data=REDACTED if tool_call else None,
-        output_data=REDACTED if tool_call else None,
-    )
-    if not tool_call:
-        request = (
-            request_types[ctx.method]
-            .model_validate({"method": ctx.method, "params": params})
-            .model_dump(exclude={"params": {"meta": "_dd_trace_context"}})
+    if llmobs:
+        from ddtrace.llmobs import LLMObs  # noqa: PLC0415
+
+        LLMObs.annotate(
+            span,
+            tags=tags,
+            metadata={"pyairbyte.version": get_version()},
+            # Seed scalar payloads so the processor can replace Input even if the
+            # tool fails before the metadata middleware runs. Never seed raw data.
+            input_data=REDACTED if tool_call else None,
+            output_data=REDACTED if tool_call else None,
         )
-        LLMObs.annotate(span, input_data=request)
+        if not tool_call:
+            request = (
+                request_types[ctx.method]
+                .model_validate({"method": ctx.method, "params": params})
+                .model_dump(exclude={"params": {"meta": "_dd_trace_context"}})
+            )
+            LLMObs.annotate(span, input_data=request)
+    else:
+        span.set_tags(tags)
 
 
 class _DatadogRequestMiddleware:
@@ -667,8 +693,8 @@ async def __call__(  # noqa: PLR0912, PLR0915
         from ddtrace import tracer  # noqa: PLC0415
         from ddtrace.llmobs import LLMObs  # noqa: PLC0415
 
-        # Native server instrumentation covered initialize/call. Listing is an
-        # explicit addition; other operations retain deployment HTTP tracing.
+        # Use APM spans for lifecycle methods and LLM Obs spans for tool calls.
+        # Other operations retain deployment HTTP tracing.
         if ctx.method not in {"initialize", "tools/list", "tools/call"}:
             return await call_next(ctx)
         span = None
@@ -696,17 +722,16 @@ async def __call__(  # noqa: PLR0912, PLR0915
                 except Exception:
                     logger.debug("Datadog MCP distributed context unavailable")
             name = str(params.get("name", "unknown_tool")) if tool_call else f"mcp.{ctx.method}"
-            start = LLMObs.tool if tool_call else LLMObs.task
-            span = start(name=name)
-            # LLMObs retains its display name independently of these APM fields.
-            span.name = f"mcp.{ctx.method}"
+            if tool_call:
+                span = LLMObs.tool(name=name)
+                span.name = f"mcp.{ctx.method}"
+            else:
+                span = tracer.trace(f"mcp.{ctx.method}", resource="server_request")
             if tool_call:
                 # Per-tool APM trace metrics group by resource; keep it bounded.
                 span.resource = name if name in _TOOL_MODULES else "unknown_tool"
-            else:
-                span.resource = "server_request"
             span.set_metric("_dd.measured", 1)
-            _annotate_request(span, ctx)
+            _annotate_request(span, ctx, llmobs=tool_call)
         except Exception:
             logger.debug("Datadog request attributes unavailable")
         try:
@@ -719,7 +744,7 @@ async def __call__(  # noqa: PLR0912, PLR0915
                         "tools/list": ListToolsResult,
                         "tools/call": CallToolResult,
                     }[ctx.method].model_validate(result)
-                    output = REDACTED if tool_call else response.model_dump(mode="json")
+                    output = REDACTED
                     if tool_call and getattr(response, "is_error", False):
                         span.error = 1
                         error_type = span.get_tag("airbyte.mcp.error_type") or "ToolError"
@@ -732,11 +757,15 @@ async def __call__(  # noqa: PLR0912, PLR0915
                         # A returned error result raised nothing to classify.
                         if span.get_tag("airbyte.mcp.error.category") is None:
                             error_attrs.update(_error_classification(None))
-                        _annotate_attributes(span, error_attrs)
-                        span.set_tag("error.message", _tool_error_message(span))
+                        _annotate_attributes(span, error_attrs, llmobs=tool_call)
+                        if span.get_tag("error.message") is None:
+                            span.set_tag("error.message", _tool_error_message(span))
                     else:
-                        _annotate_attributes(span, {"airbyte.mcp.outcome": "success"})
-                    LLMObs.annotate(span, output_data=output)
+                        _annotate_attributes(
+                            span, {"airbyte.mcp.outcome": "success"}, llmobs=tool_call
+                        )
+                    if tool_call:
+                        LLMObs.annotate(span, output_data=output)
                 except Exception:
                     logger.debug("Datadog response attributes unavailable")
             return result
@@ -747,9 +776,9 @@ async def __call__(  # noqa: PLR0912, PLR0915
                     # and traceback, which can contain arguments and results.
                     if error := sys.exc_info()[1]:
                         span.error = 1
-                        _annotate_attributes(span, _exception_attributes(error))
+                        _annotate_attributes(span, _exception_attributes(error), llmobs=tool_call)
                         if tool_call:
-                            span.set_tag("error.message", _tool_error_message(span))
+                            span.set_tag("error.message", _raised_tool_error_message(error, span))
                     span.__exit__(None, None, None)
             except Exception:
                 logger.debug("Datadog span completion failed")

airbyte/mcp/_error_handling.py

@@ -8,14 +8,21 @@
 import re
 from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any
+from urllib.parse import urlsplit
 
+import requests
 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
 from fastmcp_extensions.otel._extras import _chain, declared_parameters  # noqa: PLC2701
 from fastmcp_extensions.otel.models import TraceArg
 
-from airbyte._util.api_util import error_response_body, sdk_error_response
+from airbyte._util.api_util import (
+    SDKError,
+    error_response_body,
+    sdk_error_message,
+    sdk_error_response,
+)
 from airbyte._util.cloud_errors import (
     describe_cloud_error,
     is_valid_problem_slug,
@@ -45,6 +52,29 @@
 
 logger = logging.getLogger(__name__)
 
+_UPSTREAM_METHODS = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"})
+# Literal segments come from path templates in airbyte_api.
+_CLOUD_API_ROUTE_SEGMENTS = frozenset(
+    {
+        "connections",
+        "destinations",
+        "health",
+        "jobs",
+        "organizations",
+        "oauthCredentials",
+        "permissions",
+        "sources",
+        "initiateOAuth",
+        "streams",
+        "tags",
+        "users",
+        "workspaces",
+        "definitions",
+        "declarative_sources",
+    }
+)
+_MAX_UPSTREAM_ROUTE_LENGTH = 200
+
 
 MCP_TOOL_USER_FACING_ERRORS: tuple[type[AirbyteLibError], ...] = (
     AirbyteLibInputError,
@@ -60,6 +90,78 @@
 """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
+    for cause in _chain(error):
+        if (
+            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
+
+
+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):
+        if not isinstance(cause, SDKError):
+            continue
+        try:
+            response = getattr(cause, "raw_response", None)
+            request = getattr(response, "request", None)
+            if request is None:
+                continue
+
+            method = request.method.upper()
+            if method not in _UPSTREAM_METHODS:
+                return None
+
+            segments = [
+                segment for segment in urlsplit(str(request.url)).path.split("/") if segment
+            ]
+            version_index = next(
+                (
+                    index
+                    for index in range(len(segments) - 1, -1, -1)
+                    if re.fullmatch(r"v[0-9]{1,3}", segments[index])
+                ),
+                None,
+            )
+            route_segments: list[str] = []
+            for index, segment in enumerate(segments):
+                if version_index is None or index < version_index:
+                    route_segments.append("{id}")
+                elif index == version_index or segment in _CLOUD_API_ROUTE_SEGMENTS:
+                    route_segments.append(segment)
+                else:
+                    route_segments.append("{id}")
+            route = "/" + "/".join(route_segments)
+            if len(route) > _MAX_UPSTREAM_ROUTE_LENGTH:
+                return None
+        except Exception:
+            return None
+        else:
+            return method, route
+    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(