airbyte_ops_mcp.constants

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""Constants for the Airbyte Admin MCP server."""
  3
  4from __future__ import annotations
  5
  6from enum import Enum, StrEnum
  7
  8from airbyte.exceptions import PyAirbyteInputError
  9
 10from airbyte_ops_mcp.registry._constants import PROD_METADATA_SERVICE_BUCKET_NAME
 11
 12MCP_SERVER_NAME = "airbyte-internal-ops"
 13"""The name of the MCP server."""
 14
 15PACKAGE_NAME = "airbyte-internal-ops"
 16"""The installed package name, as declared in `pyproject.toml`."""
 17
 18
 19class ServerConfigKey(StrEnum):
 20    """Config keys for MCP server configuration arguments.
 21
 22    These keys are used both when defining server_config_args in mcp_server()
 23    and when retrieving config values via get_mcp_config().
 24    """
 25
 26    BEARER_TOKEN = "bearer_token"
 27    CLIENT_ID = "client_id"
 28    CLIENT_SECRET = "client_secret"
 29
 30
 31USER_AGENT = "Airbyte-Internal-Ops Python client"
 32"""User-Agent string for HTTP requests to Airbyte Cloud APIs."""
 33
 34# Environment variable names for internal admin authentication
 35ENV_AIRBYTE_INTERNAL_ADMIN_FLAG = "AIRBYTE_INTERNAL_ADMIN_FLAG"
 36ENV_AIRBYTE_INTERNAL_ADMIN_USER = "AIRBYTE_INTERNAL_ADMIN_USER"
 37
 38# Environment variable for GCP credentials (JSON content, not file path)
 39ENV_GCP_PROD_DB_ACCESS_CREDENTIALS = "GCP_PROD_DB_ACCESS_CREDENTIALS"
 40"""Environment variable containing GCP service account JSON credentials for prod DB access."""
 41
 42ENV_K_SERVICE = "K_SERVICE"
 43"""Cloud Run env var set to the service name, used only for runtime detection."""
 44
 45# Expected values for internal admin authentication
 46EXPECTED_ADMIN_FLAG_VALUE = "airbyte.io"
 47EXPECTED_ADMIN_EMAIL_DOMAIN = "@airbyte.io"
 48
 49# =============================================================================
 50# HTTP Header Names for Airbyte Cloud Authentication
 51# =============================================================================
 52# These headers follow the PyAirbyte convention for passing credentials
 53# via HTTP when running as an MCP HTTP server.
 54
 55HEADER_AIRBYTE_CLOUD_CLIENT_ID = "X-Airbyte-Cloud-Client-Id"
 56"""HTTP header for OAuth client ID."""
 57
 58HEADER_AIRBYTE_CLOUD_CLIENT_SECRET = "X-Airbyte-Cloud-Client-Secret"
 59"""HTTP header for OAuth client secret."""
 60
 61HEADER_AIRBYTE_CLOUD_WORKSPACE_ID = "X-Airbyte-Cloud-Workspace-Id"
 62"""HTTP header for default workspace ID."""
 63
 64HEADER_AIRBYTE_CLOUD_API_URL = "X-Airbyte-Cloud-Api-Url"
 65"""HTTP header for API root URL override."""
 66
 67# =============================================================================
 68# GCP and Prod DB Constants (from connection-retriever)
 69# =============================================================================
 70
 71GCP_PROJECT_NAME = "prod-ab-cloud-proj"
 72"""The GCP project name for Airbyte Cloud production."""
 73
 74CLOUD_SQL_INSTANCE = "prod-ab-cloud-proj:us-west3:prod-pgsql-replica"
 75"""The Cloud SQL instance connection name for the Prod DB Replica."""
 76
 77DEFAULT_CLOUD_SQL_PROXY_PORT = 15432
 78"""Default port for Cloud SQL Proxy connections."""
 79
 80CLOUD_SQL_PROXY_PID_FILE = "/tmp/airbyte-cloud-sql-proxy.pid"
 81"""PID file for tracking the Cloud SQL Proxy process."""
 82
 83CLOUD_REGISTRY_URL = (
 84    "https://connectors.airbyte.com/files/registries/v0/cloud_registry.json"
 85)
 86"""URL for the Airbyte Cloud connector registry."""
 87
 88CLOUD_REGISTRY_GCS_URL = (
 89    "https://storage.googleapis.com/"
 90    f"{PROD_METADATA_SERVICE_BUCKET_NAME}/registries/v0/cloud_registry.json"
 91)
 92"""Direct GCS URL for the Airbyte Cloud connector registry."""
 93
 94# =============================================================================
 95# Organization ID Aliases
 96# =============================================================================
 97
 98
 99class OrganizationAliasEnum(StrEnum):
100    """Organization ID aliases that can be used in place of UUIDs.
101
102    Each member's name is the alias (e.g., "@airbyte-internal") and its value
103    is the actual organization UUID. Use `OrganizationAliasEnum.resolve()` to
104    resolve aliases to actual IDs.
105    """
106
107    AIRBYTE_INTERNAL = "664c690e-5263-49ba-b01f-4a6759b3330a"
108    """The Airbyte internal organization for testing and internal operations.
109
110    Alias: @airbyte-internal
111    """
112
113    @classmethod
114    def resolve(cls, org_id: str | None) -> str | None:
115        """Resolve an organization ID alias to its actual UUID.
116
117        Accepts either an alias string (e.g., "@airbyte-internal") or an
118        OrganizationAliasEnum enum member, and returns the actual UUID.
119
120        Returns:
121            The resolved organization ID (UUID), or None if input is None.
122            If the input doesn't start with "@", it is returned unchanged.
123
124        Raises:
125            PyAirbyteInputError: If the input starts with "@" but is not a recognized alias.
126        """
127        if org_id is None:
128            return None
129
130        # Handle OrganizationAliasEnum enum members directly
131        if isinstance(org_id, cls):
132            return org_id.value
133
134        # If it doesn't look like an alias, return as-is (assume it's a UUID)
135        if not org_id.startswith("@"):
136            return org_id
137
138        # Handle alias strings or raise an error if invalid
139        alias_mapping = {
140            "@airbyte-internal": cls.AIRBYTE_INTERNAL.value,
141        }
142        if org_id not in alias_mapping:
143            raise PyAirbyteInputError(
144                message=f"Unknown organization alias: {org_id}",
145                context={
146                    "valid_aliases": list(alias_mapping.keys()),
147                },
148            )
149        return alias_mapping[org_id]
150
151
152# =============================================================================
153# Workspace ID Aliases
154# =============================================================================
155
156
157class WorkspaceAliasEnum(StrEnum):
158    """Workspace ID aliases that can be used in place of UUIDs.
159
160    Each member's name is the alias (e.g., "@devin-ai-sandbox") and its value
161    is the actual workspace UUID. Use `WorkspaceAliasEnum.resolve()` to
162    resolve aliases to actual IDs.
163    """
164
165    DEVIN_AI_SANDBOX = "266ebdfe-0d7b-4540-9817-de7e4505ba61"
166    """The Devin AI sandbox workspace for testing and development.
167
168    Alias: @devin-ai-sandbox
169    """
170
171    @classmethod
172    def resolve(cls, workspace_id: str | None) -> str | None:
173        """Resolve a workspace ID alias to its actual UUID.
174
175        Accepts either an alias string (e.g., "@devin-ai-sandbox") or a
176        WorkspaceAliasEnum enum member, and returns the actual UUID.
177
178        Returns:
179            The resolved workspace ID (UUID), or None if input is None.
180            If the input doesn't start with "@", it is returned unchanged.
181
182        Raises:
183            PyAirbyteInputError: If the input starts with "@" but is not a recognized alias.
184        """
185        if workspace_id is None:
186            return None
187
188        # Handle WorkspaceAliasEnum enum members directly
189        if isinstance(workspace_id, cls):
190            return workspace_id.value
191
192        # If it doesn't look like an alias, return as-is (assume it's a UUID)
193        if not workspace_id.startswith("@"):
194            return workspace_id
195
196        # Handle alias strings or raise an error if invalid
197        alias_mapping = {
198            "@devin-ai-sandbox": cls.DEVIN_AI_SANDBOX.value,
199        }
200        if workspace_id not in alias_mapping:
201            raise PyAirbyteInputError(
202                message=f"Unknown workspace alias: {workspace_id}",
203                context={
204                    "valid_aliases": list(alias_mapping.keys()),
205                },
206            )
207        return alias_mapping[workspace_id]
208
209
210MEDIC_MODE_ENV_VAR = "AIRBYTE_OPS_MEDIC_MODE"
211"""Environment variable to enable emergency 'medic mode' write tools.
212
213When set to '1' or 'true', destructive MCP tools for connection state and catalog
214writes are registered. These are "break glass" operations for emergency use only.
215"""
216
217CONNECTION_RETRIEVER_PG_CONNECTION_DETAILS_SECRET_ID = (
218    "projects/587336813068/secrets/CONNECTION_RETRIEVER_PG_CONNECTION_DETAILS"
219)
220"""GCP Secret Manager ID for Prod DB connection details."""
221
222
223class ConnectionObject(Enum):
224    """Types of connection objects that can be retrieved."""
225
226    CONNECTION = "connection"
227    SOURCE_ID = "source-id"
228    DESTINATION_ID = "destination-id"
229    DESTINATION_CONFIG = "destination-config"
230    SOURCE_CONFIG = "source-config"
231    CATALOG = "catalog"
232    CONFIGURED_CATALOG = "configured-catalog"
233    STATE = "state"
234    WORKSPACE_ID = "workspace-id"
235    DESTINATION_DOCKER_IMAGE = "destination-docker-image"
236    SOURCE_DOCKER_IMAGE = "source-docker-image"