fix(mcp)!: decouple external access, fix read-only mode, add X-MCP-Allow-Destructive-Tools, generic unsupported-input guard

airbyte/_util/unsupported_inputs.py

@@ -0,0 +1,43 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+"""Startup guards for removed PyAirbyte environment variables."""
+
+import os
+
+from airbyte.constants import CLOUD_MCP_SAFE_MODE_ENV_VAR
+from airbyte.exceptions import AirbyteLibInputError
+
+
+UNSUPPORTED_ENV_VARS: dict[str, str] = {
+    CLOUD_MCP_SAFE_MODE_ENV_VAR: (
+        "Remove `AIRBYTE_CLOUD_MCP_SAFE_MODE` and consider using one of these "
+        "alternatives:\n"
+        "- `AIRBYTE_CLOUD_MCP_ALLOW_PIPELINE_CHANGES=0`: block changes to "
+        "sources, destinations, and connections.\n"
+        "- `AIRBYTE_CLOUD_MCP_ALLOW_EXTERNAL_ACCESS=0`: block SQL and API "
+        "passthrough tools.\n"
+        "- `AIRBYTE_CLOUD_MCP_READONLY_MODE=1`: only expose read-only tools.\n"
+        "- `AIRBYTE_CLOUD_MCP_ALLOW_DESTRUCTIVE_TOOLS=0`: hide destructive tools.\n"
+        "- `AIRBYTE_CLOUD_MCP_LOCAL_DEMO_MODE=1`: only allow changes to "
+        "resources created in this session (local mode only; not recommended "
+        "except for demos).\n"
+    ),
+}
+"""Removed PyAirbyte environment variables, mapped to replacement guidance.
+
+Checked at startup of the `pyab` CLI and both MCP servers. Startup fails if any
+listed name is set.
+"""
+
+
+def check_unsupported_env_vars() -> None:
+    """Fail fast if a removed PyAirbyte env var is set.
+
+    Raises `AirbyteLibInputError` on the first name in `UNSUPPORTED_ENV_VARS`
+    present in the environment with any value, including an empty string.
+    """
+    for name, guidance in UNSUPPORTED_ENV_VARS.items():
+        if name in os.environ:
+            raise AirbyteLibInputError(
+                message=f"`{name}` is no longer supported.",
+                guidance=guidance,
+            )

airbyte/cli/pyab.py

@@ -67,16 +67,19 @@
 from typing import TYPE_CHECKING, Annotated, Any
 
 import yaml
-from cyclopts import App, Parameter
+from cyclopts import App, ArgumentCollection, Parameter
 
 from airbyte._util.destination_smoke_tests import run_destination_smoke_test
+from airbyte._util.unsupported_inputs import check_unsupported_env_vars
 from airbyte.destinations.util import get_destination, get_noop_destination
 from airbyte.exceptions import AirbyteLibInputError
 from airbyte.secrets.util import get_secret
 from airbyte.sources.util import get_benchmark_source, get_source
 
 
 if TYPE_CHECKING:
+    from collections.abc import Sequence
+
     from airbyte.destinations.base import Destination
     from airbyte.sources.base import Source
 
@@ -359,11 +362,21 @@ def _resolve_destination_job(
     )
 
 
+def _check_unsupported_env_vars_hook(
+    apps: Sequence[App],  # noqa: ARG001
+    commands: tuple[str, ...],  # noqa: ARG001
+    arguments: ArgumentCollection,  # noqa: ARG001
+) -> None:
+    """Fail any CLI command when a removed env var is set."""
+    check_unsupported_env_vars()
+
+
 cli = App(
     name="pyab",  # pyrefly: ignore[unexpected-keyword]
     help="PyAirbyte CLI.",  # pyrefly: ignore[unexpected-keyword]
     help_format="plaintext",  # pyrefly: ignore[unexpected-keyword]
     version_flags=[],  # pyrefly: ignore[unexpected-keyword]
+    config=[_check_unsupported_env_vars_hook],  # pyrefly: ignore[unexpected-keyword]
 )
 
 

airbyte/constants.py

@@ -306,6 +306,9 @@ def is_hosted_mcp_mode() -> bool:
 MCP_ALLOW_EXTERNAL_ACCESS_ENV_VAR: str = "AIRBYTE_CLOUD_MCP_ALLOW_EXTERNAL_ACCESS"
 """Environment variable controlling access to external-access Cloud tools."""
 
+MCP_ALLOW_DESTRUCTIVE_TOOLS_ENV_VAR: str = "AIRBYTE_CLOUD_MCP_ALLOW_DESTRUCTIVE_TOOLS"
+"""Environment variable controlling access to destructive Cloud tools."""
+
 MCP_LOCAL_DEMO_MODE_ENV_VAR: str = "AIRBYTE_CLOUD_MCP_LOCAL_DEMO_MODE"
 """Environment variable enabling opt-in local demo mode for STDIO MCP sessions."""
 
@@ -390,6 +393,9 @@ def is_hosted_mcp_mode() -> bool:
 MCP_CONFIG_ALLOW_EXTERNAL_ACCESS: str = "allow_external_access"
 """Config arg name for external-access permission."""
 
+MCP_CONFIG_ALLOW_DESTRUCTIVE_TOOLS: str = "allow_destructive_tools"
+"""Config arg name for the Cloud destructive-tools permission."""
+
 MCP_CONFIG_EXCLUDE_MODULES: str = "airbyte_exclude_modules"
 """Config arg name for the legacy AIRBYTE_MCP_DOMAINS_DISABLED setting."""
 
@@ -434,6 +440,9 @@ def is_hosted_mcp_mode() -> bool:
 MCP_ALLOW_EXTERNAL_ACCESS_HEADER: str = "X-MCP-Allow-External-Access"
 """HTTP header key for narrowing external-access permission."""
 
+MCP_ALLOW_DESTRUCTIVE_TOOLS_HEADER: str = "X-MCP-Allow-Destructive-Tools"
+"""HTTP header key for narrowing destructive-tools permission."""
+
 # Security Note: The API root and Config API root are intentionally NOT exposed as HTTP
 # headers. Each hosted MCP deployment is paired to a single backend, so allowing
 # a caller to override these URLs per-request would let them redirect the

airbyte/exceptions.py

@@ -339,25 +339,40 @@ class AirbyteLocalDemoModeError(AirbyteMCPError):
     """
 
 
+@dataclass
+class AirbyteReadOnlyModeError(AirbyteMCPError):
+    """A non-read-only tool was blocked by MCP read-only mode."""
+
+    guidance: str | None = "Unset `AIRBYTE_CLOUD_MCP_READONLY_MODE`."
+
+
 @dataclass
 class AirbytePipelineChangesDisabledError(AirbyteMCPError):
     """A pipeline-changing operation was blocked by MCP policy."""
 
     guidance: str | None = (
-        "Set `AIRBYTE_CLOUD_MCP_ALLOW_PIPELINE_CHANGES=1`, unset "
-        "`AIRBYTE_CLOUD_MCP_READONLY_MODE`, and ensure the "
+        "Set `AIRBYTE_CLOUD_MCP_ALLOW_PIPELINE_CHANGES=1` and ensure the "
         "`X-MCP-Allow-Pipeline-Changes` request header is not `0`."
     )
 
 
+@dataclass
+class AirbyteDestructiveToolsDisabledError(AirbyteMCPError):
+    """A destructive operation was blocked by MCP policy."""
+
+    guidance: str | None = (
+        "Set `AIRBYTE_CLOUD_MCP_ALLOW_DESTRUCTIVE_TOOLS=1` and ensure the "
+        "`X-MCP-Allow-Destructive-Tools` request header is not `0`."
+    )
+
+
 @dataclass
 class AirbyteExternalAccessDisabledError(AirbyteMCPError):
     """An external-access operation was blocked by MCP policy."""
 
     guidance: str | None = (
         "Set `AIRBYTE_CLOUD_MCP_ALLOW_EXTERNAL_ACCESS=1` and ensure the "
-        "`X-MCP-Allow-External-Access` request header is not `0`. Disabled pipeline changes "
-        "(including read-only mode) also disable external access when this setting is unset."
+        "`X-MCP-Allow-External-Access` request header is not `0`."
     )
 
 

airbyte/mcp/__init__.py

@@ -123,10 +123,7 @@
 `AIRBYTE_CLOUD_MCP_READONLY_MODE` environment variable.
 
 When enabled, only read-only Cloud tools are available. Write and destructive operations are
-disabled.
-
-This mode does allow running syncs on existing connectors, since sync operations
-are not considered to be modifications of the Airbyte Cloud workspace.
+disabled. Running and cancelling syncs are also disabled.
 
 Set the environment variable `AIRBYTE_CLOUD_MCP_READONLY_MODE=1` to enable read-only mode.
 
@@ -145,16 +142,18 @@
 `AIRBYTE_CLOUD_MCP_ALLOW_EXTERNAL_ACCESS` environment variable and
 `X-MCP-Allow-External-Access` request header control external-access tools, such as
 `execute_external_api_query`, which can run connector passthrough queries, search, and inspect
-external data or skill documentation.
+external data or skill documentation. The `AIRBYTE_CLOUD_MCP_ALLOW_DESTRUCTIVE_TOOLS`
+environment variable and `X-MCP-Allow-Destructive-Tools` request header control
+destructive-annotated tools.
 
 Each setting accepts `1`, `true`, or `yes` to allow, and `0`, `false`, or `no` to deny. An
 explicit environment denial cannot be overridden by a request header. An environment allow can
 be narrowed by a request header denial. If the environment setting is unset, the request header
 decides. `AIRBYTE_CLOUD_MCP_READONLY_MODE=1` also disables pipeline changes.
 
-When external access is unset, it is disabled if pipeline changes are disabled;
-otherwise external-access tools remain available. Explicitly allowing external access
-overrides these calculated defaults. Running or cancelling a sync is not a pipeline change.
+Read-only mode blocks both pipeline changes and running or cancelling syncs. However, running
+or cancelling a sync is _not_ considered a pipeline change, so those tools stay available when
+pipeline changes are disabled, unless read-only mode is also enabled.
 
 ## Authentication for Remote (HTTP) Servers
 

airbyte/mcp/_error_handling.py

@@ -35,13 +35,15 @@
     AirbyteConnectionSyncTimeoutError,
     AirbyteConnectorInUseError,
     AirbyteConnectorNotReadyError,
+    AirbyteDestructiveToolsDisabledError,
     AirbyteExternalAccessDisabledError,
     AirbyteLibError,
     AirbyteLibInputError,
     AirbyteLocalDemoModeError,
     AirbyteMCPError,
     AirbyteMissingResourceError,
     AirbytePipelineChangesDisabledError,
+    AirbyteReadOnlyModeError,
 )
 
 
@@ -81,6 +83,8 @@
     AirbyteMCPError,
     AirbyteLocalDemoModeError,
     AirbytePipelineChangesDisabledError,
+    AirbyteReadOnlyModeError,
+    AirbyteDestructiveToolsDisabledError,
     AirbyteExternalAccessDisabledError,
     AirbyteAgentsUnavailableError,
     AirbyteConnectorNotReadyError,

airbyte/mcp/_policy_middleware.py

@@ -3,16 +3,27 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from fastmcp.server.middleware import Middleware
+from fastmcp_extensions.tool_filters import (
+    ANNOTATION_DESTRUCTIVE_HINT,
+    ANNOTATION_READ_ONLY_HINT,
+    get_annotation,
+)
+from mcp.types import Tool
 
 from airbyte.exceptions import (
+    AirbyteDestructiveToolsDisabledError,
     AirbyteExternalAccessDisabledError,
     AirbytePipelineChangesDisabledError,
+    AirbyteReadOnlyModeError,
 )
 from airbyte.mcp._tool_utils import (
-    check_legacy_safe_mode_header,
+    _is_cloud_tool,
+    airbyte_readonly_mode_enabled,
+    check_unsupported_mcp_headers,
+    destructive_tools_allowed,
     external_access_allowed,
     get_tool_policy,
     pipeline_changes_allowed,
@@ -23,7 +34,8 @@
     from collections.abc import Sequence
 
     from fastmcp.server.middleware import CallNext, MiddlewareContext
-    from fastmcp.tools import Tool, ToolResult
+    from fastmcp.tools import Tool as FastMCPTool
+    from fastmcp.tools import ToolResult
     from mcp.types import CallToolRequestParams, ListToolsRequest
 
 
@@ -36,7 +48,7 @@ async def on_call_tool(
         call_next: CallNext[CallToolRequestParams, ToolResult],
     ) -> ToolResult:
         """Reject calls denied by effective pipeline or external-access settings."""
-        check_legacy_safe_mode_header()
+        check_unsupported_mcp_headers()
         fastmcp_context = context.fastmcp_context
         if fastmcp_context is None:
             return await call_next(context)
@@ -45,22 +57,43 @@ async def on_call_tool(
         if tool is None:
             return await call_next(context)
 
+        cloud_tool = _is_cloud_tool(fastmcp_context.fastmcp, cast(Tool, tool))
+        if (
+            cloud_tool
+            and airbyte_readonly_mode_enabled()
+            and not get_annotation(cast(Tool, tool), ANNOTATION_READ_ONLY_HINT, default=False)
+        ):
+            raise AirbyteReadOnlyModeError(
+                message="Read-only mode is enabled; only read-only tools are available."
+            )
         policy = get_tool_policy(tool)
-        if policy.pipeline_change and pipeline_changes_allowed(fastmcp_context) is False:
+        if (
+            cloud_tool
+            and policy.pipeline_change
+            and pipeline_changes_allowed(fastmcp_context) is False
+        ):
             raise AirbytePipelineChangesDisabledError(
                 message="Pipeline-changing tools are disabled by the MCP server policy."
             )
         if policy.external_access and external_access_allowed(fastmcp_context) is False:
             raise AirbyteExternalAccessDisabledError(
                 message="External-access tools are disabled by the MCP server policy."
             )
+        if (
+            cloud_tool
+            and get_annotation(cast(Tool, tool), ANNOTATION_DESTRUCTIVE_HINT, default=False)
+            and destructive_tools_allowed(fastmcp_context) is False
+        ):
+            raise AirbyteDestructiveToolsDisabledError(
+                message="Destructive tools are disabled by the MCP server policy."
+            )
         return await call_next(context)
 
     async def on_list_tools(
         self,
         context: MiddlewareContext[ListToolsRequest],
-        call_next: CallNext[ListToolsRequest, Sequence[Tool]],
-    ) -> Sequence[Tool]:
-        """Reject listings when the removed `X-MCP-Safe-Mode` header is present."""
-        check_legacy_safe_mode_header()
+        call_next: CallNext[ListToolsRequest, Sequence[FastMCPTool]],
+    ) -> Sequence[FastMCPTool]:
+        """Reject listings when a removed MCP request header is present."""
+        check_unsupported_mcp_headers()
         return await call_next(context)