feat: Add safety_mode to custom connector definition deletion (#831)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: aaronsteers

airbyte/_util/api_util.py

@@ -1133,8 +1133,58 @@ def delete_custom_yaml_source_definition(
     api_root: str,
     client_id: SecretString,
     client_secret: SecretString,
+    safe_mode: bool = True,
 ) -> None:
-    """Delete a custom YAML source definition."""
+    """Delete a custom YAML source definition.
+
+    Args:
+        workspace_id: The workspace ID
+        definition_id: The definition ID to delete
+        api_root: The API root URL
+        client_id: OAuth client ID
+        client_secret: OAuth client secret
+        safe_mode: If True, requires the connector name to either start with "delete:"
+            or contain "delete-me" (case insensitive) to prevent accidental deletion.
+            Defaults to True.
+
+    Raises:
+        PyAirbyteInputError: If safe_mode is True and the connector name does not meet
+            the safety requirements.
+    """
+    if safe_mode:
+        definition_info = get_custom_yaml_source_definition(
+            workspace_id=workspace_id,
+            definition_id=definition_id,
+            api_root=api_root,
+            client_id=client_id,
+            client_secret=client_secret,
+        )
+        connector_name = definition_info.name
+
+        def is_safe_to_delete(name: str) -> bool:
+            name_lower = name.lower()
+            return name_lower.startswith("delete:") or "delete-me" in name_lower
+
+        if not is_safe_to_delete(definition_info.name):
+            raise PyAirbyteInputError(
+                message=(
+                    f"Cannot delete custom connector definition '{connector_name}' "
+                    "with safe_mode enabled. "
+                    "To authorize deletion, the connector name must either:\n"
+                    "  1. Start with 'delete:' (case insensitive), OR\n"
+                    "  2. Contain 'delete-me' (case insensitive)\n\n"
+                    "Please rename the connector to meet one of these requirements "
+                    "before attempting deletion."
+                ),
+                context={
+                    "definition_id": definition_id,
+                    "connector_name": connector_name,
+                    "safe_mode": True,
+                },
+            )
+
+    # Else proceed with deletion
+
     airbyte_instance = get_airbyte_server_instance(
         api_root=api_root,
         client_id=client_id,

airbyte/cloud/connectors.py

@@ -410,12 +410,32 @@ def definition_url(self) -> str:
             or f"{self.workspace.workspace_url}/settings/{self.connector_type}"
         )
 
-    def permanently_delete(self) -> None:
-        """Permanently delete this custom source definition."""
-        self.workspace.permanently_delete_custom_source_definition(
-            self.definition_id,
-            definition_type=self.definition_type,
-        )
+    def permanently_delete(
+        self,
+        *,
+        safe_mode: bool = True,
+    ) -> None:
+        """Permanently delete this custom source definition.
+
+        Args:
+            safe_mode: If True, requires the connector name to either start with "delete:"
+                or contain "delete-me" (case insensitive) to prevent accidental deletion.
+                Defaults to True.
+        """
+        if self.definition_type == "yaml":
+            api_util.delete_custom_yaml_source_definition(
+                workspace_id=self.workspace.workspace_id,
+                definition_id=self.definition_id,
+                api_root=self.workspace.api_root,
+                client_id=self.workspace.client_id,
+                client_secret=self.workspace.client_secret,
+                safe_mode=safe_mode,
+            )
+        else:
+            raise NotImplementedError(
+                "Docker custom source definitions are not yet supported. "
+                "Only YAML manifest-based custom sources are currently available."
+            )
 
     def update_definition(
         self,

airbyte/cloud/workspaces.py

@@ -610,29 +610,3 @@ def get_custom_source_definition(
             "Docker custom source definitions are not yet supported. "
             "Only YAML manifest-based custom sources are currently available."
         )
-
-    def permanently_delete_custom_source_definition(
-        self,
-        definition_id: str,
-        *,
-        definition_type: Literal["yaml", "docker"],
-    ) -> None:
-        """Permanently delete a custom source definition.
-
-        Args:
-            definition_id: The definition ID to delete
-            definition_type: Connector type ("yaml" or "docker"). Required.
-        """
-        if definition_type == "yaml":
-            api_util.delete_custom_yaml_source_definition(
-                workspace_id=self.workspace_id,
-                definition_id=definition_id,
-                api_root=self.api_root,
-                client_id=self.client_id,
-                client_secret=self.client_secret,
-            )
-        else:
-            raise NotImplementedError(
-                "Docker custom source definitions are not yet supported. "
-                "Only YAML manifest-based custom sources are currently available."
-            )

airbyte/mcp/cloud_ops.py

@@ -591,9 +591,8 @@ def list_custom_source_definitions() -> list[dict[str, Any]]:
         {
             "definition_id": d.definition_id,
             "name": d.name,
-            "connector_type": d.connector_type,
-            "manifest": d.manifest,
             "version": d.version,
+            "connector_builder_project_url": d.connector_builder_project_url,
         }
         for d in definitions
     ]
@@ -649,6 +648,39 @@ def update_custom_source_definition(
         )
 
 
+def permanently_delete_custom_source_definition(
+    definition_id: Annotated[
+        str,
+        Field(description="The ID of the custom source definition to delete."),
+    ],
+) -> str:
+    """Permanently delete a custom YAML source definition from Airbyte Cloud.
+
+    IMPORTANT: This operation requires the connector name to either:
+    1. Start with "delete:" (case insensitive), OR
+    2. Contain "delete-me" (case insensitive)
+
+    If the connector does not meet these requirements, the deletion will be rejected with a
+    helpful error message. Instruct the user to rename the connector appropriately to authorize
+    the deletion.
+
+    Note: Only YAML (declarative) connectors are currently supported.
+    Docker-based custom sources are not yet available.
+    """
+    workspace: CloudWorkspace = _get_cloud_workspace()
+    definition = workspace.get_custom_source_definition(
+        definition_id=definition_id,
+        definition_type="yaml",
+    )
+    definition_name: str = definition.name  # Capture name before deletion
+    definition.permanently_delete(
+        safe_mode=True,  # Hard-coded safe mode for extra protection when running in LLM agents.
+    )
+    return (
+        f"Successfully deleted custom source definition '{definition_name}' (ID: {definition_id})"
+    )
+
+
 def register_cloud_ops_tools(app: FastMCP) -> None:
     """@private Register tools with the FastMCP app.
 
@@ -668,3 +700,4 @@ def register_cloud_ops_tools(app: FastMCP) -> None:
     app.tool(publish_custom_source_definition)
     app.tool(list_custom_source_definitions)
     app.tool(update_custom_source_definition)
+    app.tool(permanently_delete_custom_source_definition)

tests/integration_tests/cloud/test_custom_definitions.py

@@ -93,10 +93,11 @@ def test_publish_custom_yaml_source(
         assert updated.manifest["version"] == "0.2.0"
 
     finally:
-        cloud_workspace.permanently_delete_custom_source_definition(
+        fetched = cloud_workspace.get_custom_source_definition(
             definition_id,
             definition_type="yaml",
         )
+        fetched.permanently_delete(safe_mode=False)
 
 
 @pytest.mark.requires_creds
@@ -118,3 +119,52 @@ def test_yaml_validation_error(
         )
 
     assert "type" in str(exc_info.value).lower()
+
+
+@pytest.mark.requires_creds
+@pytest.mark.parametrize(
+    "name_template,expect_allow",
+    [
+        ("test-yaml-source-{suffix}", False),
+        ("delete:test-yaml-source-{suffix}", True),
+        ("test-delete-me-yaml-source-{suffix}", True),
+    ],
+)
+def test_safe_mode_deletion(
+    cloud_workspace: CloudWorkspace,
+    name_template: str,
+    expect_allow: bool,
+) -> None:
+    """Test safe_mode deletion behavior with different connector names."""
+    from airbyte._util import text_util
+    from airbyte.exceptions import PyAirbyteInputError
+
+    name = name_template.format(suffix=text_util.generate_random_suffix())
+
+    result = cloud_workspace.publish_custom_source_definition(
+        name=name,
+        manifest_yaml=TEST_YAML_MANIFEST,
+        unique=True,
+        pre_validate=True,
+    )
+
+    definition_id = result.definition_id
+
+    definition = cloud_workspace.get_custom_source_definition(
+        definition_id,
+        definition_type="yaml",
+    )
+
+    if expect_allow:
+        definition.permanently_delete(safe_mode=True)
+    else:
+        try:
+            with pytest.raises(PyAirbyteInputError) as exc_info:
+                definition.permanently_delete(safe_mode=True)
+
+            error_message = str(exc_info.value).lower()
+            assert "safe_mode" in error_message
+            assert "delete:" in error_message or "delete-me" in error_message
+
+        finally:
+            definition.permanently_delete(safe_mode=False)