feat(mcp): add ALLOW_EXTERNAL_ACCESS and ALLOW_PIPELINE_CHANGES policies for Cloud MCP

airbyte/constants.py

@@ -298,10 +298,16 @@ def is_hosted_mcp_mode() -> bool:
 
 
 MCP_READONLY_MODE_ENV_VAR: str = "AIRBYTE_CLOUD_MCP_READONLY_MODE"
-"""Environment variable to enable read-only mode for the MCP server.
+"""Legacy environment variable that disables pipeline-changing MCP tools when truthy."""
 
-When set to "1" or "true", only tools with readOnlyHint=True will be available.
-"""
+MCP_ALLOW_PIPELINE_CHANGES_ENV_VAR: str = "AIRBYTE_CLOUD_MCP_ALLOW_PIPELINE_CHANGES"
+"""Environment variable controlling access to Cloud pipeline-changing tools."""
+
+MCP_ALLOW_EXTERNAL_ACCESS_ENV_VAR: str = "AIRBYTE_CLOUD_MCP_ALLOW_EXTERNAL_ACCESS"
+"""Environment variable controlling access to external-access Cloud tools."""
+
+CLOUD_MCP_SAFE_MODE_ENV_VAR: str = "AIRBYTE_CLOUD_MCP_SAFE_MODE"
+"""Environment variable controlling safe mode for destructive Cloud operations."""
 
 MCP_DOMAINS_DISABLED_ENV_VAR: str = "AIRBYTE_MCP_DOMAINS_DISABLED"
 """Environment variable to disable specific MCP tool domains.
@@ -371,6 +377,12 @@ def is_hosted_mcp_mode() -> bool:
 MCP_CONFIG_READONLY_MODE: str = "airbyte_readonly_mode"
 """Config arg name for the legacy AIRBYTE_CLOUD_MCP_READONLY_MODE setting."""
 
+MCP_CONFIG_ALLOW_PIPELINE_CHANGES: str = "allow_pipeline_changes"
+"""Config arg name for the Cloud pipeline-change permission."""
+
+MCP_CONFIG_ALLOW_EXTERNAL_ACCESS: str = "allow_external_access"
+"""Config arg name for external-access permission."""
+
 MCP_CONFIG_EXCLUDE_MODULES: str = "airbyte_exclude_modules"
 """Config arg name for the legacy AIRBYTE_MCP_DOMAINS_DISABLED setting."""
 
@@ -409,6 +421,12 @@ def is_hosted_mcp_mode() -> bool:
 MCP_EXTENSIONS_HEADER: str = "X-MCP-Extensions"
 """HTTP header key for client-declared MCP extension IDs."""
 
+MCP_ALLOW_PIPELINE_CHANGES_HEADER: str = "X-MCP-Allow-Pipeline-Changes"
+"""HTTP header key for narrowing Cloud pipeline-change permission."""
+
+MCP_ALLOW_EXTERNAL_ACCESS_HEADER: str = "X-MCP-Allow-External-Access"
+"""HTTP header key for narrowing external-access 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,6 +339,29 @@ class AirbyteSafeModeError(AirbyteMCPError):
     """
 
 
+@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 legacy "
+        "`AIRBYTE_CLOUD_MCP_READONLY_MODE`, and ensure the "
+        "`X-MCP-Allow-Pipeline-Changes` 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 legacy read-only mode) and an explicit `AIRBYTE_CLOUD_MCP_SAFE_MODE=1` "
+        "also disable external access when this setting is unset."
+    )
+
+
 # Normalization Errors
 
 

airbyte/mcp/__init__.py

@@ -130,6 +130,10 @@
 and are only allowed for objects created in the current session.
 
 Set the environment variable `AIRBYTE_CLOUD_MCP_SAFE_MODE=0` to disable safe mode.
+When `AIRBYTE_CLOUD_MCP_SAFE_MODE` is explicitly set to a truthy value, external-access tools
+(for example `execute_external_api_query`) are also disabled unless
+`AIRBYTE_CLOUD_MCP_ALLOW_EXTERNAL_ACCESS=1` explicitly allows them. Safe mode's default-on
+behavior does not by itself disable external access.
 
 ### Airbyte Cloud Read-Only Mode
 
@@ -144,6 +148,26 @@
 
 Set the environment variable `AIRBYTE_CLOUD_MCP_READONLY_MODE=1` to enable read-only mode.
 
+## Airbyte Cloud Pipeline and External Access Permissions
+
+Pipeline changes and external access can be controlled independently. The
+`AIRBYTE_CLOUD_MCP_ALLOW_PIPELINE_CHANGES` environment variable and
+`X-MCP-Allow-Pipeline-Changes` request header control pipeline-changing tools. The
+`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.
+
+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. The legacy `AIRBYTE_CLOUD_MCP_READONLY_MODE=1` also disables pipeline changes.
+
+When external access is unset, it is disabled if pipeline changes are disabled or
+`AIRBYTE_CLOUD_MCP_SAFE_MODE=1` is explicitly set; otherwise external-access tools remain
+available. Explicitly allowing external access overrides these calculated defaults. Running or
+cancelling a sync is not a pipeline change.
+
 ## Authentication for Remote (HTTP) Servers
 
 The steps above run the MCP server over **stdio** — the client launches the

airbyte/mcp/_error_handling.py

@@ -6,16 +6,22 @@
 from airbyte.exceptions import (
     AirbyteAgentsUnavailableError,
     AirbyteConnectorInUseError,
+    AirbyteExternalAccessDisabledError,
     AirbyteLibError,
     AirbyteLibInputError,
     AirbyteMCPError,
     AirbyteMissingResourceError,
+    AirbytePipelineChangesDisabledError,
+    AirbyteSafeModeError,
 )
 
 
 MCP_TOOL_USER_FACING_ERRORS: tuple[type[AirbyteLibError], ...] = (
     AirbyteLibInputError,
     AirbyteMCPError,
+    AirbyteSafeModeError,
+    AirbytePipelineChangesDisabledError,
+    AirbyteExternalAccessDisabledError,
     AirbyteAgentsUnavailableError,
     AirbyteConnectorInUseError,
     AirbyteMissingResourceError,

airbyte/mcp/_policy_middleware.py

@@ -0,0 +1,53 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Runtime enforcement for MCP tool policies."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from fastmcp.server.middleware import Middleware
+
+from airbyte.exceptions import (
+    AirbyteExternalAccessDisabledError,
+    AirbytePipelineChangesDisabledError,
+)
+from airbyte.mcp._tool_utils import (
+    external_access_allowed,
+    get_tool_policy,
+    pipeline_changes_allowed,
+)
+
+
+if TYPE_CHECKING:
+    from fastmcp.server.middleware import CallNext, MiddlewareContext
+    from fastmcp.tools import ToolResult
+    from mcp.types import CallToolRequestParams
+
+
+class PolicyGuardMiddleware(Middleware):
+    """Enforce pipeline-change and external-access policies before tool execution."""
+
+    async def on_call_tool(
+        self,
+        context: MiddlewareContext[CallToolRequestParams],
+        call_next: CallNext[CallToolRequestParams, ToolResult],
+    ) -> ToolResult:
+        """Reject calls denied by effective pipeline or external-access settings."""
+        fastmcp_context = context.fastmcp_context
+        if fastmcp_context is None:
+            return await call_next(context)
+
+        tool = await fastmcp_context.fastmcp.get_tool(context.message.name)
+        if tool is None:
+            return await call_next(context)
+
+        policy = get_tool_policy(tool)
+        if 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."
+            )
+        return await call_next(context)

airbyte/mcp/_tool_utils.py

@@ -12,8 +12,9 @@
 import inspect
 import os
 from collections.abc import Callable, Iterable, Mapping
+from dataclasses import dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, TypeVar
+from typing import TYPE_CHECKING, TypeVar, cast
 
 from fastmcp.server.dependencies import get_access_token, get_http_headers
 from fastmcp_extensions import (
@@ -35,16 +36,24 @@
     CONFIG_TRUSTED_EXECUTION,
     get_annotation,
 )
+from mcp.types import Tool
 
 from airbyte.constants import (
     CLOUD_API_ROOT_ENV_VAR,
     CLOUD_BEARER_TOKEN_ENV_VAR,
     CLOUD_CLIENT_ID_ENV_VAR,
     CLOUD_CLIENT_SECRET_ENV_VAR,
     CLOUD_CONFIG_API_ROOT_ENV_VAR,
+    CLOUD_MCP_SAFE_MODE_ENV_VAR,
     CLOUD_ORGANIZATION_ID_ENV_VAR,
     CLOUD_WORKSPACE_ID_ENV_VAR,
+    MCP_ALLOW_EXTERNAL_ACCESS_ENV_VAR,
+    MCP_ALLOW_EXTERNAL_ACCESS_HEADER,
+    MCP_ALLOW_PIPELINE_CHANGES_ENV_VAR,
+    MCP_ALLOW_PIPELINE_CHANGES_HEADER,
     MCP_BEARER_TOKEN_HEADER,
+    MCP_CONFIG_ALLOW_EXTERNAL_ACCESS,
+    MCP_CONFIG_ALLOW_PIPELINE_CHANGES,
     MCP_CONFIG_API_URL,
     MCP_CONFIG_BEARER_TOKEN,
     MCP_CONFIG_CLIENT_ID,
@@ -73,7 +82,8 @@
 if TYPE_CHECKING:
     from fastmcp import FastMCP
     from fastmcp.apps import AppConfig
-    from mcp.types import Tool
+    from fastmcp.server.context import Context
+    from fastmcp.tools import Tool as FastMCPTool
 
 _MCP_TOOL_FUNC = TypeVar("_MCP_TOOL_FUNC", bound=Callable[..., object])
 
@@ -172,6 +182,24 @@ def check_guid_created_in_session(guid: str) -> None:
 the include-list opt-in.
 """
 
+ALLOW_PIPELINE_CHANGES_CONFIG_ARG = MCPServerConfigArg(
+    name=MCP_CONFIG_ALLOW_PIPELINE_CHANGES,
+    http_header_key=MCP_ALLOW_PIPELINE_CHANGES_HEADER,
+    env_var=MCP_ALLOW_PIPELINE_CHANGES_ENV_VAR,
+    default="",
+    required=False,
+)
+"""Config arg for permission to run pipeline-changing Cloud tools."""
+
+ALLOW_EXTERNAL_ACCESS_CONFIG_ARG = MCPServerConfigArg(
+    name=MCP_CONFIG_ALLOW_EXTERNAL_ACCESS,
+    http_header_key=MCP_ALLOW_EXTERNAL_ACCESS_HEADER,
+    env_var=MCP_ALLOW_EXTERNAL_ACCESS_ENV_VAR,
+    default="",
+    required=False,
+)
+"""Config arg for permission to run external-access Cloud tools."""
+
 TRUSTED_EXECUTION_CONFIG_ARG = MCPServerConfigArg(
     name=CONFIG_TRUSTED_EXECUTION,
     env_var=MCP_TRUSTED_EXECUTION_ENV_VAR,
@@ -343,6 +371,24 @@ def _parse_csv_config(value: str) -> list[str]:
     return [item.strip() for item in value.split(",") if item.strip()]
 
 
+@dataclass(frozen=True)
+class ToolPolicy:
+    pipeline_change: bool
+    external_access: bool = False
+
+
+_TOOL_POLICIES: dict[str, ToolPolicy] = {}
+
+
+def get_tool_policy(tool: Tool | FastMCPTool) -> ToolPolicy:
+    """Return registered policy metadata, falling back to the tool's read-only hint."""
+    policy = _TOOL_POLICIES.get(tool.name)
+    if policy is not None:
+        return policy
+    read_only = get_annotation(cast(Tool, tool), ANNOTATION_READ_ONLY_HINT, default=False)
+    return ToolPolicy(pipeline_change=not bool(read_only))
+
+
 def mcp_tool(  # noqa: PLR0913 - mirrors the upstream decorator's kwargs
     *,
     read_only: bool = False,
@@ -356,6 +402,8 @@ def mcp_tool(  # noqa: PLR0913 - mirrors the upstream decorator's kwargs
     app: AppConfig | None = None,
     required_capabilities: Iterable[Capability | str] | None = None,
     extra_help_text: str | None = None,
+    pipeline_change: bool | None = None,
+    external_access: bool = False,
 ) -> Callable[[_MCP_TOOL_FUNC], _MCP_TOOL_FUNC]:
     """Decorate an MCP tool with deferred Airbyte registration metadata."""
     base_decorator = _mcp_tool(
@@ -381,6 +429,10 @@ def decorator(func: _MCP_TOOL_FUNC) -> _MCP_TOOL_FUNC:
         if registered_func is not decorated:
             raise RuntimeError("Unexpected MCP tool registration state.")
         registered_annotations[ANNOTATION_MCP_MODULE] = _mcp_module_for_tool(decorated)
+        _TOOL_POLICIES[decorated.__name__] = ToolPolicy(
+            pipeline_change=not read_only if pipeline_change is None else pipeline_change,
+            external_access=external_access,
+        )
         return decorated
 
     return decorator
@@ -420,14 +472,58 @@ def _normalize_mcp_module(mcp_module: str) -> str:
 
 
 def airbyte_readonly_mode_filter(tool: Tool, app: FastMCP) -> bool:
-    """Filter tools based on legacy AIRBYTE_CLOUD_MCP_READONLY_MODE env var.
+    """Hide pipeline-changing tools when pipeline changes are disabled."""
+    return not (pipeline_changes_allowed(app) is False and get_tool_policy(tool).pipeline_change)
 
-    When set to "1", only show tools with readOnlyHint=True.
+
+def airbyte_external_access_filter(tool: Tool, app: FastMCP) -> bool:
+    """Hide external-access tools when external access is disabled."""
+    return not (get_tool_policy(tool).external_access and external_access_allowed(app) is False)
+
+
+def _resolve_policy(
+    app_or_ctx: FastMCP | Context,
+    config_name: str,
+    env_var: str,
+) -> bool | None:
+    environment_value = _str_to_bool(os.environ.get(env_var))
+    if environment_value is False:
+        return False
+    request_value = _str_to_bool(get_mcp_config(app_or_ctx, config_name))
+    if environment_value is True:
+        return request_value is not False
+    return request_value
+
+
+def pipeline_changes_allowed(app_or_ctx: FastMCP | Context) -> bool | None:
+    if _str_to_bool(os.environ.get(MCP_READONLY_MODE_ENV_VAR)) is True:
+        return False
+    return _resolve_policy(
+        app_or_ctx,
+        MCP_CONFIG_ALLOW_PIPELINE_CHANGES,
+        MCP_ALLOW_PIPELINE_CHANGES_ENV_VAR,
+    )
+
+
+def external_access_allowed(app_or_ctx: FastMCP | Context) -> bool | None:
+    """Resolve explicit permission first, then derive the external-access default.
+
+    Disabled pipeline changes and explicitly enabled safe mode disable external access
+    when no explicit permission is configured; otherwise the result remains unset.
     """
-    config_value = (get_mcp_config(app, MCP_CONFIG_READONLY_MODE) or "").lower()
-    if config_value in {"1", "true"}:
-        return bool(get_annotation(tool, ANNOTATION_READ_ONLY_HINT, default=False))
-    return True
+    explicit_value = _resolve_policy(
+        app_or_ctx,
+        MCP_CONFIG_ALLOW_EXTERNAL_ACCESS,
+        MCP_ALLOW_EXTERNAL_ACCESS_ENV_VAR,
+    )
+    if explicit_value is not None:
+        return explicit_value
+    explicitly_enabled_safe_mode = _str_to_bool(os.environ.get(CLOUD_MCP_SAFE_MODE_ENV_VAR)) is True
+    return (
+        False
+        if pipeline_changes_allowed(app_or_ctx) is False or explicitly_enabled_safe_mode
+        else None
+    )
 
 
 def _insiders_mode(app: FastMCP) -> bool | None: