feat(mcp): Add HTTP header bearer token authentication support

airbyte/cloud/auth.py

@@ -2,6 +2,7 @@
 """Authentication-related constants and utilities for the Airbyte Cloud."""
 
 from airbyte import constants
+from airbyte.mcp._util import get_bearer_token_from_headers
 from airbyte.secrets import SecretString
 from airbyte.secrets.util import get_secret, try_get_secret
 
@@ -26,8 +27,23 @@ def resolve_cloud_bearer_token(
     input_value: str | SecretString | None = None,
     /,
 ) -> SecretString | None:
-    """Get the Airbyte Cloud bearer token from the environment."""
-    return try_get_secret(constants.CLOUD_BEARER_TOKEN_ENV_VAR, default=input_value)
+    """Get the Airbyte Cloud bearer token.
+
+    Resolution order:
+    1. Explicit input_value parameter
+    2. HTTP Authorization header (when running as MCP HTTP server)
+    3. AIRBYTE_CLOUD_BEARER_TOKEN environment variable
+    """
+    if input_value:
+        return SecretString(input_value)
+
+    # Try HTTP header first (for MCP HTTP transport)
+    header_token = get_bearer_token_from_headers()
+    if header_token:
+        return SecretString(header_token)
+
+    # Fall back to environment variable
+    return try_get_secret(constants.CLOUD_BEARER_TOKEN_ENV_VAR, default=None)
 
 
 def resolve_cloud_api_url(

airbyte/cloud/connectors.py

@@ -258,7 +258,7 @@ def _from_source_response(
             workspace=workspace,
             connector_id=source_response.source_id,
         )
-        result._connector_info = source_response
+        result._connector_info = source_response  # noqa: SLF001  # Non-public API
         return result
 
 
@@ -343,7 +343,7 @@ def _from_destination_response(
             workspace=workspace,
             connector_id=destination_response.destination_id,
         )
-        result._connector_info = destination_response
+        result._connector_info = destination_response  # noqa: SLF001  # Non-public API
         return result
 
 
@@ -647,7 +647,7 @@ def _from_yaml_response(
             definition_id=response.id,
             definition_type="yaml",
         )
-        result._definition_info = response
+        result._definition_info = response  # noqa: SLF001  # Non-public API
         return result
 
     def deploy_source(

airbyte/cloud/workspaces.py

@@ -107,7 +107,7 @@ class CloudOrganization:
     """Display name of the organization."""
 
 
-@dataclass
+@dataclass  # noqa: PLR0904  # Too many public methods
 class CloudWorkspace:
     """A remote workspace on the Airbyte Cloud.
 

airbyte/mcp/_util.py

@@ -8,6 +8,7 @@
 
 import dotenv
 import yaml
+from fastmcp.server.dependencies import get_http_headers
 
 from airbyte._util.meta import is_interactive
 from airbyte.secrets import (
@@ -24,6 +25,18 @@
 AIRBYTE_MCP_DOTENV_PATH_ENVVAR = "AIRBYTE_MCP_ENV_FILE"
 
 
+def get_bearer_token_from_headers() -> str | None:
+    """Extract bearer token from HTTP Authorization header.
+
+    Returns None if not running over HTTP transport or no header present.
+    """
+    headers = get_http_headers()
+    auth_header = headers.get("authorization", "")
+    if auth_header.startswith("Bearer "):
+        return auth_header[7:]  # Strip "Bearer " prefix
+    return None
+
+
 def _load_dotenv_file(dotenv_path: Path | str) -> None:
     """Load environment variables from a .env file."""
     if isinstance(dotenv_path, str):