feat(mcp): reuse caller's verified token for Airbyte Cloud API passthrough

airbyte/mcp/cloud.py

@@ -267,22 +267,49 @@ def _get_cloud_workspace(
     return _get_cloud_client(ctx).get_workspace(resolved_workspace_id)
 
 
+def _none_if_empty(value: str) -> str | None:
+    """Return `value` unless it is empty, in which case return `None`.
+
+    Emptiness is checked with `len` rather than truthiness because a
+    `SecretString` overrides `__bool__` to always be truthy, so an empty secret
+    would otherwise be mistaken for a present value.
+    """
+    return value if len(value) > 0 else None
+
+
 def _get_cloud_client(
     ctx: Context,
     *,
     organization_id: str | None = None,
 ) -> CloudClient:
-    """Get an authenticated `CloudClient` from MCP config."""
-    bearer_token = get_mcp_config(ctx, MCP_CONFIG_BEARER_TOKEN)
-    client_id = get_mcp_config(ctx, MCP_CONFIG_CLIENT_ID)
-    client_secret = get_mcp_config(ctx, MCP_CONFIG_CLIENT_SECRET)
+    """Get an authenticated `CloudClient` from MCP config.
+
+    A bearer token takes precedence over `client_id`/`client_secret`. When the
+    server verifies transport against Airbyte Cloud's realm, the verified
+    transport token (surfaced through `BEARER_TOKEN_CONFIG_ARG`) is itself a
+    valid Cloud API bearer, so reusing it delegates downstream Cloud calls to the
+    caller's identity. Client credentials remain the fallback when no bearer
+    token is present (for example, stdio/local mode). `CloudClient` rejects
+    receiving both at once, so the client credentials are dropped when a bearer
+    token is available.
+    """
+    bearer_token = _none_if_empty(get_mcp_config(ctx, MCP_CONFIG_BEARER_TOKEN))
+    client_id = _none_if_empty(get_mcp_config(ctx, MCP_CONFIG_CLIENT_ID))
+    client_secret = _none_if_empty(get_mcp_config(ctx, MCP_CONFIG_CLIENT_SECRET))
     api_url = get_mcp_config(ctx, MCP_CONFIG_API_URL)
     config_api_url = get_mcp_config(ctx, MCP_CONFIG_CONFIG_API_URL)
 
+    if bearer_token is not None:
+        return CloudClient(
+            bearer_token=bearer_token,
+            public_api_root=api_url,
+            config_api_root=config_api_url,
+            organization_id=organization_id,
+        )
+
     return CloudClient(
         client_id=client_id,
         client_secret=client_secret,
-        bearer_token=bearer_token,
         public_api_root=api_url,
         config_api_root=config_api_url,
         organization_id=organization_id,

tests/unit_tests/test_mcp_cloud.py

@@ -9,12 +9,21 @@
 
 import pytest
 from airbyte.cloud.models import JobStatusEnum
+from airbyte.constants import (
+    MCP_CONFIG_API_URL,
+    MCP_CONFIG_BEARER_TOKEN,
+    MCP_CONFIG_CLIENT_ID,
+    MCP_CONFIG_CLIENT_SECRET,
+    MCP_CONFIG_CONFIG_API_URL,
+)
 from airbyte.mcp import cloud as cloud_mcp
 from airbyte.mcp.cloud import (
     CloudConnectionResult,
     CloudDestinationResult,
     CloudSourceResult,
+    _get_cloud_client,
 )
+from airbyte.secrets.base import SecretString
 from fastmcp import Context
 
 
@@ -258,3 +267,75 @@ def test_mcp_cloud_connections_apply_limit_after_status_filter(
     assert workspace.limits["connections"] is None
     assert len(results) == 1
     assert results[0].id == "connection-2"
+
+
+def _secret_or_none(value: SecretString | None) -> str | None:
+    """Return the plain string of a `SecretString`, or `None`."""
+    return None if value is None else str(value)
+
+
+@pytest.mark.parametrize(
+    "bearer_token, expected_bearer, expected_client_id, expected_client_secret",
+    [
+        pytest.param(
+            "transport-jwt",
+            "transport-jwt",
+            None,
+            None,
+            id="bearer_token_wins_over_client_credentials",
+        ),
+        pytest.param(
+            "",
+            None,
+            "client-id",
+            "client-secret",
+            id="falls_back_to_client_credentials_without_bearer_token",
+        ),
+        pytest.param(
+            SecretString(""),
+            None,
+            "client-id",
+            "client-secret",
+            id="empty_secret_string_bearer_falls_back_to_client_credentials",
+        ),
+        pytest.param(
+            SecretString("transport-jwt"),
+            "transport-jwt",
+            None,
+            None,
+            id="secret_string_bearer_wins_over_client_credentials",
+        ),
+    ],
+)
+def test_get_cloud_client_prefers_bearer_token_over_client_credentials(
+    monkeypatch: pytest.MonkeyPatch,
+    bearer_token: str | SecretString,
+    expected_bearer: str | None,
+    expected_client_id: str | None,
+    expected_client_secret: str | None,
+) -> None:
+    """Verify `_get_cloud_client` uses the bearer token when present, else client creds.
+
+    An empty bearer token models stdio mode, where `_resolve_transport_bearer_token`
+    yields `""` and client credentials remain the fallback. The `SecretString`
+    cases guard the empty-string comparison against `SecretString.__bool__`, which
+    is always `True` and would otherwise treat an empty secret as present.
+    """
+    config = {
+        MCP_CONFIG_BEARER_TOKEN: bearer_token,
+        MCP_CONFIG_CLIENT_ID: "client-id",
+        MCP_CONFIG_CLIENT_SECRET: "client-secret",
+        MCP_CONFIG_API_URL: None,
+        MCP_CONFIG_CONFIG_API_URL: None,
+    }
+    monkeypatch.setattr(
+        cloud_mcp,
+        "get_mcp_config",
+        lambda _ctx, key: config.get(key),
+    )
+
+    client = _get_cloud_client(cast(Context, object()))
+
+    assert _secret_or_none(client.bearer_token) == expected_bearer
+    assert _secret_or_none(client.client_id) == expected_client_id
+    assert _secret_or_none(client.client_secret) == expected_client_secret

tests/unit_tests/test_mcp_tool_utils.py

@@ -10,6 +10,7 @@
 from airbyte.mcp._tool_utils import (
     SafeModeError,
     _GUIDS_CREATED_IN_SESSION,
+    _resolve_transport_bearer_token,
     check_guid_created_in_session,
     register_guid_created_in_session,
 )
@@ -66,3 +67,40 @@ def test_duplicate_guid_registration_is_idempotent() -> None:
     register_guid_created_in_session("duplicate-guid")
     register_guid_created_in_session("duplicate-guid")
     assert "duplicate-guid" in _GUIDS_CREATED_IN_SESSION
+
+
+class _FakeAccessToken:
+    """Minimal stand-in for a FastMCP `AccessToken` carrying a token string."""
+
+    def __init__(self, token: str) -> None:
+        """Store the token value exposed via the `token` attribute."""
+        self.token = token
+
+
+@pytest.mark.parametrize(
+    "access_token, expected",
+    [
+        pytest.param(None, "", id="stdio_mode_returns_empty_string"),
+        pytest.param(_FakeAccessToken(""), "", id="empty_token_returns_empty_string"),
+        pytest.param(
+            _FakeAccessToken("verified-jwt"),
+            "verified-jwt",
+            id="verified_token_is_returned",
+        ),
+    ],
+)
+def test_resolve_transport_bearer_token(
+    access_token: _FakeAccessToken | None,
+    expected: str,
+) -> None:
+    """Verify the transport-token resolver returns the verified token or `""`.
+
+    `get_access_token` returns `None` outside a request (stdio mode), so the
+    resolver yields `""` and downstream config resolution falls back to client
+    credentials.
+    """
+    with patch(
+        "airbyte.mcp._tool_utils.get_access_token",
+        return_value=access_token,
+    ):
+        assert _resolve_transport_bearer_token() == expected