feat(mcp): add batch stream config overrides and update_cloud_connection_unsafe

airbyte/_util/api_util.py

@@ -3937,6 +3937,7 @@ def replace_connection_catalog(
     client_id: SecretString | None,
     client_secret: SecretString | None,
     bearer_token: SecretString | None,
+    skip_reset: bool = True,
 ) -> dict[str, Any]:
     """Replace the configured catalog for a connection.
 
@@ -3953,6 +3954,7 @@ def replace_connection_catalog(
         client_secret: OAuth client secret.
         bearer_token: Bearer token for authentication (alternative to client credentials).
         config_api_root: Optional explicit Config API root URL.
+        skip_reset: Whether to skip the platform reset of changed streams.
 
     Returns:
         Dictionary containing the updated WebBackendConnectionRead response.
@@ -3964,7 +3966,7 @@ def replace_connection_catalog(
             "syncCatalog": configured_catalog_dict,
             # Resets are destructive and cause customer-side data outage.
             # If a reset is desired, caller will need to decide & manage.
-            "skipReset": True,
+            "skipReset": skip_reset,
         },
         api_root=api_root,
         config_api_root=config_api_root,

airbyte/cloud/__init__.py

@@ -98,6 +98,9 @@
     ConnectionStatus,
     JobStatusEnum,
     JobTypeEnum,
+    StreamConfigChange,
+    StreamConfigOverride,
+    StreamConfigUpdateResult,
     WorkspacePrivilegeScope,
 )
 from airbyte.cloud.organizations import CloudOrganization
@@ -138,6 +141,9 @@
     "CloudClientConfig",
     "CloudDefaultContextInfo",
     "CloudWorkspaceInfo",
+    "StreamConfigOverride",
+    "StreamConfigChange",
+    "StreamConfigUpdateResult",
     "SyncResult",
     # Enums
     "ExternalApiReadOnlyAction",

airbyte/cloud/_connection_settings.py

@@ -0,0 +1,71 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Private helpers for validating and updating Cloud connection settings."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from airbyte._util import api_util
+from airbyte.exceptions import AirbyteLibInputError
+
+
+if TYPE_CHECKING:
+    from airbyte.cloud.connections import CloudConnection
+    from airbyte.cloud.models import StreamConfigOverride
+
+
+def _validate_update_connection_args(
+    *,
+    enabled: bool | None,
+    cron_expression: str | None,
+    interval_hours: int | None,
+    manual_schedule: bool | None,
+    override_stream_configs: list[StreamConfigOverride] | None,
+) -> None:
+    """Validate the connection settings before any API writes."""
+    if not any(
+        (
+            enabled is not None,
+            cron_expression is not None,
+            interval_hours is not None,
+            manual_schedule is not None,
+            bool(override_stream_configs),
+        )
+    ):
+        raise AirbyteLibInputError(
+            message=(
+                "At least one setting must be provided: 'enabled', 'interval_hours', "
+                "'cron_expression', 'manual_schedule', or 'override_stream_configs'."
+            ),
+        )
+
+    if sum((interval_hours is not None, cron_expression is not None, manual_schedule is True)) > 1:
+        raise AirbyteLibInputError(
+            message=(
+                "Cannot specify more than one of 'interval_hours', 'cron_expression', "
+                "and 'manual_schedule=True'."
+            ),
+        )
+
+    if interval_hours is not None:
+        api_util.validate_interval_hours(interval_hours)
+
+
+def _set_connection_schedule(
+    connection: CloudConnection,
+    *,
+    cron_expression: str | None,
+    interval_hours: int | None,
+    manual_schedule: bool | None,
+) -> str | None:
+    """Set the requested connection schedule and return its summary."""
+    if cron_expression is not None:
+        connection.set_schedule(cron_expression=cron_expression)
+        return f"schedule set to '{cron_expression}'"
+    if interval_hours is not None:
+        connection.set_schedule(interval_hours=interval_hours)
+        return f"schedule set to 'every {interval_hours} hours'"
+    if manual_schedule is True:
+        connection.set_manual_schedule()
+        return "schedule set to 'manual'"
+    return None