airbyte_ops_mcp.mcp.cloud_auth

Shared Cloud API authentication resolution for MCP wrappers.

View Source

 1"""Shared Cloud API authentication resolution for MCP wrappers."""
 2
 3from fastmcp import Context
 4from fastmcp_extensions import get_mcp_config
 5
 6from airbyte_ops_mcp.cloud_admin.auth import CloudAuthError
 7from airbyte_ops_mcp.cloud_admin.version_overrides import ResolvedCloudAuth
 8from airbyte_ops_mcp.constants import ServerConfigKey
 9
10
11def resolve_cloud_auth(ctx: Context) -> ResolvedCloudAuth:
12    """Resolve Cloud API credentials from MCP transport configuration."""
13    bearer_token = get_mcp_config(ctx, ServerConfigKey.BEARER_TOKEN)
14    if bearer_token:
15        return ResolvedCloudAuth(bearer_token=bearer_token)
16
17    try:
18        return ResolvedCloudAuth(
19            client_id=get_mcp_config(ctx, ServerConfigKey.CLIENT_ID),
20            client_secret=get_mcp_config(ctx, ServerConfigKey.CLIENT_SECRET),
21        )
22    except ValueError as e:
23        raise CloudAuthError(
24            f"Failed to resolve credentials. Ensure credentials are provided via "
25            "Authorization header (Bearer token), HTTP headers "
26            "(X-Airbyte-Cloud-Client-Id, X-Airbyte-Cloud-Client-Secret), or "
27            f"environment variables. Error: {e}"
28        ) from e

def resolve_cloud_auth( ctx: fastmcp.server.context.Context) -> airbyte_ops_mcp.cloud_admin.version_overrides.ResolvedCloudAuth: View Source

12def resolve_cloud_auth(ctx: Context) -> ResolvedCloudAuth:
13    """Resolve Cloud API credentials from MCP transport configuration."""
14    bearer_token = get_mcp_config(ctx, ServerConfigKey.BEARER_TOKEN)
15    if bearer_token:
16        return ResolvedCloudAuth(bearer_token=bearer_token)
17
18    try:
19        return ResolvedCloudAuth(
20            client_id=get_mcp_config(ctx, ServerConfigKey.CLIENT_ID),
21            client_secret=get_mcp_config(ctx, ServerConfigKey.CLIENT_SECRET),
22        )
23    except ValueError as e:
24        raise CloudAuthError(
25            f"Failed to resolve credentials. Ensure credentials are provided via "
26            "Authorization header (Bearer token), HTTP headers "
27            "(X-Airbyte-Cloud-Client-Id, X-Airbyte-Cloud-Client-Secret), or "
28            f"environment variables. Error: {e}"
29        ) from e

Resolve Cloud API credentials from MCP transport configuration.