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