Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .changes/unreleased/BugFix-20260412-670.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
kind: Bug Fix
body: Improve error messages when dbt Cloud platform credentials expire for Semantic Layer tools (query_metrics, execute_sql). The error now clearly indicates this is a platform-side issue and directs users to the dbt Cloud UI to refresh their development credentials.
time: 2026-04-12T21:00:00.000000-07:00
46 changes: 44 additions & 2 deletions src/dbt_mcp/semantic_layer/client.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import asyncio
import base64
import json
import re
from collections.abc import Callable
from contextlib import AbstractContextManager
from datetime import date, datetime, time, timedelta
Expand Down Expand Up @@ -38,6 +39,35 @@
ListMetricsResponse,
)

# Patterns that indicate expired dbt Cloud platform credentials rather than
# local authentication failures. These are matched case-insensitively against
# the raw error string returned by the Semantic Layer / data platform.
_PLATFORM_CREDENTIAL_ERROR_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"SSO authentication has expired", re.IGNORECASE),
re.compile(r"re-connect to Snowflake", re.IGNORECASE),
re.compile(r"refresh.snowflake.oauth.credentials", re.IGNORECASE),
re.compile(r"authentication token has expired", re.IGNORECASE),
re.compile(r"oauth.*token.*expired", re.IGNORECASE),
re.compile(r"token.*expired.*re-?authenticate", re.IGNORECASE),
]
Comment on lines +45 to +52
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Left a comment related to these lines in the issue. Let me know what you think! We can continue that convo there for now.


_PLATFORM_CREDENTIAL_HINT = (
"\n\n"
"Hint: This error indicates that your dbt Cloud development credentials "
"have expired. This is a dbt Cloud platform credential issue, not a local "
"authentication problem.\n"
"To fix this, go to the dbt Cloud UI → Profile → Credentials and refresh "
"your development environment credentials for the relevant project.\n"
"See: https://docs.getdbt.com/docs/dbt-cloud-environments/develop-in-the-cloud#access-the-cloud-ide"
)


def _is_platform_credential_error(error_message: str) -> bool:
"""Detect whether an error message indicates expired dbt Cloud platform credentials."""
return any(
pattern.search(error_message) for pattern in _PLATFORM_CREDENTIAL_ERROR_PATTERNS
)


def DEFAULT_RESULT_FORMATTER(table: pa.Table) -> str:
"""Convert PyArrow Table to JSON string with ISO date formatting.
Expand Down Expand Up @@ -268,7 +298,12 @@ async def get_entities(
return self.entities_cache[metrics_key]

def _format_semantic_layer_error(self, error: Exception) -> str:
"""Format semantic layer errors by cleaning up common error message patterns."""
"""Format semantic layer errors by cleaning up common error message patterns.

Additionally detects errors caused by expired dbt Cloud platform credentials
and appends an actionable hint directing users to refresh their credentials
in the dbt Cloud UI rather than re-authenticating locally.
"""
error_str = str(error)
formatted = (
error_str.replace("QueryFailedError(", "")
Expand All @@ -285,7 +320,14 @@ def _format_semantic_layer_error(self, error: Exception) -> str:
.strip()
)
if not formatted:
return error_str or f"Semantic layer query failed: {type(error).__name__}"
formatted = (
error_str or f"Semantic layer query failed: {type(error).__name__}"
)

# Detect platform credential expiry and append an actionable hint
if _is_platform_credential_error(error_str):
formatted += _PLATFORM_CREDENTIAL_HINT

return formatted

def _format_get_metrics_compiled_sql_error(
Expand Down
120 changes: 120 additions & 0 deletions tests/unit/semantic_layer/test_platform_credential_errors.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
"""Tests for platform credential error detection in Semantic Layer tools.

Verifies that when query_metrics or get_metrics_compiled_sql fail due to
expired dbt Cloud platform credentials, the error message clearly indicates
this is a platform-side issue and directs users to the dbt Cloud UI.

Related: https://github.com/dbt-labs/dbt-mcp/issues/670
"""

import pytest

from dbt_mcp.semantic_layer.client import (
SemanticLayerFetcher,
_PLATFORM_CREDENTIAL_HINT,
_is_platform_credential_error,
)


@pytest.fixture
def fetcher():
from unittest.mock import AsyncMock

return SemanticLayerFetcher(client_provider=AsyncMock())


class TestPlatformCredentialErrorDetection:
"""Tests for _is_platform_credential_error."""

def test_detects_sso_authentication_expired(self) -> None:
assert _is_platform_credential_error(
"SSO authentication has expired, please re-connect to Snowflake"
)

def test_detects_reconnect_to_snowflake(self) -> None:
assert _is_platform_credential_error(
"Please re-connect to Snowflake: https://docs.getdbt.com/faqs"
)

def test_detects_refresh_snowflake_oauth_url(self) -> None:
assert _is_platform_credential_error(
"https://docs.getdbt.com/faqs/Troubleshooting/refresh-snowflake-oauth-credentials"
)

def test_detects_authentication_token_expired(self) -> None:
assert _is_platform_credential_error(
"Authentication token has expired for the data warehouse"
)

def test_detects_oauth_token_expired(self) -> None:
assert _is_platform_credential_error(
"OAuth access token has expired for this connection"
)

def test_case_insensitive(self) -> None:
assert _is_platform_credential_error("sso AUTHENTICATION Has Expired")

def test_does_not_match_generic_query_error(self) -> None:
assert not _is_platform_credential_error("column 'revenue' not found in table")

def test_does_not_match_syntax_error(self) -> None:
assert not _is_platform_credential_error(
"SQL compilation error: syntax error at position 42"
)

def test_does_not_match_timeout_error(self) -> None:
assert not _is_platform_credential_error("Query timed out after 60 seconds")

def test_does_not_match_empty_string(self) -> None:
assert not _is_platform_credential_error("")


class TestFormatSemanticLayerErrorWithCredentialHint:
"""Tests that _format_semantic_layer_error appends the credential hint."""

def test_sso_expired_includes_platform_hint(self, fetcher) -> None:
error = Exception(
"SSO authentication has expired, please re-connect to Snowflake: "
"https://docs.getdbt.com/faqs/Troubleshooting/refresh-snowflake-oauth-credentials"
)
result = fetcher._format_semantic_layer_error(error)
assert "dbt Cloud" in result
assert "Profile → Credentials" in result
assert "not a local authentication problem" in result

def test_sso_expired_preserves_original_message(self, fetcher) -> None:
error = Exception(
"SSO authentication has expired, please re-connect to Snowflake"
)
result = fetcher._format_semantic_layer_error(error)
assert "SSO authentication has expired" in result

def test_generic_error_does_not_include_platform_hint(self, fetcher) -> None:
error = Exception("column 'revenue' not found")
result = fetcher._format_semantic_layer_error(error)
assert _PLATFORM_CREDENTIAL_HINT not in result
assert "dbt Cloud UI" not in result

def test_query_failed_error_with_credential_message_includes_hint(
self, fetcher
) -> None:
"""QueryFailedError wrapping a credential expiry should also get the hint."""
error = Exception(
'QueryFailedError(["SSO authentication has expired, '
'please re-connect to Snowflake"])'
)
result = fetcher._format_semantic_layer_error(error)
assert "dbt Cloud" in result
assert "Profile → Credentials" in result

def test_format_query_failed_error_with_credential_message(self, fetcher) -> None:
"""_format_query_failed_error should also include the hint for credential errors."""
from dbtsl.error import QueryFailedError

error = QueryFailedError(
message="SSO authentication has expired, please re-connect to Snowflake",
status="FAILED",
)
result = fetcher._format_query_failed_error(error)
assert "dbt Cloud" in result.error
assert "Profile → Credentials" in result.error