feat(mcp): Add AIRBYTE_CLOUD_CONFIG_API_URL environment variable

airbyte/_util/api_util.py

@@ -21,7 +21,8 @@
 import requests
 from airbyte_api import api, models
 
-from airbyte.constants import CLOUD_API_ROOT, CLOUD_CONFIG_API_ROOT
+from airbyte.cloud.auth import resolve_cloud_config_api_url
+from airbyte.constants import CLOUD_API_ROOT, CLOUD_CONFIG_API_ROOT, CLOUD_CONFIG_API_ROOT_ENV_VAR
 from airbyte.exceptions import (
     AirbyteConnectionSyncError,
     AirbyteError,
@@ -58,11 +59,36 @@ def status_ok(status_code: int) -> bool:
 
 
 def get_config_api_root(api_root: str) -> str:
-    """Get the configuration API root from the main API root."""
+    """Get the configuration API root from the main API root.
+
+    Resolution order:
+    1. If `AIRBYTE_CLOUD_CONFIG_API_URL` environment variable is set, use that value.
+    2. If `api_root` matches the default Cloud API root, return the default Config API root.
+    3. Otherwise, raise NotImplementedError (cannot derive Config API from custom API root).
+
+    Args:
+        api_root: The main API root URL being used.
+
+    Returns:
+        The configuration API root URL.
+
+    Raises:
+        NotImplementedError: If the Config API root cannot be determined.
+    """
+    # First, check if the Config API URL is explicitly set via environment variable
+    config_api_override = resolve_cloud_config_api_url()
+    if config_api_override:
+        return config_api_override
+
+    # Fall back to deriving from the main API root
     if api_root == CLOUD_API_ROOT:
         return CLOUD_CONFIG_API_ROOT
 
-    raise NotImplementedError("Configuration API root not implemented for this API root.")
+    raise NotImplementedError(
+        f"Configuration API root not implemented for api_root='{api_root}'. "
+        f"Set the '{CLOUD_CONFIG_API_ROOT_ENV_VAR}' environment variable "
+        "to specify the Config API URL."
+    )
 
 
 def get_web_url_root(api_root: str) -> str:
@@ -171,7 +197,8 @@ def get_workspace(
         resource_type="workspace",
         context={
             "workspace_id": workspace_id,
-            "response": response,
+            "api_root": api_root,
+            "status_code": response.status_code,
         },
     )
 
@@ -419,6 +446,10 @@ def get_connection(
         resource_name_or_id=connection_id,
         resource_type="connection",
         log_text=response.raw_response.text,
+        context={
+            "api_root": api_root,
+            "status_code": response.status_code,
+        },
     )
 
 
@@ -517,6 +548,8 @@ def get_job_logs(  # noqa: PLR0913  # Too many arguments - needed for auth flexi
         context={
             "workspace_id": workspace_id,
             "connection_id": connection_id,
+            "api_root": api_root,
+            "status_code": response.status_code,
         },
     )
 
@@ -548,6 +581,10 @@ def get_job_info(
         resource_name_or_id=str(job_id),
         resource_type="job",
         log_text=response.raw_response.text,
+        context={
+            "api_root": api_root,
+            "status_code": response.status_code,
+        },
     )
 
 
@@ -620,6 +657,10 @@ def get_source(
         resource_name_or_id=source_id,
         resource_type="source",
         log_text=response.raw_response.text,
+        context={
+            "api_root": api_root,
+            "status_code": response.status_code,
+        },
     )
 
 
@@ -863,6 +904,10 @@ def get_destination(
         resource_name_or_id=destination_id,
         resource_type="destination",
         log_text=response.raw_response.text,
+        context={
+            "api_root": api_root,
+            "status_code": response.status_code,
+        },
     )
 
 

airbyte/cloud/auth.py

@@ -64,3 +64,21 @@ def resolve_cloud_workspace_id(
 ) -> str:
     """Get the Airbyte Cloud workspace ID from the environment, or return None if not set."""
     return str(get_secret(constants.CLOUD_WORKSPACE_ID_ENV_VAR, default=input_value))
+
+
+def resolve_cloud_config_api_url(
+    input_value: str | None = None,
+    /,
+) -> str | None:
+    """Get the Airbyte Cloud Config API URL from the environment, or return None if not set.
+
+    The Config API is a separate internal API used for certain operations like
+    connector builder projects and custom source definitions.
+
+    Returns:
+        The Config API URL if set via environment variable or input, None otherwise.
+    """
+    result = try_get_secret(constants.CLOUD_CONFIG_API_ROOT_ENV_VAR, default=input_value)
+    if result:
+        return str(result)
+    return None

airbyte/constants.py

@@ -219,6 +219,16 @@ def _str_to_bool(value: str) -> bool:
 CLOUD_API_ROOT_ENV_VAR: str = "AIRBYTE_CLOUD_API_URL"
 """The environment variable name for the Airbyte Cloud API URL."""
 
+CLOUD_CONFIG_API_ROOT_ENV_VAR: str = "AIRBYTE_CLOUD_CONFIG_API_URL"
+"""The environment variable name for the Airbyte Cloud Config API URL.
+
+The Config API is a separate internal API used for certain operations like
+connector builder projects and custom source definitions. This environment
+variable allows overriding the default Config API URL, which is useful when
+the public API URL has been overridden and the Config API cannot be derived
+from it automatically.
+"""
+
 CLOUD_WORKSPACE_ID_ENV_VAR: str = "AIRBYTE_CLOUD_WORKSPACE_ID"
 """The environment variable name for the Airbyte Cloud workspace ID."""