feat(cloud): Add CloudCredentials class and bearer token authentication support

airbyte/cloud/__init__.py

@@ -55,25 +55,28 @@
 
 from airbyte.cloud.connections import CloudConnection
 from airbyte.cloud.constants import JobStatusEnum
+from airbyte.cloud.credentials import CloudCredentials
 from airbyte.cloud.sync_results import SyncResult
 from airbyte.cloud.workspaces import CloudWorkspace
 
 
 # Submodules imported here for documentation reasons: https://github.com/mitmproxy/pdoc/issues/757
 if TYPE_CHECKING:
     # ruff: noqa: TC004
-    from airbyte.cloud import connections, constants, sync_results, workspaces
+    from airbyte.cloud import connections, constants, credentials, sync_results, workspaces
 
 
 __all__ = [
     # Submodules
     "workspaces",
     "connections",
     "constants",
+    "credentials",
     "sync_results",
     # Classes
     "CloudWorkspace",
     "CloudConnection",
+    "CloudCredentials",
     "SyncResult",
     # Enums
     "JobStatusEnum",

airbyte/cloud/auth.py

@@ -6,6 +6,31 @@
 from airbyte.secrets.util import get_secret, try_get_secret
 
 
+def resolve_cloud_bearer_token(
+    input_value: str | SecretString | None = None,
+    /,
+) -> SecretString | None:
+    """Get the Airbyte Cloud bearer token from the environment.
+
+    Unlike other resolve functions, this returns None if no bearer token is found,
+    since bearer token authentication is optional (client credentials can be used instead).
+
+    Args:
+        input_value: Optional explicit bearer token value. If provided, it will be
+            returned directly (wrapped in SecretString if needed).
+
+    Returns:
+        The bearer token as a SecretString, or None if not found.
+    """
+    if input_value is not None:
+        return SecretString(input_value)
+
+    result = try_get_secret(constants.CLOUD_BEARER_TOKEN_ENV_VAR, default=None)
+    if result:
+        return SecretString(result)
+    return None
+
+
 def resolve_cloud_client_secret(
     input_value: str | SecretString | None = None,
     /,

airbyte/cloud/connections.py

@@ -62,6 +62,7 @@ def _fetch_connection_info(self) -> ConnectionResponse:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
 
     @classmethod
@@ -180,6 +181,7 @@ def run_sync(
             workspace_id=self.workspace.workspace_id,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
         sync_result = SyncResult(
             workspace=self.workspace,
@@ -242,6 +244,7 @@ def get_previous_sync_logs(
             order_by=order_by,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
         return [
             SyncResult(
@@ -298,6 +301,7 @@ def get_state_artifacts(self) -> list[dict[str, Any]] | None:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
         if state_response.get("stateType") == "not_set":
             return None
@@ -319,6 +323,7 @@ def get_catalog_artifact(self) -> dict[str, Any] | None:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
         return connection_response.get("syncCatalog")
 
@@ -336,6 +341,7 @@ def rename(self, name: str) -> CloudConnection:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
             name=name,
         )
         self._connection_info = updated_response
@@ -355,6 +361,7 @@ def set_table_prefix(self, prefix: str) -> CloudConnection:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
             prefix=prefix,
         )
         self._connection_info = updated_response
@@ -379,6 +386,7 @@ def set_selected_streams(self, stream_names: list[str]) -> CloudConnection:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
             configurations=configurations,
         )
         self._connection_info = updated_response

airbyte/cloud/connectors.py

@@ -165,6 +165,7 @@ def check(
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
         check_result = CheckResult(
             success=result[0],
@@ -197,6 +198,7 @@ def _fetch_connector_info(self) -> api_models.SourceResponse:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
 
     def rename(self, name: str) -> CloudSource:
@@ -213,6 +215,7 @@ def rename(self, name: str) -> CloudSource:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
             name=name,
         )
         self._connector_info = updated_response
@@ -235,6 +238,7 @@ def update_config(self, config: dict[str, Any]) -> CloudSource:
             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=config,
         )
         self._connector_info = updated_response
@@ -279,6 +283,7 @@ def _fetch_connector_info(self) -> api_models.DestinationResponse:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
 
     def rename(self, name: str) -> CloudDestination:
@@ -295,6 +300,7 @@ def rename(self, name: str) -> CloudDestination:
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
             name=name,
         )
         self._connector_info = updated_response
@@ -317,6 +323,7 @@ def update_config(self, config: dict[str, Any]) -> CloudDestination:
             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=config,
         )
         self._connector_info = updated_response
@@ -377,6 +384,7 @@ def _fetch_definition_info(
                 api_root=self.workspace.api_root,
                 client_id=self.workspace.client_id,
                 client_secret=self.workspace.client_secret,
+                bearer_token=self.workspace.bearer_token,
             )
         raise NotImplementedError(
             "Docker custom source definitions are not yet supported. "
@@ -463,6 +471,7 @@ def connector_builder_project_id(self) -> str | None:
                 api_root=self.workspace.api_root,
                 client_id=self.workspace.client_id,
                 client_secret=self.workspace.client_secret,
+                bearer_token=self.workspace.bearer_token,
             )
         )
 
@@ -510,6 +519,7 @@ def permanently_delete(
                 api_root=self.workspace.api_root,
                 client_id=self.workspace.client_id,
                 client_secret=self.workspace.client_secret,
+                bearer_token=self.workspace.bearer_token,
                 safe_mode=safe_mode,
             )
         else:
@@ -579,6 +589,7 @@ def update_definition(
                 api_root=self.workspace.api_root,
                 client_id=self.workspace.client_id,
                 client_secret=self.workspace.client_secret,
+                bearer_token=self.workspace.bearer_token,
             )
             return CustomCloudSourceDefinition._from_yaml_response(self.workspace, result)
 
@@ -685,6 +696,7 @@ def deploy_source(
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
         return CloudSource._from_source_response(  # noqa: SLF001  # Accessing Non-Public API
             workspace=self.workspace,
@@ -754,6 +766,7 @@ def set_testing_values(
             api_root=self.workspace.api_root,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
         )
 
         return self