diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py
index b28a33092..9fc64e750 100644
--- a/airbyte/_util/api_util.py
+++ b/airbyte/_util/api_util.py
@@ -2776,6 +2776,42 @@ def get_connection_catalog(
     )
 
 
+def get_refreshed_connection_catalog(
+    connection_id: str,
+    *,
+    api_root: str,
+    client_id: SecretString | None,
+    client_secret: SecretString | None,
+    bearer_token: SecretString | None,
+) -> dict[str, Any]:
+    """Get the configured catalog for a connection with a refreshed schema from the source.
+
+    Uses the Config API endpoint: POST /v1/web_backend/connections/get
+    with ``withRefreshedCatalog: true``, which triggers a discover operation
+    on the connection's source and returns the updated catalog.
+
+    This is equivalent to clicking "Refresh source schema" in the Airbyte UI.
+
+    Args:
+        connection_id: The connection ID to get catalog for.
+        api_root: The API root URL.
+        client_id: OAuth client ID.
+        client_secret: OAuth client secret.
+        bearer_token: Bearer token for authentication (alternative to client credentials).
+
+    Returns:
+        Dictionary containing the connection info with refreshed syncCatalog.
+    """
+    return _make_config_api_request(
+        path="/web_backend/connections/get",
+        json={"connectionId": connection_id, "withRefreshedCatalog": True},
+        api_root=api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+    )
+
+
 def replace_connection_catalog(
     connection_id: str,
     configured_catalog_dict: dict[str, Any],
diff --git a/airbyte/cloud/connections.py b/airbyte/cloud/connections.py
index 14b05b883..2909edae5 100644
--- a/airbyte/cloud/connections.py
+++ b/airbyte/cloud/connections.py
@@ -807,6 +807,170 @@ def import_raw_catalog(self, catalog: dict[str, Any]) -> None:
             config_api_root=self.workspace.config_api_root,
         )
 
+    def refresh_catalog(self) -> dict[str, Any]:
+        """Refresh the connection's catalog by re-discovering the source schema.
+
+        Triggers a discover operation on the connection's source connector, then
+        replaces the connection's catalog with the refreshed result. This is
+        equivalent to clicking "Refresh source schema" in the Airbyte UI.
+
+        This is useful after pinning a new connector version that advertises
+        new streams or updated sync mode support.
+
+        Returns:
+            The refreshed syncCatalog dict.
+
+        Raises:
+            PyAirbyteInputError: If the refreshed catalog is empty or missing.
+        """
+        refreshed_response = api_util.get_refreshed_connection_catalog(
+            connection_id=self.connection_id,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
+        )
+        refreshed_catalog = refreshed_response.get("syncCatalog")
+        if not refreshed_catalog:
+            raise PyAirbyteInputError(
+                message="Refreshed catalog is empty.",
+                context={"connection_id": self.connection_id},
+            )
+
+        api_util.replace_connection_catalog(
+            connection_id=self.connection_id,
+            configured_catalog_dict=refreshed_catalog,
+            api_root=self.workspace.api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
+        )
+        return refreshed_catalog
+
+    def set_stream_sync_mode(
+        self,
+        stream_name: str,
+        *,
+        sync_mode: str,
+        stream_namespace: str | None = None,
+        destination_sync_mode: str | None = None,
+        cursor_field: str | None = None,
+    ) -> None:
+        """Set the sync mode for a specific stream on this connection.
+
+        Safely modifies only the specified stream in the connection's syncCatalog.
+        Validates that the requested sync mode is supported by the stream before
+        applying the change.
+
+        Args:
+            stream_name: The name of the stream to modify.
+            sync_mode: The source sync mode to set (``"incremental"`` or ``"full_refresh"``).
+            stream_namespace: The namespace of the stream to modify. If not provided and
+                multiple streams share the same name, a ``PyAirbyteInputError`` is raised.
+            destination_sync_mode: The destination sync mode to set
+                (``"append"``, ``"overwrite"``, or ``"append_dedup"``). If not provided,
+                the existing destination sync mode is preserved.
+            cursor_field: The cursor field to use for incremental syncs. Required when
+                switching to incremental mode if the stream does not have a default
+                cursor. If not provided, the existing cursor field is preserved.
+
+        Raises:
+            PyAirbyteInputError: If the stream is not found in the catalog, the requested
+                sync mode is not in the stream's ``supportedSyncModes``, the stream name
+                is ambiguous (multiple matches without a namespace), or incremental mode
+                is requested without a usable cursor field.
+        """
+        catalog = self.dump_raw_catalog(normalize=False)
+        if not catalog or "streams" not in catalog:
+            raise PyAirbyteInputError(
+                message="Connection catalog is empty or missing.",
+                context={"connection_id": self.connection_id},
+            )
+
+        # Find all matching streams by name (and optionally namespace)
+        matching_entries: list[dict[str, Any]] = []
+        for entry in catalog["streams"]:
+            stream_def = entry.get("stream", {})
+            if stream_def.get("name") != stream_name:
+                continue
+            if stream_namespace is not None and stream_def.get("namespace") != stream_namespace:
+                continue
+            matching_entries.append(entry)
+
+        if not matching_entries:
+            available_streams = [
+                f"{e.get('stream', {}).get('namespace', '')}.{e.get('stream', {}).get('name', '')}"
+                for e in catalog["streams"]
+            ]
+            raise PyAirbyteInputError(
+                message=f"Stream '{stream_name}' not found in connection catalog.",
+                context={
+                    "connection_id": self.connection_id,
+                    "available_streams": available_streams,
+                },
+            )
+
+        if len(matching_entries) > 1:
+            matching_namespaces = [e.get("stream", {}).get("namespace") for e in matching_entries]
+            raise PyAirbyteInputError(
+                message=(
+                    f"Stream name '{stream_name}' is ambiguous "
+                    f"({len(matching_entries)} matches found)."
+                ),
+                context={
+                    "connection_id": self.connection_id,
+                    "stream_name": stream_name,
+                    "matching_namespaces": matching_namespaces,
+                },
+            )
+
+        target_entry = matching_entries[0]
+
+        # Validate the sync mode is supported
+        stream_def = target_entry.get("stream", {})
+        supported_sync_modes: list[str] = stream_def.get("supportedSyncModes", [])
+        if sync_mode not in supported_sync_modes:
+            raise PyAirbyteInputError(
+                message=(f"Sync mode '{sync_mode}' is not supported by stream '{stream_name}'."),
+                context={
+                    "stream_name": stream_name,
+                    "requested_sync_mode": sync_mode,
+                    "supported_sync_modes": supported_sync_modes,
+                },
+            )
+
+        # Update the stream config
+        config = target_entry.get("config") or {}
+
+        # Guard: fail fast if switching to incremental without a usable cursor
+        if sync_mode == "incremental":
+            existing_cursor = config.get("cursorField") or stream_def.get("defaultCursorField")
+            source_defined_cursor = bool(stream_def.get("sourceDefinedCursor"))
+            if cursor_field is None and not existing_cursor and not source_defined_cursor:
+                raise PyAirbyteInputError(
+                    message=(
+                        f"Stream '{stream_name}' needs a cursor field before switching "
+                        "to incremental sync mode."
+                    ),
+                    context={
+                        "connection_id": self.connection_id,
+                        "stream_name": stream_name,
+                    },
+                )
+
+        config["syncMode"] = sync_mode
+
+        if destination_sync_mode is not None:
+            config["destinationSyncMode"] = destination_sync_mode
+
+        if cursor_field is not None:
+            config["cursorField"] = [cursor_field]
+
+        target_entry["config"] = config
+
+        # Save the updated catalog
+        self.import_raw_catalog(catalog)
+
     def rename(self, name: str) -> CloudConnection:
         """Rename the connection.
 
diff --git a/airbyte/mcp/_config_submit.py b/airbyte/mcp/_config_submit.py
new file mode 100644
index 000000000..90e63b1d8
--- /dev/null
+++ b/airbyte/mcp/_config_submit.py
@@ -0,0 +1,363 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Stateless encrypted action capabilities for connector configuration forms.
+
+The action token is encrypted with a process-shared key so its claims, including
+Cloud credentials, are not visible to the MCP client. The process-random fallback
+is suitable only for a single-process spike. Multi-replica deployments must
+configure the same `AIRBYTE_MCP_FORM_SIGNING_KEY` on every replica; replay
+tracking is still process-local best effort, while token expiration is the actual
+lifetime bound.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+import json
+import os
+import secrets
+import threading
+import time
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Any, Literal, cast
+
+import jsonschema
+from cryptography.exceptions import InvalidTag
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Route
+
+from airbyte import get_source
+from airbyte._util.api_util import SDKError
+from airbyte.cloud.client import CloudClient
+from airbyte.exceptions import AirbyteError, PyAirbyteInputError
+from airbyte.mcp._tool_utils import (
+    SafeModeError,
+    _resolve_transport_bearer_token,
+    check_guid_created_in_session,
+)
+
+
+if TYPE_CHECKING:
+    from starlette.requests import Request
+
+
+DEFAULT_FORM_TOKEN_TTL_SECONDS = 600
+_FORM_SIGNING_KEY_ENV = "AIRBYTE_MCP_FORM_SIGNING_KEY"
+_PROCESS_SIGNING_KEY = secrets.token_bytes(32)
+_ACTION_NAMES = frozenset({"create", "update", "validate"})
+_AES_GCM_NONCE_SIZE = 12
+_SEEN_JTIS: dict[str, int] = {}
+_SEEN_JTIS_LOCK = threading.Lock()
+
+
+class ConfigSubmitError(ValueError):
+    """Raised when an encrypted configuration-submit capability is invalid."""
+
+
+def _signing_key() -> bytes:
+    configured = os.getenv(_FORM_SIGNING_KEY_ENV, "")
+    return configured.encode() if configured else _PROCESS_SIGNING_KEY
+
+
+def _encryption_key() -> bytes:
+    return hashlib.sha256(_signing_key()).digest()
+
+
+def _tenant_claim(token: str | None = None) -> str:
+    if token is None:
+        token = _resolve_transport_bearer_token()
+    return hashlib.sha256(token.encode()).hexdigest() if token else ""
+
+
+def _encode(value: bytes) -> str:
+    return base64.urlsafe_b64encode(value).rstrip(b"=").decode()
+
+
+def _decode(value: str) -> bytes:
+    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
+
+
+def mint_action_token(  # noqa: PLR0913
+    action: Literal["create", "update", "validate"],
+    connector_name: str,
+    *,
+    non_secret_defaults: Mapping[str, Any] | None = None,
+    workspace_id: str | None = None,
+    source_id: str | None = None,
+    source_name: str | None = None,
+    bearer_token: str | None = None,
+    client_id: str | None = None,
+    client_secret: str | None = None,
+    api_url: str | None = None,
+    config_api_url: str | None = None,
+    tenant_claim: str | None = None,
+    ttl_seconds: int = DEFAULT_FORM_TOKEN_TTL_SECONDS,
+) -> str:
+    """Mint an encrypted, one-shot action capability for the connector form."""
+    if action not in _ACTION_NAMES:
+        raise ConfigSubmitError("Invalid configuration action.")
+    if not connector_name:
+        raise ConfigSubmitError("Connector name is required.")
+    if ttl_seconds <= 0:
+        raise ConfigSubmitError("The form token TTL must be positive.")
+    if (client_id is None) != (client_secret is None):
+        raise ConfigSubmitError("Client credentials must be provided together.")
+
+    claims: dict[str, Any] = {
+        "action": action,
+        "connector_name": connector_name,
+        "tenant_claim": tenant_claim if tenant_claim is not None else _tenant_claim(),
+        "exp": int(time.time()) + ttl_seconds,
+        "jti": secrets.token_urlsafe(18),
+    }
+    optional_claims = {
+        "non_secret_defaults": dict(non_secret_defaults)
+        if non_secret_defaults is not None
+        else None,
+        "workspace_id": workspace_id,
+        "source_id": source_id,
+        "source_name": source_name,
+        "bearer_token": bearer_token,
+        "client_id": client_id,
+        "client_secret": client_secret,
+        "api_url": api_url,
+        "config_api_url": config_api_url,
+    }
+    claims.update({key: value for key, value in optional_claims.items() if value is not None})
+    plaintext = json.dumps(claims, separators=(",", ":")).encode()
+    nonce = secrets.token_bytes(12)
+    ciphertext = AESGCM(_encryption_key()).encrypt(nonce, plaintext, None)
+    return _encode(nonce + ciphertext)
+
+
+def _validate_claims(payload: object) -> dict[str, Any]:
+    if not isinstance(payload, dict):
+        raise ConfigSubmitError("Invalid configuration submit token.")
+    required = ("action", "connector_name", "tenant_claim", "exp", "jti")
+    if any(key not in payload for key in required):
+        raise ConfigSubmitError("Invalid configuration submit token.")
+    valid_types = (
+        isinstance(payload["action"], str)
+        and payload["action"] in _ACTION_NAMES
+        and isinstance(payload["connector_name"], str)
+        and isinstance(payload["tenant_claim"], str)
+        and isinstance(payload["exp"], int)
+        and not isinstance(payload["exp"], bool)
+        and isinstance(payload["jti"], str)
+    )
+    if not valid_types:
+        raise ConfigSubmitError("Invalid configuration submit token.")
+    for key in (
+        "non_secret_defaults",
+        "workspace_id",
+        "source_id",
+        "source_name",
+        "bearer_token",
+        "client_id",
+        "client_secret",
+        "api_url",
+        "config_api_url",
+    ):
+        if key == "non_secret_defaults":
+            if key in payload and not isinstance(payload[key], dict):
+                raise ConfigSubmitError("Invalid configuration submit token.")
+            continue
+        if key in payload and not isinstance(payload[key], str):
+            raise ConfigSubmitError("Invalid configuration submit token.")
+    if ("client_id" in payload) != ("client_secret" in payload):
+        raise ConfigSubmitError("Invalid configuration submit token.")
+    return cast(dict[str, Any], payload)
+
+
+def _prune_seen_jtis_locked(now: int) -> None:
+    for jti, expires_at in list(_SEEN_JTIS.items()):
+        if expires_at <= now:
+            del _SEEN_JTIS[jti]
+
+
+def decrypt_action_token(token: str, *, consume: bool = True) -> dict[str, Any]:
+    """Decrypt and validate a one-shot action capability, optionally consuming it."""
+    try:
+        encoded = _decode(token)
+        if len(encoded) <= _AES_GCM_NONCE_SIZE:
+            raise ConfigSubmitError("Invalid configuration submit token.")
+        plaintext = AESGCM(_encryption_key()).decrypt(
+            encoded[:_AES_GCM_NONCE_SIZE],
+            encoded[_AES_GCM_NONCE_SIZE:],
+            None,
+        )
+        claims = _validate_claims(json.loads(plaintext))
+    except (binascii.Error, InvalidTag, json.JSONDecodeError, TypeError, ValueError) as error:
+        if isinstance(error, ConfigSubmitError):
+            raise
+        raise ConfigSubmitError("Invalid configuration submit token.") from error
+
+    now = int(time.time())
+    if claims["exp"] <= now:
+        raise ConfigSubmitError("Configuration submit token expired.")
+    if consume:
+        with _SEEN_JTIS_LOCK:
+            _prune_seen_jtis_locked(now)
+            if claims["jti"] in _SEEN_JTIS:
+                raise ConfigSubmitError("Configuration submit token already used.")
+            _SEEN_JTIS[claims["jti"]] = claims["exp"]
+    return claims
+
+
+def _schema_secret_paths(schema: Mapping[str, Any], prefix: str = "") -> set[str]:
+    paths: set[str] = set()
+    if prefix and any(
+        (
+            schema.get("airbyte_secret") is True,
+            schema.get("writeOnly") is True,
+            schema.get("format") == "password",
+        )
+    ):
+        paths.add(prefix)
+    properties = schema.get("properties", {})
+    if isinstance(properties, Mapping):
+        for name, child in properties.items():
+            if not isinstance(name, str) or not isinstance(child, Mapping):
+                continue
+            path = f"{prefix}.{name}" if prefix else name
+            paths.update(_schema_secret_paths(child, path))
+    items = schema.get("items")
+    if isinstance(items, Mapping):
+        paths.update(_schema_secret_paths(items, prefix))
+    for branch_key in ("oneOf", "anyOf", "allOf"):
+        branches = schema.get(branch_key, [])
+        if not isinstance(branches, list):
+            continue
+        for branch in branches:
+            if isinstance(branch, Mapping):
+                paths.update(_schema_secret_paths(branch, prefix))
+    return paths
+
+
+def _cloud_client_from_claims(claims: Mapping[str, Any]) -> CloudClient:
+    return CloudClient(
+        bearer_token=claims.get("bearer_token"),
+        client_id=claims.get("client_id"),
+        client_secret=claims.get("client_secret"),
+        public_api_root=claims.get("api_url"),
+        config_api_root=claims.get("config_api_url"),
+    )
+
+
+def _execute_action(claims: Mapping[str, Any], config: dict[str, Any]) -> dict[str, str]:
+    action = claims["action"]
+    connector_name = claims["connector_name"]
+    if action == "validate":
+        source = get_source(connector_name, no_executor=True)
+        source.set_config(config, validate=True)
+        return {"status": "success", "action": "validated"}
+
+    workspace_id = claims.get("workspace_id")
+    if not workspace_id:
+        raise ConfigSubmitError("Cloud workspace is unavailable.")
+    workspace = _cloud_client_from_claims(claims).get_workspace(workspace_id)
+    if action == "create":
+        source = get_source(connector_name, no_executor=True)
+        source.set_config(config, validate=True)
+        deployed_source = workspace.deploy_source(
+            name=claims.get("source_name") or connector_name,
+            source=source,
+            unique=True,
+        )
+        return {
+            "status": "success",
+            "action": "created",
+            "connector_id": deployed_source.connector_id,
+            "connector_url": deployed_source.connector_url,
+        }
+
+    source_id = claims.get("source_id")
+    if not source_id:
+        raise ConfigSubmitError("Cloud source is unavailable.")
+    check_guid_created_in_session(source_id)
+    source = workspace.get_source(source_id=source_id)
+    source.update_config(config=config)
+    return {
+        "status": "success",
+        "action": "updated",
+        "connector_id": source.connector_id,
+        "connector_url": source.connector_url,
+    }
+
+
+async def connector_config_submit_endpoint(request: Request) -> Response:  # noqa: PLR0911
+    """Execute a one-shot connector configuration action without echoing config."""
+    if request.method == "OPTIONS":
+        return Response(status_code=204, headers=_cors_headers())
+    authorization = request.headers.get("authorization", "")
+    if not authorization.lower().startswith("bearer "):
+        return _cors_response({"error": "Unauthorized."}, status_code=401)
+    token = authorization[7:].strip()
+    try:
+        claims = decrypt_action_token(token)
+    except ConfigSubmitError:
+        return _cors_response({"error": "Invalid configuration submit request."}, status_code=403)
+
+    try:
+        body = await request.json()
+    except (TypeError, ValueError):
+        return _cors_response({"error": "Invalid configuration submit request."}, status_code=400)
+    config = body.get("config") if isinstance(body, dict) else None
+    if not isinstance(config, dict):
+        return _cors_response({"error": "Invalid configuration submit request."}, status_code=400)
+    try:
+        return _cors_response(_execute_action(claims, config))
+    except (
+        AirbyteError,
+        ConfigSubmitError,
+        KeyError,
+        PyAirbyteInputError,
+        SafeModeError,
+        SDKError,
+        TypeError,
+        ValueError,
+        jsonschema.ValidationError,
+    ):
+        return _cors_response(
+            {"error": "Connector configuration could not be submitted."}, status_code=422
+        )
+
+
+def _cors_response(body: dict[str, Any], *, status_code: int = 200) -> JSONResponse:
+    return JSONResponse(
+        body,
+        status_code=status_code,
+        headers=_cors_headers(),
+    )
+
+
+def _cors_headers() -> dict[str, str]:
+    return {
+        "Access-Control-Allow-Origin": "*",
+        "Access-Control-Allow-Headers": "Authorization, Content-Type",
+        "Access-Control-Allow-Methods": "POST, OPTIONS",
+    }
+
+
+def connector_config_submit_routes() -> list[Route]:
+    """Return the Starlette routes for connector configuration submission."""
+    return [
+        Route(
+            "/connector-config-submit",
+            connector_config_submit_endpoint,
+            methods=["POST", "OPTIONS"],
+        ),
+    ]
+
+
+__all__ = [
+    "ConfigSubmitError",
+    "DEFAULT_FORM_TOKEN_TTL_SECONDS",
+    "_schema_secret_paths",
+    "connector_config_submit_endpoint",
+    "connector_config_submit_routes",
+    "decrypt_action_token",
+    "mint_action_token",
+]
diff --git a/airbyte/mcp/_tool_utils.py b/airbyte/mcp/_tool_utils.py
index ccd6c0058..8ae164134 100644
--- a/airbyte/mcp/_tool_utils.py
+++ b/airbyte/mcp/_tool_utils.py
@@ -13,7 +13,7 @@
 import os
 from collections.abc import Callable, Mapping
 from pathlib import Path
-from typing import TYPE_CHECKING, TypeVar
+from typing import TYPE_CHECKING, Any, TypeVar, cast
 
 from fastmcp.server.dependencies import get_access_token, get_http_headers
 from fastmcp_extensions import (
@@ -71,6 +71,7 @@
 
 if TYPE_CHECKING:
     from fastmcp import FastMCP
+    from fastmcp.apps import AppConfig
     from mcp.types import Tool
 
 _MCP_TOOL_FUNC = TypeVar("_MCP_TOOL_FUNC", bound=Callable[..., object])
@@ -423,6 +424,10 @@ def register_mcp_tools(
             excluded = [name for name in exclude_args if name in params]
             tool_exclude_args = excluded or None
 
+        tool_app = tool_annotations.get(_TOOL_APP_KEY)
+        if callable(tool_app):
+            tool_app = tool_app()
+        tool_app = cast("AppConfig | bool | dict[str, Any] | None", tool_app)
         app.tool(
             func,
             annotations={
@@ -432,7 +437,7 @@ def register_mcp_tools(
             },
             exclude_args=tool_exclude_args,
             meta=tool_annotations.get(_TOOL_META_KEY),
-            app=tool_annotations.get(_TOOL_APP_KEY),
+            app=tool_app,
         )
 
     matching_providers = [
diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index e84deee18..2975b75b8 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -2778,6 +2778,142 @@ def update_cloud_connection(
     )
 
 
+@mcp_tool(
+    destructive=True,
+    open_world=True,
+    extra_help_text=CLOUD_AUTH_TIP_TEXT,
+)
+def refresh_connection_catalog(
+    ctx: Context,
+    connection_id: Annotated[
+        str,
+        Field(description="The ID of the connection to refresh the catalog for."),
+    ],
+    *,
+    workspace_id: Annotated[
+        str | None,
+        Field(
+            description=WORKSPACE_ID_TIP_TEXT,
+            default=None,
+        ),
+    ],
+) -> str:
+    """Refresh the catalog for a connection by re-discovering the source schema.
+
+    Triggers a discover operation on the connection's source connector and updates
+    the connection's catalog with the latest stream definitions and supported sync
+    modes. This is equivalent to clicking "Refresh source schema" in the Airbyte UI.
+
+    This is useful after pinning a new connector version that advertises new streams
+    or updated sync mode support.
+    """
+    check_guid_created_in_session(connection_id)
+    workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
+    connection = workspace.get_connection(connection_id=connection_id)
+
+    refreshed_catalog = connection.refresh_catalog()
+    stream_count = len(refreshed_catalog.get("streams", []))
+
+    return (
+        f"Successfully refreshed catalog for connection '{connection_id}'. "
+        f"Catalog now contains {stream_count} stream(s). "
+        f"URL: {connection.connection_url}"
+    )
+
+
+@mcp_tool(
+    destructive=True,
+    open_world=True,
+    extra_help_text=CLOUD_AUTH_TIP_TEXT,
+)
+def set_stream_sync_mode(
+    ctx: Context,
+    connection_id: Annotated[
+        str,
+        Field(description="The ID of the connection to modify."),
+    ],
+    stream_name: Annotated[
+        str,
+        Field(description="The name of the stream to change the sync mode for."),
+    ],
+    sync_mode: Annotated[
+        Literal["incremental", "full_refresh"],
+        Field(description="The source sync mode to set: 'incremental' or 'full_refresh'."),
+    ],
+    *,
+    stream_namespace: Annotated[
+        str | None,
+        Field(
+            description=(
+                "The namespace of the stream to modify. Required when multiple streams "
+                "share the same name but differ by namespace."
+            ),
+            default=None,
+        ),
+    ],
+    destination_sync_mode: Annotated[
+        Literal["append", "overwrite", "append_dedup"] | None,
+        Field(
+            description=(
+                "The destination sync mode to set: 'append', 'overwrite', or 'append_dedup'. "
+                "If not provided, the existing destination sync mode is preserved."
+            ),
+            default=None,
+        ),
+    ],
+    cursor_field: Annotated[
+        str | None,
+        Field(
+            description=(
+                "The cursor field to use for incremental syncs. "
+                "Required when switching to incremental mode if the stream does not have "
+                "a default cursor. If not provided, the existing cursor field is preserved."
+            ),
+            default=None,
+        ),
+    ],
+    workspace_id: Annotated[
+        str | None,
+        Field(
+            description=WORKSPACE_ID_TIP_TEXT,
+            default=None,
+        ),
+    ],
+) -> str:
+    """Set the sync mode for a specific stream on a connection.
+
+    Safely modifies only the specified stream in the connection's syncCatalog.
+    Validates that the requested sync mode is supported by the stream before
+    applying the change.
+
+    This is useful when switching a stream from full_refresh to incremental
+    (or vice versa) after a connector version upgrade that adds incremental support.
+    """
+    check_guid_created_in_session(connection_id)
+    workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
+    connection = workspace.get_connection(connection_id=connection_id)
+
+    connection.set_stream_sync_mode(
+        stream_name=stream_name,
+        sync_mode=sync_mode,
+        stream_namespace=stream_namespace,
+        destination_sync_mode=destination_sync_mode,
+        cursor_field=cursor_field,
+    )
+
+    return (
+        f"Successfully set sync mode for stream '{stream_name}' "
+        f"on connection '{connection_id}' to '{sync_mode}'"
+        + (
+            f" with destination sync mode '{destination_sync_mode}'"
+            if destination_sync_mode
+            else ""
+        )
+        + (f" and cursor field '{cursor_field}'" if cursor_field else "")
+        + f". URL: {connection.connection_url}"
+    )
+
+
 @mcp_tool(
     read_only=True,
     idempotent=True,
diff --git a/airbyte/mcp/interactive/__init__.py b/airbyte/mcp/interactive/__init__.py
index 07df22b6d..07ece2928 100644
--- a/airbyte/mcp/interactive/__init__.py
+++ b/airbyte/mcp/interactive/__init__.py
@@ -6,6 +6,10 @@
 from typing import TYPE_CHECKING
 
 from airbyte.mcp._tool_utils import register_mcp_tools
+from airbyte.mcp.interactive._connector_config_form_ui import (
+    register_connector_config_form_resource,
+    show_connector_config_form,
+)
 from airbyte.mcp.interactive._registry_ui import show_connectors_list
 from airbyte.mcp.interactive._sync_history_ui import show_connection_sync_history
 from airbyte.mcp.interactive._workspace_sync_status_ui import show_workspace_sync_status
@@ -18,10 +22,12 @@
 def register_interactive_tools(app: FastMCP) -> None:
     """Register UI-presenting tools."""
     register_mcp_tools(app, mcp_module="interactive")
+    register_connector_config_form_resource(app)
 
 
 __all__ = [
     "register_interactive_tools",
+    "show_connector_config_form",
     "show_connectors_list",
     "show_connection_sync_history",
     "show_workspace_sync_status",
diff --git a/airbyte/mcp/interactive/_connector_config_form_ui.py b/airbyte/mcp/interactive/_connector_config_form_ui.py
new file mode 100644
index 000000000..f83502895
--- /dev/null
+++ b/airbyte/mcp/interactive/_connector_config_form_ui.py
@@ -0,0 +1,558 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Interactive connector configuration form MCP App."""
+
+from __future__ import annotations
+
+import json
+import os
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Annotated, Any, Literal
+from urllib.parse import urlencode, urlparse
+
+from fastmcp import Context  # noqa: TC002
+from fastmcp.apps import UI_MIME_TYPE, AppConfig, ResourceCSP
+from fastmcp.tools.base import ToolResult
+from fastmcp_extensions import get_mcp_config
+from pydantic import Field
+from starlette.responses import HTMLResponse
+
+from airbyte import exceptions as exc
+from airbyte._util.registry_spec import get_connector_spec_from_registry
+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_WORKSPACE_ID_ENV_VAR,
+)
+from airbyte.mcp._config_submit import (
+    ConfigSubmitError,
+    _schema_secret_paths,
+    decrypt_action_token,
+    mint_action_token,
+)
+from airbyte.mcp._tool_utils import (
+    INTERACTIVE_UI_ANNOTATION,
+    _resolve_transport_bearer_token,
+    mcp_tool,
+)
+
+
+if TYPE_CHECKING:
+    from fastmcp import FastMCP
+    from starlette.requests import Request
+
+
+CONNECTOR_FORM_RESOURCE_URI = "ui://airbyte/connector-config-form"
+MCP_SERVER_URL_ENV = "MCP_SERVER_URL"
+DEFAULT_HTTP_PORT = 8080
+DEFAULT_MCP_SERVER_URL = f"http://localhost:{DEFAULT_HTTP_PORT}"
+_MAX_AGENT_CONTENT_LENGTH = 12_000
+
+_HTML_RESOURCE = """<!doctype html>
+<html><head><meta charset="utf-8"><title>Connector configuration</title>
+<style>
+body{font:14px system-ui,sans-serif;margin:0;padding:20px;color:#172033}
+form{display:grid;gap:14px;max-width:620px}label{display:grid;gap:5px;font-weight:600}
+input{border:1px solid #aab4c3;border-radius:5px;padding:8px;font:inherit}
+button{background:#635bff;color:white;border:0;border-radius:5px;
+padding:9px 14px;font:inherit;cursor:pointer}
+#status{min-height:20px}.error{color:#b42318}.success{color:#067647}
+</style></head>
+<body><h2 id="title">Connector configuration</h2><form id="form"></form><p id="status"></p>
+<script>
+window.__AIRBYTE_FORM_PAYLOAD__ = null;
+(() => {
+  const state = { result: null };
+  const standalonePayload = window.__AIRBYTE_FORM_PAYLOAD__;
+  const standalone = standalonePayload !== null;
+  const form = document.getElementById("form");
+  const status = document.getElementById("status");
+  const title = document.getElementById("title");
+  const post = (message) => window.parent.postMessage(message, "*");
+  const setStatus = (className, message, resultStatus) => {
+    status.className = className;
+    status.textContent = message;
+    if (standalone || resultStatus === "pending") return;
+    const notify = document.createElement("button");
+    notify.type = "button";
+    notify.textContent = "Notify agent";
+    notify.addEventListener("click", () => post({
+      jsonrpc: "2.0",
+      method: "ui/message",
+      params: {content: [{
+        type: "text",
+        text: `Connector configuration ${state.result.action} result: ${resultStatus}`
+      }]}
+    }));
+    status.appendChild(document.createTextNode(" "));
+    status.appendChild(notify);
+  };
+  const requestResize = () => {
+    if (standalone) return;
+    const contentHeight = Math.max(
+      document.documentElement.scrollHeight,
+      document.body.scrollHeight
+    );
+    post({jsonrpc: "2.0", method: "ui/notifications/size-changed",
+      params: {height: Math.min(contentHeight + 20, 600)}});
+  };
+  const pathValue = (object, path) => path.split(".").reduce((value, key) =>
+    value && typeof value === "object" ? value[key] : undefined, object);
+  const setPath = (object, path, value) => {
+    const keys = path.split(".");
+    if (keys.some((key) => ["__proto__", "constructor", "prototype"].includes(key))) return;
+    let target = object;
+    keys.slice(0, -1).forEach((key) => {
+      if (typeof target[key] !== "object" || target[key] === null || Array.isArray(target[key]))
+        target[key] = {};
+      target = target[key];
+    });
+    target[keys[keys.length - 1]] = value;
+  };
+  const fields = (schema, prefix = "") => {
+    const properties = schema && schema.properties || {};
+    return Object.entries(properties).flatMap(([name, child]) => {
+      const path = prefix ? `${prefix}.${name}` : name;
+      if (child && ["oneOf", "anyOf", "allOf"].some((key) => Array.isArray(child[key])))
+        return [{ path, schema: child }];
+      if (child && child.type === "object" && child.properties)
+        return fields(child, path);
+      return [{ path, schema: child || {} }];
+    });
+  };
+  const render = (result) => {
+    state.result = result;
+    title.textContent = `${result.connector_name} configuration`;
+    form.replaceChildren();
+    const required = new Set((result.spec_schema.required || []));
+    fields(result.spec_schema).sort((a,b) =>
+      (required.has(b.path.split(".").pop()) ? 1 : 0) -
+      (required.has(a.path.split(".").pop()) ? 1 : 0) ||
+      (a.schema.order || 999) - (b.schema.order || 999)
+    ).forEach(({path, schema}) => {
+      const label = document.createElement("label");
+      label.textContent = `${schema.title || path}${required.has(path) ? " *" : ""}`;
+      if (["oneOf", "anyOf", "allOf"].some((key) => Array.isArray(schema[key]))) {
+        const notice = document.createElement("span");
+        notice.textContent = "Complex authentication objects are not supported by this form.";
+        notice.setAttribute("aria-disabled", "true");
+        label.appendChild(notice); form.appendChild(label);
+        return;
+      }
+      const input = document.createElement("input");
+      input.name = path;
+      input.type = result.secret_fields.includes(path) ? "password" : "text";
+      const value = pathValue(result.non_secret_defaults || {}, path);
+      if (value !== undefined)
+        input.value = typeof value === "string" ? value : JSON.stringify(value);
+      label.appendChild(input); form.appendChild(label);
+    });
+    const button = document.createElement("button");
+    button.type = "submit"; button.textContent = "Save configuration"; form.appendChild(button);
+    requestResize();
+  };
+  form.addEventListener("submit", async (event) => {
+    event.preventDefault(); setStatus("", "Saving…", "pending");
+    requestResize();
+    const config = JSON.parse(JSON.stringify(state.result.non_secret_defaults || {}));
+    const visible = JSON.parse(JSON.stringify(state.result.non_secret_defaults || {}));
+    form.querySelectorAll("input").forEach((input) => {
+      if (!input.value) return;
+      setPath(config, input.name, input.value);
+      if (!state.result.secret_fields.includes(input.name))
+        setPath(visible, input.name, input.value);
+    });
+    try {
+      const response = await fetch(state.result.submit_endpoint, {
+        method: "POST", headers: {"Authorization": `Bearer ${state.result.submit_token}`,
+        "Content-Type": "application/json"}, body: JSON.stringify({config})
+      });
+      if (!response.ok) throw new Error("The server rejected the configuration.");
+      const body = await response.json();
+      const payload = {status: body.status, action: body.action, visible_config: visible};
+      if (typeof body.connector_id === "string") payload.connector_id = body.connector_id;
+      if (typeof body.connector_url === "string") payload.connector_url = body.connector_url;
+      if (!standalone)
+        post({jsonrpc: "2.0", method: "ui/updateModelContext", params: {content: payload}});
+      setStatus("success", "Configuration submitted.", "success");
+      requestResize();
+    } catch (error) {
+      setStatus("error", "The configuration could not be submitted.", "error");
+      requestResize();
+    }
+  });
+  if (standalonePayload) {
+    render(standalonePayload);
+  } else {
+    window.addEventListener("message", (event) => {
+      if (event.source !== window.parent) return;
+      const message = event.data || {};
+      if (message.id === 1 && message.result) {
+        post({jsonrpc: "2.0", method: "ui/notifications/initialized"});
+        return;
+      }
+      if (message.method === "ui/notifications/tool-result" && message.params) {
+        render(
+          message.params.structuredContent ||
+          message.params.structured_content ||
+          message.params
+        );
+        return;
+      }
+      const result = message.params && (message.params.tool_result || message.params.result);
+      if (result) render(result.structuredContent || result.structured_content || result);
+    });
+    post({jsonrpc: "2.0", id: 1, method: "ui/initialize", params: {
+      protocolVersion: "2025-06-18", appCapabilities: {},
+      appInfo: {name: "airbyte-form", version: "0.1"}
+    }});
+  }
+})();
+</script></body></html>"""
+
+
+def _server_url() -> str:
+    server_url = os.getenv(MCP_SERVER_URL_ENV, "").strip() or DEFAULT_MCP_SERVER_URL
+    parsed = urlparse(server_url)
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        raise exc.PyAirbyteInputError(
+            message="MCP_SERVER_URL must be a valid HTTP(S) URL.",
+        )
+    if parsed.scheme == "http" and parsed.hostname not in {"localhost", "127.0.0.1"}:
+        raise exc.PyAirbyteInputError(
+            message="MCP_SERVER_URL must use HTTPS outside localhost.",
+        )
+    return server_url.rstrip("/")
+
+
+def _server_origin() -> str:
+    parsed = urlparse(_server_url())
+    return f"{parsed.scheme}://{parsed.netloc}"
+
+
+def _submit_endpoint() -> str:
+    return f"{_server_url()}/connector-config-submit"
+
+
+def _standalone_endpoint(token: str) -> str:
+    return f"{_server_url()}/connector-config-form?{urlencode({'token': token})}"
+
+
+def _cloud_config_value(env_name: str) -> str | None:
+    value = os.getenv(env_name, "").strip()
+    return value or None
+
+
+def _resolved_cloud_config_value(
+    ctx: Context | None,
+    config_name: str,
+    env_name: str,
+) -> str | None:
+    if ctx is not None:
+        configured = get_mcp_config(ctx, config_name)
+        if configured:
+            return configured
+    return _cloud_config_value(env_name)
+
+
+def _mint_form_token(
+    connector_name: str,
+    *,
+    config_defaults: dict[str, Any],
+    ctx: Context | None,
+    source_id: str | None,
+    workspace_id: str | None,
+    source_name: str | None,
+) -> tuple[str, str]:
+    resolved_workspace_id = workspace_id or _resolved_cloud_config_value(
+        ctx, "workspace_id", CLOUD_WORKSPACE_ID_ENV_VAR
+    )
+    bearer_token = _resolve_transport_bearer_token() or _resolved_cloud_config_value(
+        ctx, "bearer_token", CLOUD_BEARER_TOKEN_ENV_VAR
+    )
+    client_id = _resolved_cloud_config_value(ctx, "client_id", CLOUD_CLIENT_ID_ENV_VAR)
+    client_secret = _resolved_cloud_config_value(ctx, "client_secret", CLOUD_CLIENT_SECRET_ENV_VAR)
+    has_credentials = bool(bearer_token or (client_id and client_secret))
+    action: Literal["create", "update", "validate"]
+    if source_id:
+        if not (has_credentials and resolved_workspace_id):
+            raise exc.PyAirbyteInputError(
+                message="Cloud credentials and a workspace ID are required to update a source."
+            )
+        action = "update"
+    elif has_credentials and resolved_workspace_id:
+        action = "create"
+    else:
+        action = "validate"
+    token = mint_action_token(
+        action,
+        connector_name,
+        non_secret_defaults=config_defaults,
+        workspace_id=resolved_workspace_id,
+        source_id=source_id,
+        source_name=source_name,
+        bearer_token=bearer_token,
+        client_id=client_id if not bearer_token else None,
+        client_secret=client_secret if not bearer_token else None,
+        api_url=_resolved_cloud_config_value(ctx, "api_url", CLOUD_API_ROOT_ENV_VAR),
+        config_api_url=_resolved_cloud_config_value(
+            ctx, "config_api_url", CLOUD_CONFIG_API_ROOT_ENV_VAR
+        ),
+    )
+    return action, token
+
+
+def _paths_present(value: object, prefix: str = "") -> set[str]:
+    if isinstance(value, (list, tuple)):
+        paths: set[str] = set()
+        for item in value:
+            paths.update(_paths_present(item, prefix))
+        return paths
+    if not isinstance(value, Mapping):
+        return set()
+    paths: set[str] = set()
+    for name, child in value.items():
+        if not isinstance(name, str):
+            continue
+        path = f"{prefix}.{name}" if prefix else name
+        paths.add(path)
+        paths.update(_paths_present(child, path))
+    return paths
+
+
+def _agent_content(
+    connector_name: str,
+    config_defaults: dict[str, Any],
+    secret_fields: list[str],
+    standalone_url: str,
+) -> str:
+    """Build bounded text for agents that cannot render the form."""
+    note = (
+        "The user will enter secrets in the form. The model will only receive "
+        "a confirmation and non-secret configuration."
+    )
+    payload: dict[str, Any] = {
+        "connector_name": connector_name,
+        "non_secret_defaults": config_defaults,
+        "secret_fields": secret_fields,
+        "standalone_url": standalone_url,
+        "note": note,
+        "agent_preview_max_chars": _MAX_AGENT_CONTENT_LENGTH,
+    }
+    content = json.dumps(payload, separators=(",", ":"))
+    if len(content) <= _MAX_AGENT_CONTENT_LENGTH:
+        return content
+
+    payload["non_secret_defaults"] = {}
+    payload["agent_preview_truncated"] = True
+    content = json.dumps(payload, separators=(",", ":"))
+    if len(content) <= _MAX_AGENT_CONTENT_LENGTH:
+        return content
+
+    payload["secret_fields"] = []
+    payload["connector_name"] = connector_name[:256]
+    return json.dumps(payload, separators=(",", ":"))
+
+
+def _connector_spec_schema(connector_name: str) -> dict[str, Any] | None:
+    spec_schema = get_connector_spec_from_registry(connector_name, platform="cloud")
+    if spec_schema is None:
+        spec_schema = get_connector_spec_from_registry(connector_name, platform="oss")
+    return spec_schema
+
+
+def _form_payload(
+    connector_name: str,
+    spec_schema: dict[str, Any],
+    config_defaults: dict[str, Any],
+    secret_fields: list[str],
+    action: str,
+    submit_token: str,
+) -> dict[str, Any]:
+    return {
+        "connector_name": connector_name,
+        "schema": spec_schema,
+        "spec_schema": spec_schema,
+        "non_secret_defaults": config_defaults,
+        "secret_fields": secret_fields,
+        "action": action,
+        "submit_token": submit_token,
+        "submit_endpoint": _submit_endpoint(),
+    }
+
+
+def connector_config_form_page(payload: Mapping[str, Any]) -> str:
+    """Render the connector configuration form as a standalone browser page."""
+    encoded_payload = json.dumps(payload, separators=(",", ":"), ensure_ascii=True).replace(
+        "</", "<\\/"
+    )
+    assignment = f"window.__AIRBYTE_FORM_PAYLOAD__ = {encoded_payload};"
+    return _HTML_RESOURCE.replace(
+        "window.__AIRBYTE_FORM_PAYLOAD__ = null;",
+        assignment,
+        1,
+    )
+
+
+_INVALID_FORM_PAGE = """<!doctype html>
+<html><head><meta charset="utf-8"><title>Invalid form link</title></head>
+<body><p>This link is invalid or has expired.</p></body></html>"""
+
+
+async def connector_config_form_endpoint(request: Request) -> HTMLResponse:  # noqa: RUF029
+    """Render a connector configuration form from a one-shot action capability."""
+    token = request.query_params.get("token", "")
+    if not token:
+        return HTMLResponse(_INVALID_FORM_PAGE, status_code=403)
+    try:
+        claims = decrypt_action_token(token, consume=False)
+        connector_name = claims["connector_name"]
+        config_defaults = claims.get("non_secret_defaults", {})
+        if not isinstance(config_defaults, dict):
+            return HTMLResponse(_INVALID_FORM_PAGE, status_code=403)
+        spec_schema = _connector_spec_schema(connector_name)
+        if not spec_schema:
+            return HTMLResponse(_INVALID_FORM_PAGE, status_code=403)
+        secret_fields = sorted(_schema_secret_paths(spec_schema))
+        payload = _form_payload(
+            connector_name,
+            spec_schema,
+            config_defaults,
+            secret_fields,
+            claims["action"],
+            token,
+        )
+    except (ConfigSubmitError, exc.PyAirbyteError, KeyError, TypeError, ValueError):
+        return HTMLResponse(_INVALID_FORM_PAGE, status_code=403)
+    try:
+        return HTMLResponse(connector_config_form_page(payload))
+    except (TypeError, ValueError):
+        return HTMLResponse(_INVALID_FORM_PAGE, status_code=403)
+
+
+def connector_config_form_resource() -> str:
+    """Return the raw HTML connector configuration form."""
+    return _HTML_RESOURCE
+
+
+def _connector_form_app_config() -> AppConfig:
+    return AppConfig(
+        resource_uri=CONNECTOR_FORM_RESOURCE_URI,
+        csp=ResourceCSP(connect_domains=[_server_origin()]),
+    )
+
+
+def register_connector_config_form_resource(app: FastMCP) -> None:
+    """Register the connector form resource with its CSP metadata."""
+    app.resource(
+        CONNECTOR_FORM_RESOURCE_URI,
+        description="Self-contained Airbyte connector configuration form.",
+        mime_type=UI_MIME_TYPE,
+        app=AppConfig(csp=ResourceCSP(connect_domains=[_server_origin()])),
+    )(connector_config_form_resource)
+
+
+@mcp_tool(
+    read_only=True,
+    idempotent=True,
+    open_world=True,
+    annotations={INTERACTIVE_UI_ANNOTATION: True},
+    app=_connector_form_app_config,
+)
+def show_connector_config_form(
+    connector_name: Annotated[
+        str,
+        Field(description="Connector name, such as `source-pokeapi`."),
+    ],
+    config_defaults: Annotated[
+        dict[str, Any] | str | None,
+        Field(
+            description=(
+                "Non-secret configuration defaults to prefill. Secret values must "
+                "be entered by the user in the form."
+            ),
+        ),
+    ] = None,
+    *,
+    ctx: Context | None = None,
+    source_id: Annotated[
+        str | None,
+        Field(description="Existing Cloud source ID to update.", default=None),
+    ] = None,
+    workspace_id: Annotated[
+        str | None,
+        Field(description="Cloud workspace ID for creating the source.", default=None),
+    ] = None,
+    source_name: Annotated[
+        str | None,
+        Field(description="Name to use when creating the source.", default=None),
+    ] = None,
+) -> ToolResult:
+    """Show a hosted-safe connector configuration form."""
+    if config_defaults is None:
+        config_defaults = {}
+    if isinstance(config_defaults, str):
+        try:
+            config_defaults = json.loads(config_defaults)
+        except json.JSONDecodeError as error:
+            raise exc.PyAirbyteInputError(
+                message="config_defaults must be a JSON object.",
+                context={"connector_name": connector_name},
+            ) from error
+    if not isinstance(config_defaults, dict):
+        raise exc.PyAirbyteInputError(
+            message="config_defaults must be an object.",
+            context={"connector_name": connector_name},
+        )
+
+    spec_schema = _connector_spec_schema(connector_name)
+    if not spec_schema:
+        raise exc.PyAirbyteInputError(
+            message=f"Could not fetch a configuration schema for '{connector_name}'.",
+            context={"connector_name": connector_name},
+        )
+    secret_fields = sorted(_schema_secret_paths(spec_schema))
+    supplied_fields = _paths_present(config_defaults)
+    supplied_secrets = sorted(supplied_fields.intersection(secret_fields))
+    if supplied_secrets:
+        raise exc.PyAirbyteInputError(
+            message=(
+                "Secret values cannot be provided in config_defaults. "
+                "The user must enter them in the form."
+            ),
+            context={"secret_fields": supplied_secrets},
+        )
+
+    action, submit_token = _mint_form_token(
+        connector_name,
+        config_defaults=config_defaults,
+        ctx=ctx,
+        source_id=source_id,
+        workspace_id=workspace_id,
+        source_name=source_name,
+    )
+    standalone_url = _standalone_endpoint(submit_token)
+    content = _agent_content(
+        connector_name,
+        config_defaults,
+        secret_fields,
+        standalone_url,
+    )
+    structured_content = _form_payload(
+        connector_name,
+        spec_schema,
+        config_defaults,
+        secret_fields,
+        action,
+        submit_token,
+    )
+    structured_content["standalone_url"] = standalone_url
+    return ToolResult(content=content, structured_content=structured_content)
+
+
+__all__ = [
+    "connector_config_form_endpoint",
+    "show_connector_config_form",
+]
diff --git a/airbyte/mcp/server.py b/airbyte/mcp/server.py
index 55b322701..5c1615c51 100644
--- a/airbyte/mcp/server.py
+++ b/airbyte/mcp/server.py
@@ -61,7 +61,7 @@
     build_mcp_auth,
     mcp_server,
 )
-from starlette.responses import JSONResponse
+from starlette.responses import JSONResponse, Response
 
 
 if TYPE_CHECKING:
@@ -73,6 +73,7 @@
 from airbyte._util.telemetry import DO_NOT_TRACK, PYAIRBYTE_APP_TRACKING_KEY
 from airbyte.constants import AIRBYTE_OFFLINE_MODE, _str_to_bool, is_hosted_mcp_mode
 from airbyte.mcp._config import load_secrets_to_env_vars
+from airbyte.mcp._config_submit import connector_config_submit_endpoint
 from airbyte.mcp._tool_utils import (
     AIRBYTE_EXCLUDE_MODULES_CONFIG_ARG,
     AIRBYTE_INCLUDE_MODULES_CONFIG_ARG,
@@ -93,6 +94,9 @@
 from airbyte.mcp.agents import register_agents_tools
 from airbyte.mcp.cloud import register_cloud_tools
 from airbyte.mcp.interactive import register_interactive_tools
+from airbyte.mcp.interactive._connector_config_form_ui import (
+    connector_config_form_endpoint,
+)
 from airbyte.mcp.local import register_local_tools
 from airbyte.mcp.prompts import register_prompts
 from airbyte.mcp.registry import register_registry_tools
@@ -385,6 +389,18 @@ async def health_check(request: Request) -> JSONResponse:  # noqa: ARG001, RUF02
     return JSONResponse({"status": "ok"})
 
 
+@app.custom_route("/connector-config-submit", methods=["POST", "OPTIONS"])
+async def connector_config_submit(request: Request) -> Response:
+    """Execute a connector configuration action from the MCP App form."""
+    return await connector_config_submit_endpoint(request)
+
+
+@app.custom_route("/connector-config-form", methods=["GET"])
+async def connector_config_form(request: Request) -> Response:
+    """Render the connector configuration form as a standalone browser page."""
+    return await connector_config_form_endpoint(request)
+
+
 def main() -> None:
     """@private Main entry point for the MCP server.
 
diff --git a/tests/unit_tests/test_mcp_config_submit.py b/tests/unit_tests/test_mcp_config_submit.py
new file mode 100644
index 000000000..2fe402d4f
--- /dev/null
+++ b/tests/unit_tests/test_mcp_config_submit.py
@@ -0,0 +1,258 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Tests for stateless MCP connector configuration submission."""
+
+from __future__ import annotations
+
+import base64
+import time
+from typing import Any, cast
+
+import pytest
+from airbyte_api.errors import SDKError
+from starlette.applications import Starlette
+from starlette.testclient import TestClient
+
+from airbyte.mcp import _config_submit
+from airbyte.mcp._config_submit import (
+    ConfigSubmitError,
+    _schema_secret_paths,
+    connector_config_submit_routes,
+    decrypt_action_token,
+    mint_action_token,
+)
+
+
+def test_token_encrypts_and_decrypts_claims(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("AIRBYTE_MCP_FORM_SIGNING_KEY", "test-signing-key")
+    token = mint_action_token(
+        "create",
+        "source-example",
+        workspace_id="workspace-id",
+        source_name="Example",
+        bearer_token="bearer-secret",
+        api_url="https://api.example.com",
+    )
+
+    assert "bearer-secret" not in token
+    claims = decrypt_action_token(token)
+    assert claims["action"] == "create"
+    assert claims["connector_name"] == "source-example"
+    assert claims["workspace_id"] == "workspace-id"
+    assert claims["bearer_token"] == "bearer-secret"
+
+
+def test_token_rejects_expiry_and_replay(monkeypatch: pytest.MonkeyPatch) -> None:
+    token = mint_action_token("validate", "source-example", ttl_seconds=1)
+    now = time.time()
+    monkeypatch.setattr(_config_submit.time, "time", lambda: now + 2)
+
+    with pytest.raises(ConfigSubmitError, match="expired"):
+        decrypt_action_token(token)
+
+    monkeypatch.setattr(_config_submit.time, "time", lambda: now)
+    token = mint_action_token("validate", "source-example")
+    decrypt_action_token(token)
+    with pytest.raises(ConfigSubmitError, match="already used"):
+        decrypt_action_token(token)
+
+
+def test_token_rejects_ciphertext_tampering() -> None:
+    token = mint_action_token("validate", "source-example")
+    encoded = bytearray(base64.urlsafe_b64decode(token + "=" * (-len(token) % 4)))
+    encoded[-1] ^= 1
+    tampered = base64.urlsafe_b64encode(encoded).rstrip(b"=").decode()
+
+    with pytest.raises(ConfigSubmitError, match="Invalid"):
+        decrypt_action_token(tampered)
+
+
+def test_token_rejects_invalid_claims() -> None:
+    with pytest.raises(ConfigSubmitError, match="Invalid configuration action"):
+        mint_action_token("invalid", "source-example")  # type: ignore[arg-type]
+
+
+def test_schema_secret_paths_include_nested_branches() -> None:
+    schema = {
+        "oneOf": [
+            {
+                "type": "object",
+                "properties": {
+                    "credentials": {
+                        "type": "object",
+                        "properties": {
+                            "api_key": {"type": "string", "airbyte_secret": True},
+                        },
+                    },
+                },
+            },
+            {
+                "type": "object",
+                "properties": {
+                    "credentials": {
+                        "type": "object",
+                        "properties": {
+                            "password": {"type": "string", "airbyte_secret": True},
+                        },
+                    },
+                },
+            },
+        ]
+    }
+
+    assert _schema_secret_paths(schema) == {
+        "credentials.api_key",
+        "credentials.password",
+    }
+
+
+def test_schema_secret_paths_include_hydration_markers_and_items() -> None:
+    schema = {
+        "type": "object",
+        "properties": {
+            "write_only": {"type": "string", "writeOnly": True},
+            "password": {"type": "string", "format": "password"},
+            "items": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "token": {"type": "string", "format": "password"},
+                    },
+                },
+            },
+        },
+    }
+
+    assert _schema_secret_paths(schema) == {
+        "write_only",
+        "password",
+        "items.token",
+    }
+
+
+class _SourceStub:
+    def __init__(self) -> None:
+        self.config: dict[str, object] | None = None
+
+    def set_config(self, config: dict[str, object], *, validate: bool = True) -> None:
+        self.config = config
+        assert validate
+
+
+def test_validate_endpoint_executes_full_config_without_echo(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    source = _SourceStub()
+    monkeypatch.setattr(_config_submit, "get_source", lambda *args, **kwargs: source)
+    token = mint_action_token("validate", "source-example")
+    app = Starlette(routes=connector_config_submit_routes())
+    config = {"credentials": {"api_key": "secret-value"}, "region": "us-east-1"}
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/connector-config-submit",
+            headers={"Authorization": f"Bearer {token}"},
+            json={"config": config},
+        )
+
+    assert response.status_code == 200
+    assert response.json() == {"status": "success", "action": "validated"}
+    assert source.config == config
+    assert "secret-value" not in response.text
+
+
+def test_endpoint_returns_safe_422_for_cloud_sdk_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def raise_sdk_error(*args: Any, **kwargs: Any) -> dict[str, str]:
+        raise SDKError("Cloud rejected the configuration.", 422, "", cast(Any, None))
+
+    monkeypatch.setattr(_config_submit, "_execute_action", raise_sdk_error)
+    app = Starlette(routes=connector_config_submit_routes())
+    token = mint_action_token("validate", "source-example")
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/connector-config-submit",
+            headers={"Authorization": f"Bearer {token}"},
+            json={"config": {"api_key": "secret-value"}},
+        )
+
+    assert response.status_code == 422
+    assert response.json() == {
+        "error": "Connector configuration could not be submitted.",
+    }
+    assert "secret-value" not in response.text
+
+
+def test_endpoint_rejects_missing_malformed_and_reused_tokens(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        _config_submit, "get_source", lambda *args, **kwargs: _SourceStub()
+    )
+    app = Starlette(routes=connector_config_submit_routes())
+
+    with TestClient(app) as client:
+        missing = client.post(
+            "/connector-config-submit",
+            json={"config": {"password": "value"}},
+        )
+        malformed = client.post(
+            "/connector-config-submit",
+            headers={"Authorization": "Bearer invalid"},
+            json={"config": {}},
+        )
+        token = mint_action_token("validate", "source-example")
+        first = client.post(
+            "/connector-config-submit",
+            headers={"Authorization": f"Bearer {token}"},
+            json={"config": {}},
+        )
+        reused = client.post(
+            "/connector-config-submit",
+            headers={"Authorization": f"Bearer {token}"},
+            json={"config": {}},
+        )
+
+    assert missing.status_code == 401
+    assert malformed.status_code == 403
+    assert first.status_code == 200
+    assert reused.status_code == 403
+
+
+def test_endpoint_rejects_missing_config_without_echoing_secrets() -> None:
+    app = Starlette(routes=connector_config_submit_routes())
+    token = mint_action_token("validate", "source-example")
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/connector-config-submit",
+            headers={"Authorization": f"Bearer {token}"},
+            json={"not_config": "secret-value"},
+        )
+
+    assert response.status_code == 400
+    assert "secret-value" not in response.text
+
+
+def test_options_response_is_bodyless_and_keeps_cors_headers() -> None:
+    app = Starlette(routes=connector_config_submit_routes())
+
+    with TestClient(app) as client:
+        response = client.options(
+            "/connector-config-submit",
+            headers={
+                "Origin": "https://example.com",
+                "Access-Control-Request-Method": "POST",
+            },
+        )
+
+    assert response.status_code == 204
+    assert response.content == b""
+    assert response.headers["access-control-allow-origin"] == "*"
+    assert (
+        response.headers["access-control-allow-headers"]
+        == "Authorization, Content-Type"
+    )
+    assert response.headers["access-control-allow-methods"] == "POST, OPTIONS"
diff --git a/tests/unit_tests/test_mcp_connector_config_form.py b/tests/unit_tests/test_mcp_connector_config_form.py
new file mode 100644
index 000000000..7a999e660
--- /dev/null
+++ b/tests/unit_tests/test_mcp_connector_config_form.py
@@ -0,0 +1,381 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Tests for the connector configuration form MCP App."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+from starlette.applications import Starlette
+from starlette.routing import Route
+from starlette.testclient import TestClient
+
+from airbyte.exceptions import PyAirbyteInputError
+from airbyte.mcp import _config_submit
+from airbyte.mcp._config_submit import connector_config_submit_routes, mint_action_token
+from airbyte.mcp.interactive import _connector_config_form_ui as form
+
+
+SCHEMA = {
+    "type": "object",
+    "properties": {
+        "credentials": {
+            "type": "object",
+            "properties": {
+                "api_key": {"type": "string", "airbyte_secret": True},
+                "username": {"type": "string"},
+            },
+        },
+        "region": {"type": "string"},
+    },
+}
+
+
+def test_form_result_contains_schema_and_submit_data(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: SCHEMA
+    )
+
+    result = form.show_connector_config_form(
+        "source-example",
+        {"credentials": {"username": "user"}, "region": "us-east-1"},
+    )
+
+    content = result.content[0].text
+    structured = result.structured_content
+    assert "source-example" in content
+    assert "api_key" in content
+    assert structured["spec_schema"] == SCHEMA
+    assert structured["submit_endpoint"].endswith("/connector-config-submit")
+    assert structured["submit_token"]
+    assert structured["standalone_url"].endswith(
+        f"/connector-config-form?token={structured['submit_token']}"
+    )
+    assert structured["standalone_url"] in content
+    assert "intake_token" not in structured
+    assert "intake_endpoint" not in structured
+    assert "secret-value" not in content
+    assert "secret-value" not in json.dumps(structured)
+
+
+def test_form_rejects_secret_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: SCHEMA
+    )
+
+    with pytest.raises(PyAirbyteInputError, match="Secret values cannot"):
+        form.show_connector_config_form(
+            "source-example",
+            {"credentials": {"api_key": "secret-value"}},
+        )
+
+
+def test_form_rejects_secrets_in_array_defaults(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    schema = {
+        "type": "object",
+        "properties": {
+            "streams": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "token": {"type": "string", "airbyte_secret": True},
+                    },
+                },
+            },
+        },
+    }
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: schema
+    )
+
+    with pytest.raises(PyAirbyteInputError, match="Secret values cannot"):
+        form.show_connector_config_form(
+            "source-example",
+            {"streams": [{"token": "secret-value"}]},
+        )
+
+
+def test_submit_endpoint_preserves_server_path(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv(form.MCP_SERVER_URL_ENV, "https://example.com/cloud-mcp/")
+
+    assert (
+        form._submit_endpoint()
+        == "https://example.com/cloud-mcp/connector-config-submit"
+    )
+    assert (
+        form._standalone_endpoint("token")
+        == "https://example.com/cloud-mcp/connector-config-form?token=token"
+    )
+    assert form._server_origin() == "https://example.com"
+
+
+def test_server_origin_rejects_insecure_nonlocal_url(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv(form.MCP_SERVER_URL_ENV, "http://example.com/cloud-mcp")
+
+    with pytest.raises(PyAirbyteInputError, match="HTTPS"):
+        form._server_origin()
+
+
+@pytest.mark.parametrize(
+    "server_url",
+    ["//attacker.example", "ftp://example.com", "example.com"],
+)
+def test_server_origin_rejects_invalid_url(
+    monkeypatch: pytest.MonkeyPatch,
+    server_url: str,
+) -> None:
+    monkeypatch.setenv(form.MCP_SERVER_URL_ENV, server_url)
+
+    with pytest.raises(PyAirbyteInputError, match="valid HTTP"):
+        form._server_origin()
+
+
+def test_server_origin_allows_local_http_url(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv(form.MCP_SERVER_URL_ENV, "http://localhost:8080/cloud-mcp")
+
+    assert form._server_origin() == "http://localhost:8080"
+    assert (
+        form._submit_endpoint()
+        == "http://localhost:8080/cloud-mcp/connector-config-submit"
+    )
+
+
+def test_form_selects_create_action_with_cloud_credentials(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv(form.CLOUD_BEARER_TOKEN_ENV_VAR, "cloud-secret")
+    monkeypatch.setenv(form.CLOUD_WORKSPACE_ID_ENV_VAR, "workspace-id")
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: SCHEMA
+    )
+
+    result = form.show_connector_config_form("source-example")
+
+    assert result.structured_content["action"] == "create"
+    assert "cloud-secret" not in json.dumps(result.structured_content)
+
+
+def test_form_selects_update_action_with_source_id(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv(form.CLOUD_BEARER_TOKEN_ENV_VAR, "cloud-token")
+    monkeypatch.setenv(form.CLOUD_WORKSPACE_ID_ENV_VAR, "workspace-id")
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: SCHEMA
+    )
+
+    result = form.show_connector_config_form("source-example", source_id="source-id")
+
+    assert result.structured_content["action"] == "update"
+
+
+def test_form_rejects_update_without_cloud_credentials_or_workspace(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv(form.CLOUD_BEARER_TOKEN_ENV_VAR, raising=False)
+    monkeypatch.delenv(form.CLOUD_CLIENT_ID_ENV_VAR, raising=False)
+    monkeypatch.delenv(form.CLOUD_CLIENT_SECRET_ENV_VAR, raising=False)
+    monkeypatch.delenv(form.CLOUD_WORKSPACE_ID_ENV_VAR, raising=False)
+    monkeypatch.setattr(form, "_resolve_transport_bearer_token", lambda: "")
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: SCHEMA
+    )
+
+    with pytest.raises(
+        PyAirbyteInputError,
+        match="Cloud credentials and a workspace ID are required",
+    ):
+        form.show_connector_config_form("source-example", source_id="source-id")
+
+
+def test_form_selects_validate_without_cloud_credentials(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv(form.CLOUD_BEARER_TOKEN_ENV_VAR, raising=False)
+    monkeypatch.delenv(form.CLOUD_CLIENT_ID_ENV_VAR, raising=False)
+    monkeypatch.delenv(form.CLOUD_CLIENT_SECRET_ENV_VAR, raising=False)
+    monkeypatch.delenv(form.CLOUD_WORKSPACE_ID_ENV_VAR, raising=False)
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: SCHEMA
+    )
+
+    result = form.show_connector_config_form("source-example")
+
+    assert result.structured_content["action"] == "validate"
+
+
+def test_form_blocks_prototype_pollution_paths() -> None:
+    html = form.connector_config_form_resource()
+
+    assert '["__proto__", "constructor", "prototype"]' in html
+    assert "Array.isArray(target[key])" in html
+
+
+def test_form_handles_one_of_schema_without_plaintext_input(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    schema = {
+        "type": "object",
+        "properties": {
+            "credentials": {
+                "type": "object",
+                "oneOf": [
+                    {
+                        "properties": {
+                            "api_key": {"type": "string", "airbyte_secret": True},
+                        }
+                    }
+                ],
+            }
+        },
+    }
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: schema
+    )
+
+    result = form.show_connector_config_form("source-example")
+
+    assert result.structured_content["secret_fields"] == ["credentials.api_key"]
+    html = form.connector_config_form_resource()
+    assert "Complex authentication objects are not supported by this form." in html
+    assert "Array.isArray(child[key])" in html
+    assert "state.result.submit_endpoint" in html
+    assert "JSON.stringify({config})" in html
+    assert "appCapabilities: {}" in html
+    assert 'appInfo: {name: "airbyte-form", version: "0.1"}' in html
+    assert "message.id === 1 && message.result" in html
+    assert 'method: "ui/notifications/initialized"' in html
+    assert 'method: "ui/notifications/size-changed"' in html
+    assert 'method: "ui/message"' in html
+    assert "Notify agent" in html
+    assert "Math.min(contentHeight + 20, 600)" in html
+
+
+def test_connector_form_resource_includes_csp_metadata(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from fastmcp_extensions import mcp_server
+
+    from airbyte.mcp import interactive
+
+    monkeypatch.setenv(form.MCP_SERVER_URL_ENV, "https://late.example.com/cloud-mcp")
+    app = mcp_server(name="test")
+    interactive.register_interactive_tools(app)
+
+    resource = next(
+        item
+        for item in asyncio.run(app.list_resources())
+        if str(item.uri) == form.CONNECTOR_FORM_RESOURCE_URI
+    )
+
+    assert resource.meta is not None
+    assert resource.meta["ui"]["csp"]["connectDomains"] == ["https://late.example.com"]
+
+    provider = getattr(app, "_local_provider")
+    tool = provider._components["tool:show_connector_config_form@"]  # noqa: SLF001
+    assert tool.meta is not None
+    assert tool.meta["ui"]["csp"]["connectDomains"] == ["https://late.example.com"]
+
+
+class _StandaloneSource:
+    def set_config(self, config: dict[str, object], *, validate: bool = True) -> None:
+        assert validate
+
+
+def _standalone_test_app() -> Starlette:
+    return Starlette(
+        routes=[
+            Route(
+                "/connector-config-form",
+                form.connector_config_form_endpoint,
+                methods=["GET"],
+            ),
+            *connector_config_submit_routes(),
+        ]
+    )
+
+
+def test_standalone_form_renders_and_get_does_not_consume_token(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: SCHEMA
+    )
+    monkeypatch.setattr(
+        _config_submit, "get_source", lambda *args, **kwargs: _StandaloneSource()
+    )
+    token = mint_action_token(
+        "validate",
+        "source-example",
+        non_secret_defaults={"region": "us-east-1"},
+        bearer_token="bearer-secret",
+        client_id="client-id",
+        client_secret="client-secret",
+    )
+
+    with TestClient(_standalone_test_app()) as client:
+        response = client.get(f"/connector-config-form?token={token}")
+        again = client.get(f"/connector-config-form?token={token}")
+        submitted = client.post(
+            "/connector-config-submit",
+            headers={"Authorization": f"Bearer {token}"},
+            json={"config": {"region": "us-east-1"}},
+        )
+
+    assert response.status_code == 200
+    assert again.status_code == 200
+    assert "source-example" in response.text
+    assert '"region":"us-east-1"' in response.text
+    assert "bearer-secret" not in response.text
+    assert "client-secret" not in response.text
+    assert submitted.status_code == 200
+
+
+@pytest.mark.parametrize("query", ["", "?token=invalid"])
+def test_standalone_form_rejects_invalid_or_missing_token(query: str) -> None:
+    with TestClient(_standalone_test_app()) as client:
+        response = client.get(f"/connector-config-form{query}")
+
+    assert response.status_code == 403
+    assert "This link is invalid or has expired." in response.text
+
+
+def test_standalone_form_rejects_expired_token(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    token = mint_action_token("validate", "source-example", ttl_seconds=1)
+    monkeypatch.setattr(_config_submit.time, "time", lambda: 2_000_000_000)
+
+    with TestClient(_standalone_test_app()) as client:
+        response = client.get(f"/connector-config-form?token={token}")
+
+    assert response.status_code == 403
+    assert "This link is invalid or has expired." in response.text
+
+
+def test_standalone_form_escapes_script_terminators(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        form, "get_connector_spec_from_registry", lambda *args, **kwargs: SCHEMA
+    )
+    token = mint_action_token(
+        "validate",
+        "source-example",
+        non_secret_defaults={"region": "</script><script>alert(1)</script>"},
+    )
+
+    with TestClient(_standalone_test_app()) as client:
+        response = client.get(f"/connector-config-form?token={token}")
+
+    assert response.status_code == 200
+    assert "</script><script>alert(1)</script>" not in response.text
+    assert "<\\/script>" in response.text