From 924605975b29c5c132ab53556976a51e7e0c9f58 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 24 Jul 2026 18:56:38 +0000 Subject: [PATCH 1/2] fix(mcp): route interactive Cloud API reads through the Config API Interactive OIDC logins produce a user-realm bearer token with no client credentials. The public API (api.airbyte.com) rejects such tokens, so workspace and connection metadata reads returned 401. Route bearer-only reads through the Config API (which accepts user tokens) while keeping the public API path for application client credentials. The Config API connection response embeds the source/destination objects and sync catalog, so connector names and stream names resolve from one response instead of extra public API lookups. Co-Authored-By: AJ Steers --- airbyte/_util/api_util.py | 60 +++++ airbyte/cloud/connections.py | 74 +++++- airbyte/cloud/models.py | 59 ++++- airbyte/cloud/workspaces.py | 39 ++++ airbyte/mcp/cloud.py | 16 +- .../test_cloud_config_api_routing.py | 210 ++++++++++++++++++ 6 files changed, 434 insertions(+), 24 deletions(-) create mode 100644 tests/unit_tests/test_cloud_config_api_routing.py diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py index e3f0a9eb5..35edb1aeb 100644 --- a/airbyte/_util/api_util.py +++ b/airbyte/_util/api_util.py @@ -285,6 +285,34 @@ def get_workspace( ) +def get_workspace_via_config_api( + workspace_id: str, + *, + api_root: str, + client_id: SecretString | None, + client_secret: SecretString | None, + bearer_token: SecretString | None, + config_api_root: str | None = None, +) -> dict[str, Any]: + """Get a workspace via the Config API endpoint `POST /v1/workspaces/get`. + + Unlike `get_workspace`, this uses the internal Config API rather than the public API. + The Config API accepts user-realm bearer tokens (from an interactive OIDC login), + which the public API rejects because it only accepts application-client tokens. + + Returns the raw `WorkspaceRead` response as a dictionary. + """ + return _make_config_api_request( + path="/workspaces/get", + json={"workspaceId": workspace_id}, + api_root=api_root, + config_api_root=config_api_root, + client_id=client_id, + client_secret=client_secret, + bearer_token=bearer_token, + ) + + def create_workspace( *, name: str, @@ -798,6 +826,38 @@ def get_connection( ) +def get_connection_via_config_api( + connection_id: str, + *, + api_root: str, + client_id: SecretString | None, + client_secret: SecretString | None, + bearer_token: SecretString | None, + config_api_root: str | None = None, +) -> dict[str, Any]: + """Get a connection via the Config API endpoint `POST /v1/web_backend/connections/get`. + + Unlike `get_connection`, this uses the internal Config API rather than the public API. + The Config API accepts user-realm bearer tokens (from an interactive OIDC login), + which the public API rejects because it only accepts application-client tokens. + + The `WebBackendConnectionRead` response embeds the full `source` and `destination` + objects (including their names), so callers can resolve connector names without + issuing separate public API lookups. + + Returns the raw `WebBackendConnectionRead` response as a dictionary. + """ + return _make_config_api_request( + path="/web_backend/connections/get", + json={"connectionId": connection_id, "withRefreshedCatalog": False}, + api_root=api_root, + config_api_root=config_api_root, + client_id=client_id, + client_secret=client_secret, + bearer_token=bearer_token, + ) + + def run_connection( workspace_id: str, connection_id: str, diff --git a/airbyte/cloud/connections.py b/airbyte/cloud/connections.py index 528b91de9..90286925c 100644 --- a/airbyte/cloud/connections.py +++ b/airbyte/cloud/connections.py @@ -111,16 +111,32 @@ def _fetch_connection_info( self._verify_workspace_match(self._connection_info) return self._connection_info - # Fetch from API - connection_info = api_util.get_connection( - workspace_id=self.workspace.workspace_id, - connection_id=self.connection_id, - api_root=self.workspace.api_root, - client_id=self.workspace.client_id, - client_secret=self.workspace.client_secret, - bearer_token=self.workspace.bearer_token, - ) - result = CloudConnectionInfo.from_api_response(connection_info) + # Fetch from API. A bearer-only (interactive OIDC) token is rejected by the + # public API, so route those reads through the Config API, which also embeds + # the source/destination objects and sync catalog in a single response. + if self.workspace._uses_bearer_only_auth: # noqa: SLF001 # Internal auth check + raw = api_util.get_connection_via_config_api( + connection_id=self.connection_id, + api_root=self.workspace.api_root, + client_id=self.workspace.client_id, + client_secret=self.workspace.client_secret, + bearer_token=self.workspace.bearer_token, + config_api_root=self.workspace.config_api_root, + ) + result = CloudConnectionInfo.from_config_api_response( + raw, + fallback_workspace_id=self.workspace.workspace_id, + ) + else: + connection_info = api_util.get_connection( + workspace_id=self.workspace.workspace_id, + connection_id=self.connection_id, + api_root=self.workspace.api_root, + client_id=self.workspace.client_id, + client_secret=self.workspace.client_secret, + bearer_token=self.workspace.bearer_token, + ) + result = CloudConnectionInfo.from_api_response(connection_info) self._connection_info = result @@ -242,13 +258,49 @@ def destination(self) -> CloudDestination: ) return self._cloud_destination_object + @property + def source_name(self) -> str | None: + """The display name of the source, if available. + + Prefers a name embedded in the connection info (Config API reads). Falls back to + a separate source lookup for public API reads that do not embed the name. + """ + if not self._connection_info: + self._connection_info = self._fetch_connection_info() + + if self._connection_info.source_name is not None: + return self._connection_info.source_name + + return self.source.name + + @property + def destination_name(self) -> str | None: + """The display name of the destination, if available. + + Prefers a name embedded in the connection info (Config API reads). Falls back to + a separate destination lookup for public API reads that do not embed the name. + """ + if not self._connection_info: + self._connection_info = self._fetch_connection_info() + + if self._connection_info.destination_name is not None: + return self._connection_info.destination_name + + return self.destination.name + @property def stream_names(self) -> list[str]: """The stream names.""" if not self._connection_info: self._connection_info = self._fetch_connection_info() - return [stream.name for stream in self._connection_info.configurations.streams or []] + if self._connection_info.stream_names is not None: + return self._connection_info.stream_names + + configurations = self._connection_info.configurations + if configurations is None: + return [] + return [stream.name for stream in configurations.streams or []] @property def table_prefix(self) -> str: diff --git a/airbyte/cloud/models.py b/airbyte/cloud/models.py index ead68fb63..6f88a54b9 100644 --- a/airbyte/cloud/models.py +++ b/airbyte/cloud/models.py @@ -133,8 +133,12 @@ class CloudConnectionInfo(BaseModel): name: str """The connection name.""" - configurations: Any - """Stream configuration details for the connection.""" + configurations: Any = None + """Stream configuration details for the connection. + + Populated from the public API response. `None` when the connection was read via + the Config API, in which case `stream_names` carries the selected stream names. + """ prefix: str | None = None """The destination table prefix.""" @@ -142,6 +146,15 @@ class CloudConnectionInfo(BaseModel): status: str """The connection status.""" + source_name: str | None = None + """The source display name, when embedded in the response (Config API only).""" + + destination_name: str | None = None + """The destination display name, when embedded in the response (Config API only).""" + + stream_names: list[str] | None = None + """Selected stream names, when the response embeds the sync catalog (Config API only).""" + @classmethod def from_api_response(cls, connection: _ConnectionResponseLike) -> CloudConnectionInfo: """Create a public model from an internal API connection response.""" @@ -156,6 +169,48 @@ def from_api_response(cls, connection: _ConnectionResponseLike) -> CloudConnecti status=_enum_value(connection.status), ) + @classmethod + def from_config_api_response( + cls, + connection: Mapping[str, Any], + *, + fallback_workspace_id: str | None = None, + ) -> CloudConnectionInfo: + """Create a public model from a Config API `WebBackendConnectionRead` response. + + The Config API response embeds the full `source` and `destination` objects and + the `syncCatalog`, so connector names and selected streams are resolved from this + single response rather than via separate public API lookups. + """ + source: Mapping[str, Any] = connection.get("source") or {} + destination: Mapping[str, Any] = connection.get("destination") or {} + sync_catalog: Mapping[str, Any] = connection.get("syncCatalog") or {} + stream_entries: list[Mapping[str, Any]] = sync_catalog.get("streams") or [] + stream_names = [ + entry["stream"]["name"] + for entry in stream_entries + if isinstance(entry.get("stream"), Mapping) and entry["stream"].get("name") + ] + workspace_id = ( + source.get("workspaceId") + or destination.get("workspaceId") + or fallback_workspace_id + or "" + ) + return cls( + connection_id=connection["connectionId"], + workspace_id=workspace_id, + source_id=connection["sourceId"], + destination_id=connection["destinationId"], + name=connection["name"], + configurations=None, + prefix=connection.get("prefix"), + status=_enum_value(connection["status"]), + source_name=source.get("name"), + destination_name=destination.get("name"), + stream_names=stream_names, + ) + class CloudJobInfo(BaseModel): """Information about an Airbyte Cloud job.""" diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 58f5539e7..8dd9d0aac 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -213,6 +213,45 @@ def workspace_url(self) -> str | None: """The web URL of the workspace.""" return f"{get_web_url_root(self.api_root)}/workspaces/{self.workspace_id}" + @property + def _uses_bearer_only_auth(self) -> bool: + """Whether this workspace authenticates with a bearer token and no client credentials. + + An interactive OIDC login produces a user-realm bearer token with no client + credentials. The public API rejects such tokens (it only accepts application-client + tokens), while the Config API accepts them, so metadata reads must route through the + Config API in this case. + """ + return ( + self.bearer_token is not None and self.client_id is None and self.client_secret is None + ) + + def get_workspace_info(self) -> CloudWorkspaceInfo: + """Get metadata about this workspace. + + Routes bearer-only (interactive OIDC) reads through the Config API and all other + reads through the public API. + """ + if self._uses_bearer_only_auth: + raw = api_util.get_workspace_via_config_api( + workspace_id=self.workspace_id, + api_root=self.api_root, + client_id=self.client_id, + client_secret=self.client_secret, + bearer_token=self.bearer_token, + config_api_root=self.config_api_root, + ) + return CloudWorkspaceInfo.from_mapping(raw) + + response = api_util.get_workspace( + workspace_id=self.workspace_id, + api_root=self.api_root, + client_id=self.client_id, + client_secret=self.client_secret, + bearer_token=self.bearer_token, + ) + return CloudWorkspaceInfo.from_api_response(response) + @cached_property def _organization_info(self) -> dict[str, Any]: """Fetch and cache organization info for this workspace. diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 0beab1b41..b19138d4d 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -18,7 +18,6 @@ from pydantic import BaseModel, Field from airbyte import cloud, get_destination, get_source -from airbyte._util import api_util from airbyte.cloud.client import CloudClient from airbyte.cloud.connectors import CustomCloudSourceDefinition from airbyte.cloud.constants import FAILED_STATUSES @@ -567,14 +566,9 @@ def check_airbyte_cloud_workspace( """ workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) - # Get workspace details from the public API using workspace's credentials - workspace_response = api_util.get_workspace( - workspace_id=workspace.workspace_id, - api_root=workspace.api_root, - client_id=workspace.client_id, - client_secret=workspace.client_secret, - bearer_token=workspace.bearer_token, - ) + # Get workspace details. Reads are routed through the Config API for bearer-only + # (interactive OIDC) credentials, which the public API rejects. + workspace_response = workspace.get_workspace_info() # Try to get organization info (including billing), but fail gracefully if we don't have # permissions. Fetching organization info requires ORGANIZATION_READER permissions on the @@ -1006,9 +1000,9 @@ def describe_cloud_connection( connection_name=cast(str, connection.name), connection_url=cast(str, connection.connection_url), source_id=connection.source_id, - source_name=cast(str, connection.source.name), + source_name=cast(str, connection.source_name), destination_id=connection.destination_id, - destination_name=cast(str, connection.destination.name), + destination_name=cast(str, connection.destination_name), selected_streams=connection.stream_names, table_prefix=connection.table_prefix, ) diff --git a/tests/unit_tests/test_cloud_config_api_routing.py b/tests/unit_tests/test_cloud_config_api_routing.py new file mode 100644 index 000000000..8de558a12 --- /dev/null +++ b/tests/unit_tests/test_cloud_config_api_routing.py @@ -0,0 +1,210 @@ +# Copyright (c) 2024 Airbyte, Inc., all rights reserved. +"""Unit tests for routing Cloud reads through the Config API for bearer-only auth. + +Interactive OIDC logins produce a user-realm bearer token with no client credentials. +Such tokens are rejected by the public API but accepted by the Config API, so metadata +reads must route through the Config API in that case while continuing to use the public +API for application-client credentials. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest + +from airbyte._util import api_util +from airbyte.cloud import CloudWorkspace +from airbyte.cloud.connections import CloudConnection +from airbyte.cloud.models import CloudConnectionInfo, CloudWorkspaceInfo +from airbyte.secrets.base import SecretString + + +WORKSPACE_ID = "266ebdfe-0d7b-4540-9817-de7e4505ba61" +CONNECTION_ID = "991db3a4-c432-4aa9-8e10-31b74921d4b5" +SOURCE_ID = "11111111-1111-1111-1111-111111111111" +DESTINATION_ID = "22222222-2222-2222-2222-222222222222" + + +def _web_backend_connection_payload() -> dict[str, Any]: + """Return a minimal `WebBackendConnectionRead`-shaped Config API response.""" + return { + "connectionId": CONNECTION_ID, + "name": "My Postgres to Snowflake", + "sourceId": SOURCE_ID, + "destinationId": DESTINATION_ID, + "prefix": "raw_", + "status": "active", + "source": { + "sourceId": SOURCE_ID, + "workspaceId": WORKSPACE_ID, + "name": "My Postgres", + "sourceName": "Postgres", + }, + "destination": { + "destinationId": DESTINATION_ID, + "workspaceId": WORKSPACE_ID, + "name": "My Snowflake", + "destinationName": "Snowflake", + }, + "syncCatalog": { + "streams": [ + {"stream": {"name": "users"}, "config": {"selected": True}}, + {"stream": {"name": "orders"}, "config": {"selected": True}}, + ] + }, + } + + +@pytest.mark.parametrize( + "client_id,client_secret,bearer_token,expected", + [ + pytest.param(None, None, "token", True, id="bearer_only"), + pytest.param("id", "secret", None, False, id="client_credentials"), + ], +) +def test_uses_bearer_only_auth( + client_id: str | None, + client_secret: str | None, + bearer_token: str | None, + expected: bool, +) -> None: + workspace = CloudWorkspace( + workspace_id=WORKSPACE_ID, + client_id=client_id, + client_secret=client_secret, + bearer_token=bearer_token, + ) + assert workspace._uses_bearer_only_auth is expected + + +def test_workspace_info_from_config_api_mapping() -> None: + info = CloudWorkspaceInfo.from_mapping({ + "workspaceId": WORKSPACE_ID, + "name": "Acme Workspace", + "organizationId": "org-123", + "slug": "acme", + "initialSetupComplete": True, + }) + assert info.workspace_id == WORKSPACE_ID + assert info.name == "Acme Workspace" + assert info.organization_id == "org-123" + + +def test_connection_info_from_config_api_response() -> None: + info = CloudConnectionInfo.from_config_api_response( + _web_backend_connection_payload() + ) + + assert info.connection_id == CONNECTION_ID + assert info.workspace_id == WORKSPACE_ID + assert info.source_id == SOURCE_ID + assert info.destination_id == DESTINATION_ID + assert info.name == "My Postgres to Snowflake" + assert info.prefix == "raw_" + assert info.status == "active" + assert info.source_name == "My Postgres" + assert info.destination_name == "My Snowflake" + assert info.stream_names == ["users", "orders"] + + +def test_connection_info_from_config_api_response_uses_fallback_workspace_id() -> None: + payload = _web_backend_connection_payload() + payload["source"].pop("workspaceId") + payload["destination"].pop("workspaceId") + + info = CloudConnectionInfo.from_config_api_response( + payload, + fallback_workspace_id=WORKSPACE_ID, + ) + assert info.workspace_id == WORKSPACE_ID + + +def test_get_workspace_info_routes_bearer_only_to_config_api( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[str] = [] + + def _config_api(**_kwargs: Any) -> dict[str, Any]: + calls.append("config") + return {"workspaceId": WORKSPACE_ID, "name": "Acme"} + + def _public_api(**_kwargs: Any) -> object: + calls.append("public") + raise AssertionError("Public API should not be called for bearer-only auth.") + + monkeypatch.setattr(api_util, "get_workspace_via_config_api", _config_api) + monkeypatch.setattr(api_util, "get_workspace", _public_api) + + workspace = CloudWorkspace( + workspace_id=WORKSPACE_ID, + bearer_token=SecretString("token"), + ) + info = workspace.get_workspace_info() + + assert calls == ["config"] + assert info.workspace_id == WORKSPACE_ID + assert info.name == "Acme" + + +def test_get_workspace_info_routes_client_credentials_to_public_api( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[str] = [] + + def _config_api(**_kwargs: Any) -> dict[str, Any]: + calls.append("config") + raise AssertionError("Config API should not be called for client credentials.") + + def _public_api(**_kwargs: Any) -> object: + calls.append("public") + return SimpleNamespace( + workspace_id=WORKSPACE_ID, + name="Acme", + data_residency=None, + organization_id="org-123", + notifications=None, + ) + + monkeypatch.setattr(api_util, "get_workspace_via_config_api", _config_api) + monkeypatch.setattr(api_util, "get_workspace", _public_api) + + workspace = CloudWorkspace( + workspace_id=WORKSPACE_ID, + client_id=SecretString("id"), + client_secret=SecretString("secret"), + ) + info = workspace.get_workspace_info() + + assert calls == ["public"] + assert info.workspace_id == WORKSPACE_ID + + +def test_fetch_connection_info_routes_bearer_only_to_config_api( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[str] = [] + + def _config_api(**_kwargs: Any) -> dict[str, Any]: + calls.append("config") + return _web_backend_connection_payload() + + def _public_api(**_kwargs: Any) -> object: + calls.append("public") + raise AssertionError("Public API should not be called for bearer-only auth.") + + monkeypatch.setattr(api_util, "get_connection_via_config_api", _config_api) + monkeypatch.setattr(api_util, "get_connection", _public_api) + + workspace = CloudWorkspace( + workspace_id=WORKSPACE_ID, + bearer_token=SecretString("token"), + ) + connection = CloudConnection(workspace=workspace, connection_id=CONNECTION_ID) + + assert connection.source_name == "My Postgres" + assert connection.destination_name == "My Snowflake" + assert connection.stream_names == ["users", "orders"] + assert connection.table_prefix == "raw_" + assert calls == ["config"] From 8dd2ebf7b2588d3ee541803e3dfb449226fb211b Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 24 Jul 2026 19:04:12 +0000 Subject: [PATCH 2/2] fix(mcp): only include selected streams; route connect() via Config API Address review findings: - from_config_api_response now filters syncCatalog to selected streams (config.selected), matching public API stream_names behavior; a missing config defaults to selected. - CloudWorkspace.connect() now uses get_workspace_info() so bearer-only (interactive OIDC) credentials are validated via the Config API. Co-Authored-By: AJ Steers --- airbyte/cloud/models.py | 19 +++++++++++++++++-- airbyte/cloud/workspaces.py | 8 +------- .../test_cloud_config_api_routing.py | 8 ++++++-- 3 files changed, 24 insertions(+), 11 deletions(-) diff --git a/airbyte/cloud/models.py b/airbyte/cloud/models.py index 6f88a54b9..6ead38a9e 100644 --- a/airbyte/cloud/models.py +++ b/airbyte/cloud/models.py @@ -180,7 +180,9 @@ def from_config_api_response( The Config API response embeds the full `source` and `destination` objects and the `syncCatalog`, so connector names and selected streams are resolved from this - single response rather than via separate public API lookups. + single response rather than via separate public API lookups. The `syncCatalog` + lists every discovered stream with a per-stream `config.selected` flag, so only + selected streams are returned, matching the public API's `stream_names` behavior. """ source: Mapping[str, Any] = connection.get("source") or {} destination: Mapping[str, Any] = connection.get("destination") or {} @@ -189,7 +191,9 @@ def from_config_api_response( stream_names = [ entry["stream"]["name"] for entry in stream_entries - if isinstance(entry.get("stream"), Mapping) and entry["stream"].get("name") + if isinstance(entry.get("stream"), Mapping) + and entry["stream"].get("name") + and _is_stream_selected(entry.get("config")) ] workspace_id = ( source.get("workspaceId") @@ -315,6 +319,17 @@ def _notifications_to_dict(notifications: object) -> dict[str, object | None]: return {} +def _is_stream_selected(config: object) -> bool: + """Return whether a `syncCatalog` stream entry is selected for sync. + + A stream is selected when its `config.selected` flag is truthy. A missing `config` or + missing `selected` flag defaults to selected, matching the platform default. + """ + if not isinstance(config, Mapping): + return True + return bool(config.get("selected", True)) + + def _enum_value(value: object) -> str: """Return the string value for an enum-like object.""" if isinstance(value, Enum): diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 8dd9d0aac..77a4b6a0d 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -349,13 +349,7 @@ def connect(self) -> None: serves primarily as a simple check to ensure that the workspace is reachable and credentials are correct. """ - _ = api_util.get_workspace( - api_root=self.api_root, - workspace_id=self.workspace_id, - client_id=self.client_id, - client_secret=self.client_secret, - bearer_token=self.bearer_token, - ) + _ = self.get_workspace_info() print(f"Successfully connected to workspace: {self.workspace_url}") # Get sources, destinations, and connections diff --git a/tests/unit_tests/test_cloud_config_api_routing.py b/tests/unit_tests/test_cloud_config_api_routing.py index 8de558a12..e791049c9 100644 --- a/tests/unit_tests/test_cloud_config_api_routing.py +++ b/tests/unit_tests/test_cloud_config_api_routing.py @@ -52,6 +52,8 @@ def _web_backend_connection_payload() -> dict[str, Any]: "streams": [ {"stream": {"name": "users"}, "config": {"selected": True}}, {"stream": {"name": "orders"}, "config": {"selected": True}}, + {"stream": {"name": "audit_log"}, "config": {"selected": False}}, + {"stream": {"name": "legacy"}}, ] }, } @@ -106,7 +108,9 @@ def test_connection_info_from_config_api_response() -> None: assert info.status == "active" assert info.source_name == "My Postgres" assert info.destination_name == "My Snowflake" - assert info.stream_names == ["users", "orders"] + # `audit_log` (selected=False) is excluded; `legacy` (no config) defaults to selected. + assert info.stream_names == ["users", "orders", "legacy"] + assert "audit_log" not in info.stream_names def test_connection_info_from_config_api_response_uses_fallback_workspace_id() -> None: @@ -205,6 +209,6 @@ def _public_api(**_kwargs: Any) -> object: assert connection.source_name == "My Postgres" assert connection.destination_name == "My Snowflake" - assert connection.stream_names == ["users", "orders"] + assert connection.stream_names == ["users", "orders", "legacy"] assert connection.table_prefix == "raw_" assert calls == ["config"]