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