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