From 3051ff3d82ba7c1e02b51061921adfbd8f59b618 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 14 Oct 2026 04:22:06 +0000
Subject: [PATCH 1/8] feat(mcp)!: split manifest_yaml into manifest_yaml_text
 and manifest_yaml_file

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/_arg_resolvers.py                 |  33 +++---
 airbyte/mcp/cloud.py                          |  83 ++++++++++----
 tests/unit_tests/fixtures/arg_trace_plan.json |   6 +-
 tests/unit_tests/test_mcp_arg_trace_parity.py |  14 ++-
 .../unit_tests/test_mcp_trusted_execution.py  | 102 ++++++++++++------
 5 files changed, 164 insertions(+), 74 deletions(-)

diff --git a/airbyte/mcp/_arg_resolvers.py b/airbyte/mcp/_arg_resolvers.py
index 52f600937..a1d822f8f 100644
--- a/airbyte/mcp/_arg_resolvers.py
+++ b/airbyte/mcp/_arg_resolvers.py
@@ -16,7 +16,6 @@
 from airbyte.constants import SECRETS_HYDRATION_PREFIX
 from airbyte.exceptions import AirbyteLibInputError
 from airbyte.mcp._guards import (
-    is_trusted_execution_enabled,
     raise_if_untrusted_execution_context,
 )
 from airbyte.secrets.hydration import deep_update, detect_hardcoded_secrets
@@ -92,28 +91,24 @@ def resolve_list_of_strings(value: str | list[str] | set[str] | None) -> list[st
     return [item.strip() for item in value.split(",") if item.strip()]
 
 
-def resolve_manifest_yaml(manifest_yaml: str | Path | None) -> str | Path | None:
-    """Resolve inline YAML or gate local manifest paths behind trusted execution.
-
-    A local manifest path is a trusted-machine input and is rejected when trusted execution is
-    disabled.
-    """
-    if manifest_yaml is None:
-        return None
+def resolve_manifest_yaml(
+    *,
+    manifest_yaml_text: str | None,
+    manifest_yaml_file: str | None,
+) -> str | Path | None:
+    """Resolve manifest text or gate an explicit local manifest path behind trusted execution."""
+    if manifest_yaml_text is not None and manifest_yaml_file is not None:
+        raise AirbyteLibInputError(
+            message="Provide only one of `manifest_yaml_text` or `manifest_yaml_file`."
+        )
 
-    if isinstance(manifest_yaml, Path):
+    if manifest_yaml_file is not None:
         raise_if_untrusted_execution_context(
-            "Reading a connector manifest from a local file (`manifest_yaml` path)"
+            "Reading a connector manifest from a local file (`manifest_yaml_file`)"
         )
-        return manifest_yaml
-
-    if "\n" in manifest_yaml:
-        return manifest_yaml
-
-    if is_trusted_execution_enabled():
-        return Path(manifest_yaml)
+        return Path(manifest_yaml_file)
 
-    return manifest_yaml
+    return manifest_yaml_text
 
 
 def resolve_connector_config(  # noqa: PLR0912
diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index e9bfb4ffd..2765d52ae 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -12,7 +12,6 @@
 
 from collections.abc import Callable
 from http import HTTPStatus
-from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, Any, Final, Literal, TypeVar, cast
 
 import requests
@@ -88,6 +87,7 @@
     render_connector_docs_result,
 )
 from airbyte.mcp._error_handling import format_user_facing_error, redact_agent_text
+from airbyte.mcp._guards import is_trusted_execution_enabled
 from airbyte.mcp._scope import record_default_workspace, record_resolved_organization
 from airbyte.mcp._tool_utils import (
     AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET,
@@ -2970,7 +2970,7 @@ def _get_custom_source_definition_description(
 
 
 @mcp_tool()
-def publish_custom_source_definition(
+def publish_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MCP tool schema.
     ctx: Context,
     name: Annotated[
         str,
@@ -2986,13 +2986,24 @@ def publish_custom_source_definition(
         ),
         TraceArg.OMIT,
     ],
-    manifest_yaml: Annotated[
-        str | Path | None,
+    manifest_yaml_text: Annotated[
+        str | None,
+        Field(
+            description=(
+                "Full YAML contents of the declarative (low-code) connector manifest: the "
+                "file's text, not a filename or path. If the manifest is in a file, read the "
+                "file and pass its full contents here."
+            ),
+            default=None,
+        ),
+    ] = None,
+    manifest_yaml_file: Annotated[
+        str | None,
         Field(
             description=(
-                "The Low-code CDK manifest as inline YAML or a file path. File paths are "
-                "honored only when trusted execution is enabled; otherwise, the value is "
-                "treated as inline YAML. Required for YAML connectors."
+                "Path to a manifest YAML file on the machine running this MCP server. Use "
+                "instead of `manifest_yaml_text`; provide only one. Available only when "
+                "trusted execution is enabled."
             ),
             default=None,
         ),
@@ -3044,7 +3055,17 @@ def publish_custom_source_definition(
     Note: Only YAML (declarative) connectors are currently supported.
     Docker-based custom sources are not yet available.
     """
-    processed_manifest = resolve_manifest_yaml(manifest_yaml)
+    processed_manifest = resolve_manifest_yaml(
+        manifest_yaml_text=manifest_yaml_text,
+        manifest_yaml_file=manifest_yaml_file,
+    )
+    if processed_manifest is None:
+        raise AirbyteLibInputError(
+            message=(
+                "Provide the manifest in `manifest_yaml_text` (full YAML contents) or "
+                "`manifest_yaml_file` (local path; trusted execution only)."
+            ),
+        )
 
     # Resolve testing values from inline config and/or secret
     testing_values_dict: dict[str, Any] | None = None
@@ -3223,24 +3244,36 @@ def get_connector_builder_draft_manifest(
 @mcp_tool(
     destructive=True,
 )
-def update_custom_source_definition(
+def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MCP tool schema.
     ctx: Context,
     definition_id: Annotated[
         str,
         Field(description="The ID of the definition to update."),
     ],
-    manifest_yaml: Annotated[
-        str | Path | None,
+    *,
+    manifest_yaml_text: Annotated[
+        str | None,
         Field(
             description=(
-                "New manifest as inline YAML or a file path. File paths are honored only when "
-                "trusted execution is enabled; otherwise, the value is treated as inline YAML. "
-                "Optional; omit to update only testing values."
+                "New manifest. Full YAML contents of the declarative (low-code) connector "
+                "manifest: the file's text, not a filename or path. If the manifest is in a "
+                "file, read the file and pass its full contents here. Omit to leave the "
+                "manifest unchanged."
+            ),
+            default=None,
+        ),
+    ] = None,
+    manifest_yaml_file: Annotated[
+        str | None,
+        Field(
+            description=(
+                "Path to a manifest YAML file on the machine running this MCP server. Use "
+                "instead of `manifest_yaml_text`; provide only one. Available only when "
+                "trusted execution is enabled. Omit to leave the manifest unchanged."
             ),
             default=None,
         ),
     ] = None,
-    *,
     workspace_id: Annotated[
         str | None,
         Field(
@@ -3300,8 +3333,8 @@ def update_custom_source_definition(
 
     Updates the definition name, manifest, and/or testing values for an existing custom source
     definition. Renaming also updates the Connector Builder project name.
-    At least one of name, manifest_yaml, testing_values, or testing_values_secret_name must be
-    provided.
+    At least one of name, manifest_yaml_text, manifest_yaml_file, testing_values, or
+    testing_values_secret_name must be provided.
     """
     check_guid_created_in_session(definition_id)
 
@@ -3309,13 +3342,14 @@ def update_custom_source_definition(
 
     if (
         name is None
-        and manifest_yaml is None
+        and manifest_yaml_text is None
+        and manifest_yaml_file is None
         and testing_values is None
         and testing_values_secret_name is None
     ):
         raise AirbyteLibInputError(
             message=(
-                "At least one of name, manifest_yaml, testing_values, or "
+                "At least one of name, manifest_yaml_text, manifest_yaml_file, testing_values, or "
                 "testing_values_secret_name must be provided to update a custom source definition."
             ),
             context={
@@ -3324,7 +3358,10 @@ def update_custom_source_definition(
             },
         )
 
-    processed_manifest = resolve_manifest_yaml(manifest_yaml)
+    processed_manifest = resolve_manifest_yaml(
+        manifest_yaml_text=manifest_yaml_text,
+        manifest_yaml_file=manifest_yaml_file,
+    )
 
     # Resolve testing values from inline config and/or secret
     testing_values_dict: dict[str, Any] | None = None
@@ -4035,11 +4072,13 @@ def register_cloud_tools(app: FastMCP) -> None:
     Args:
         app: FastMCP application instance
     """
-    exclude_args = ["workspace_id"] if AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET else None
+    exclude_args = ["workspace_id"] if AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET else []
+    if not is_trusted_execution_enabled():
+        exclude_args.append("manifest_yaml_file")
     if exclude_args:
         _add_defaults_for_exclude_args(exclude_args)
     register_mcp_tools(
         app,
         mcp_module=__name__,
-        exclude_args=exclude_args,
+        exclude_args=exclude_args or None,
     )
diff --git a/tests/unit_tests/fixtures/arg_trace_plan.json b/tests/unit_tests/fixtures/arg_trace_plan.json
index 90abfd011..5e31e3781 100644
--- a/tests/unit_tests/fixtures/arg_trace_plan.json
+++ b/tests/unit_tests/fixtures/arg_trace_plan.json
@@ -230,7 +230,8 @@
     "workspace_id": "omit"
   },
   "publish_custom_source_definition": {
-    "manifest_yaml": "presence",
+    "manifest_yaml_file": "hash",
+    "manifest_yaml_text": "hash",
     "name": "fingerprint",
     "pre_validate": "value allowed=[false,true]",
     "testing_values": "presence",
@@ -335,7 +336,8 @@
   },
   "update_custom_source_definition": {
     "definition_id": "hash",
-    "manifest_yaml": "presence",
+    "manifest_yaml_file": "hash",
+    "manifest_yaml_text": "hash",
     "name": "fingerprint",
     "pre_validate": "value allowed=[false,true]",
     "testing_values": "presence",
diff --git a/tests/unit_tests/test_mcp_arg_trace_parity.py b/tests/unit_tests/test_mcp_arg_trace_parity.py
index 5302276c6..c074876bc 100644
--- a/tests/unit_tests/test_mcp_arg_trace_parity.py
+++ b/tests/unit_tests/test_mcp_arg_trace_parity.py
@@ -68,8 +68,20 @@ def _plan() -> dict[str, dict[str, str]]:
 def test_inventory_matches_fixture() -> None:
     plan = _plan()
     if os.environ.get("UPDATE_ARG_TRACE_FIXTURE"):
+        assert all(
+            "manifest_yaml_file" in plan[name]
+            for name in (
+                "publish_custom_source_definition",
+                "update_custom_source_definition",
+            )
+        ), "Enable trusted execution when regenerating the fixture."
         FIXTURE.write_text(json.dumps(plan, indent=2, sort_keys=True) + "\n")
-    assert plan == json.loads(FIXTURE.read_text())
+
+    expected = json.loads(FIXTURE.read_text())
+    for name in ("publish_custom_source_definition", "update_custom_source_definition"):
+        if "manifest_yaml_file" not in plan[name]:
+            expected[name].pop("manifest_yaml_file", None)
+    assert plan == expected
 
 
 @pytest.fixture
diff --git a/tests/unit_tests/test_mcp_trusted_execution.py b/tests/unit_tests/test_mcp_trusted_execution.py
index 75ea06a5d..cba7a916d 100644
--- a/tests/unit_tests/test_mcp_trusted_execution.py
+++ b/tests/unit_tests/test_mcp_trusted_execution.py
@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 import os
@@ -10,12 +11,14 @@
 from typing import TYPE_CHECKING, Callable
 
 import pytest
+from fastmcp import Client, FastMCP
+
 from airbyte.constants import MCP_TRUSTED_EXECUTION_ENV_VAR
 from airbyte.exceptions import (
-    AirbyteTrustedExecutionRequiredError,
     AirbyteLibInputError,
+    AirbyteTrustedExecutionRequiredError,
 )
-from airbyte.mcp import local
+from airbyte.mcp import cloud, local
 from airbyte.mcp._arg_resolvers import resolve_connector_config, resolve_manifest_yaml
 from airbyte.mcp._guards import (
     is_trusted_execution_enabled,
@@ -165,18 +168,21 @@ def test_resolve_connector_config_allows_when_trusted(
     assert resolve_connector_config(**kwargs) == expected
 
 
-def test_resolve_manifest_yaml_rejects_path_when_untrusted(
+def test_resolve_manifest_yaml_rejects_file_when_untrusted(
     monkeypatch: MonkeyPatch,
     tmp_path: Path,
 ) -> None:
-    """A `Path` manifest input requires trusted execution."""
+    """A local manifest file input requires trusted execution."""
     _set_trusted(monkeypatch, enabled=False)
     with pytest.raises(AirbyteTrustedExecutionRequiredError):
-        resolve_manifest_yaml(tmp_path / "manifest.yaml")
+        resolve_manifest_yaml(
+            manifest_yaml_text=None,
+            manifest_yaml_file=str(tmp_path / "manifest.yaml"),
+        )
 
 
 @pytest.mark.parametrize(
-    "manifest_yaml",
+    "manifest_yaml_text",
     [
         pytest.param("manifest.yaml", id="single-line-text"),
         pytest.param(
@@ -185,56 +191,92 @@ def test_resolve_manifest_yaml_rejects_path_when_untrusted(
         pytest.param("version: 0.1.0\ntype: DeclarativeSource", id="multiline-yaml"),
     ],
 )
-def test_resolve_manifest_yaml_preserves_inline_values_when_untrusted(
+@pytest.mark.parametrize("enabled", [False, True], ids=["untrusted", "trusted"])
+def test_resolve_manifest_yaml_preserves_text(
     monkeypatch: MonkeyPatch,
-    manifest_yaml: str,
+    manifest_yaml_text: str,
+    enabled: bool,
 ) -> None:
-    """Untrusted execution preserves manifest strings as inline YAML."""
-    _set_trusted(monkeypatch, enabled=False)
-    resolved = resolve_manifest_yaml(manifest_yaml)
-    assert isinstance(resolved, str)
-    assert resolved == manifest_yaml
+    """Manifest text is never reinterpreted as a path."""
+    _set_trusted(monkeypatch, enabled=enabled)
+    resolved = resolve_manifest_yaml(
+        manifest_yaml_text=manifest_yaml_text,
+        manifest_yaml_file=None,
+    )
+    assert resolved == manifest_yaml_text
 
 
-def test_resolve_manifest_yaml_preserves_existing_path_when_trusted(
+def test_resolve_manifest_yaml_returns_file_path_when_trusted(
     monkeypatch: MonkeyPatch,
     tmp_path: Path,
 ) -> None:
-    """A `Path` manifest input is returned when trusted execution is enabled."""
+    """A local manifest file is resolved to a `Path` when trusted execution is enabled."""
     _set_trusted(monkeypatch, enabled=True)
     manifest_path = tmp_path / "manifest.yaml"
-    assert resolve_manifest_yaml(manifest_path) is manifest_path
+    assert (
+        resolve_manifest_yaml(
+            manifest_yaml_text=None,
+            manifest_yaml_file=str(manifest_path),
+        )
+        == manifest_path
+    )
 
 
-def test_resolve_manifest_yaml_converts_single_line_string_when_trusted(
+def test_resolve_manifest_yaml_rejects_both_inputs(
     monkeypatch: MonkeyPatch,
 ) -> None:
-    """A single-line manifest string is treated as a file path when trusted."""
+    """Manifest text and a local file path cannot be supplied together."""
     _set_trusted(monkeypatch, enabled=True)
-    assert resolve_manifest_yaml("manifest.yaml") == Path("manifest.yaml")
+    with pytest.raises(
+        AirbyteLibInputError,
+        match="Provide only one of `manifest_yaml_text` or `manifest_yaml_file`\\.",
+    ):
+        resolve_manifest_yaml(
+            manifest_yaml_text="version: 0.1.0",
+            manifest_yaml_file="manifest.yaml",
+        )
 
 
 @pytest.mark.parametrize("enabled", [False, True], ids=["untrusted", "trusted"])
-def test_resolve_manifest_yaml_preserves_none(
+def test_resolve_manifest_yaml_returns_none_when_inputs_omitted(
     monkeypatch: MonkeyPatch,
     enabled: bool,
 ) -> None:
     """A missing manifest stays missing in either execution mode."""
     _set_trusted(monkeypatch, enabled=enabled)
-    assert resolve_manifest_yaml(None) is None
+    assert (
+        resolve_manifest_yaml(
+            manifest_yaml_text=None,
+            manifest_yaml_file=None,
+        )
+        is None
+    )
 
 
-def test_resolve_manifest_yaml_does_not_read_file_when_untrusted(
+def test_manifest_file_schema_depends_on_trusted_execution(
     monkeypatch: MonkeyPatch,
-    tmp_path: Path,
 ) -> None:
-    """A path string remains inline YAML when trusted execution is disabled."""
-    _set_trusted(monkeypatch, enabled=False)
-    manifest_path = tmp_path / "manifest.yaml"
-    manifest_path.write_text("version: 0.1.0\ntype: DeclarativeSource\n")
-    resolved = resolve_manifest_yaml(str(manifest_path))
-    assert isinstance(resolved, str)
-    assert resolved == str(manifest_path)
+    """The file argument is hidden from untrusted schemas and exposed to trusted callers."""
+    for enabled in (False, True):
+        _set_trusted(monkeypatch, enabled=enabled)
+        app = FastMCP(f"manifest-schema-{enabled}")
+        cloud.register_cloud_tools(app)
+
+        async def list_tools() -> dict[str, object]:
+            async with Client(app) as client:
+                return {tool.name: tool for tool in await client.list_tools()}
+
+        tools = asyncio.run(list_tools())
+
+        for tool_name in (
+            "publish_custom_source_definition",
+            "update_custom_source_definition",
+        ):
+            tool = tools[tool_name]
+            properties = tool.input_schema["properties"]  # type: ignore[attr-defined]
+            assert "manifest_yaml_text" in properties
+            assert ("manifest_yaml_file" in properties) is enabled
+            assert "manifest_yaml" not in properties
 
 
 _UNTRUSTED_LOCAL_HELPERS: list[Callable[[], object]] = [

From 6c1824602cdaf0cfb45933bb65a50248911238a4 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 14 Oct 2026 04:27:52 +0000
Subject: [PATCH 2/8] docs(mcp): drop redundant trusted-execution note from
 manifest_yaml_file description

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/cloud.py | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index 2765d52ae..729609671 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -3002,8 +3002,7 @@ def publish_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror M
         Field(
             description=(
                 "Path to a manifest YAML file on the machine running this MCP server. Use "
-                "instead of `manifest_yaml_text`; provide only one. Available only when "
-                "trusted execution is enabled."
+                "instead of `manifest_yaml_text`; provide only one."
             ),
             default=None,
         ),
@@ -3268,8 +3267,8 @@ def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MC
         Field(
             description=(
                 "Path to a manifest YAML file on the machine running this MCP server. Use "
-                "instead of `manifest_yaml_text`; provide only one. Available only when "
-                "trusted execution is enabled. Omit to leave the manifest unchanged."
+                "instead of `manifest_yaml_text`; provide only one. Omit to leave the "
+                "manifest unchanged."
             ),
             default=None,
         ),

From d87cd3d5a0b9616061b9596a4edf5b63ff330225 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 14 Oct 2026 04:39:06 +0000
Subject: [PATCH 3/8] fix(mcp): make manifest arg guidance mode-aware

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/cloud.py                          | 80 +++++++++++++++----
 .../unit_tests/test_mcp_trusted_execution.py  | 33 ++++++++
 2 files changed, 96 insertions(+), 17 deletions(-)

diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index 729609671..d85060d0a 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -10,12 +10,14 @@
 # tool / helper definitions as a redundant "API Documentation" list.
 __all__: list[str] = []
 
-from collections.abc import Callable
+from collections.abc import Callable, Mapping, Sequence
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Annotated, Any, Final, Literal, TypeVar, cast
 
 import requests
 from fastmcp import Context, FastMCP
+from fastmcp.server.transforms import GetToolNext, Transform, VersionSpec
+from fastmcp.tools.base import Tool
 from fastmcp_extensions import TraceArg, get_mcp_config, register_mcp_tools
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -118,6 +120,9 @@
     "rename it themselves first."
 )
 WORKSPACE_ID_TIP_TEXT = "Optional; defaults to the session's workspace."
+MANIFEST_YAML_SOURCE_HINT = (
+    "Provide the manifest as either `manifest_yaml_text` or `manifest_yaml_file`, not both."
+)
 SKILL_DOCS_SECTION_HINT = """\
 Skill doc sections for direct API queries and actions are named with the pattern
 `actions.<entity_type>.<action>`, matching the `entity_type` and `action` arguments of the
@@ -3000,10 +3005,7 @@ def publish_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror M
     manifest_yaml_file: Annotated[
         str | None,
         Field(
-            description=(
-                "Path to a manifest YAML file on the machine running this MCP server. Use "
-                "instead of `manifest_yaml_text`; provide only one."
-            ),
+            description="Path to a manifest YAML file on the machine running this MCP server.",
             default=None,
         ),
     ] = None,
@@ -3059,11 +3061,14 @@ def publish_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror M
         manifest_yaml_file=manifest_yaml_file,
     )
     if processed_manifest is None:
+        error_message = (
+            "Provide the manifest in `manifest_yaml_text` (full YAML contents) or "
+            "`manifest_yaml_file` (local path)."
+            if is_trusted_execution_enabled()
+            else "Provide the manifest's full YAML contents in `manifest_yaml_text`."
+        )
         raise AirbyteLibInputError(
-            message=(
-                "Provide the manifest in `manifest_yaml_text` (full YAML contents) or "
-                "`manifest_yaml_file` (local path; trusted execution only)."
-            ),
+            message=error_message,
         )
 
     # Resolve testing values from inline config and/or secret
@@ -3266,9 +3271,8 @@ def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MC
         str | None,
         Field(
             description=(
-                "Path to a manifest YAML file on the machine running this MCP server. Use "
-                "instead of `manifest_yaml_text`; provide only one. Omit to leave the "
-                "manifest unchanged."
+                "Path to a manifest YAML file on the machine running this MCP server. "
+                "Omit to leave the manifest unchanged."
             ),
             default=None,
         ),
@@ -3332,8 +3336,7 @@ def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MC
 
     Updates the definition name, manifest, and/or testing values for an existing custom source
     definition. Renaming also updates the Connector Builder project name.
-    At least one of name, manifest_yaml_text, manifest_yaml_file, testing_values, or
-    testing_values_secret_name must be provided.
+    Provide at least one value to update.
     """
     check_guid_created_in_session(definition_id)
 
@@ -3346,10 +3349,14 @@ def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MC
         and testing_values is None
         and testing_values_secret_name is None
     ):
+        required_args = ["name", "manifest_yaml_text"]
+        if is_trusted_execution_enabled():
+            required_args.append("manifest_yaml_file")
+        required_args.extend(("testing_values", "testing_values_secret_name"))
         raise AirbyteLibInputError(
             message=(
-                "At least one of name, manifest_yaml_text, manifest_yaml_file, testing_values, or "
-                "testing_values_secret_name must be provided to update a custom source definition."
+                f"At least one of {', '.join(required_args)} must be provided to update "
+                "a custom source definition."
             ),
             context={
                 "definition_id": definition_id,
@@ -4065,15 +4072,52 @@ def _add_defaults_for_exclude_args(
             func.__signature__ = sig.replace(parameters=new_params)  # type: ignore[attr-defined]
 
 
+class _ToolDescriptionSuffix(Transform):
+    """Append mode-specific guidance to selected tool descriptions."""
+
+    def __init__(self, suffixes: Mapping[str, str]) -> None:
+        self._suffixes = suffixes
+
+    def _apply(self, tool: Tool) -> Tool:
+        if tool.name not in self._suffixes:
+            return tool
+        suffix = self._suffixes[tool.name]
+        return tool.model_copy(
+            update={"description": f"{tool.description or ''}\n\n{suffix}".strip()}
+        )
+
+    async def list_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]:
+        return [self._apply(tool) for tool in tools]
+
+    async def get_tool(
+        self,
+        name: str,
+        call_next: GetToolNext,
+        *,
+        version: VersionSpec | None = None,
+    ) -> Tool | None:
+        tool = await call_next(name, version=version)
+        return self._apply(tool) if tool is not None else None
+
+
 def register_cloud_tools(app: FastMCP) -> None:
     """Register cloud tools with the FastMCP app.
 
     Args:
         app: FastMCP application instance
     """
+    trusted = is_trusted_execution_enabled()
     exclude_args = ["workspace_id"] if AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET else []
-    if not is_trusted_execution_enabled():
+    description_suffix: _ToolDescriptionSuffix | None = None
+    if not trusted:
         exclude_args.append("manifest_yaml_file")
+    else:
+        description_suffix = _ToolDescriptionSuffix(
+            {
+                "publish_custom_source_definition": MANIFEST_YAML_SOURCE_HINT,
+                "update_custom_source_definition": MANIFEST_YAML_SOURCE_HINT,
+            }
+        )
     if exclude_args:
         _add_defaults_for_exclude_args(exclude_args)
     register_mcp_tools(
@@ -4081,3 +4125,5 @@ def register_cloud_tools(app: FastMCP) -> None:
         mcp_module=__name__,
         exclude_args=exclude_args or None,
     )
+    if description_suffix is not None:
+        app.add_transform(description_suffix)
diff --git a/tests/unit_tests/test_mcp_trusted_execution.py b/tests/unit_tests/test_mcp_trusted_execution.py
index cba7a916d..1dafff09a 100644
--- a/tests/unit_tests/test_mcp_trusted_execution.py
+++ b/tests/unit_tests/test_mcp_trusted_execution.py
@@ -277,6 +277,39 @@ async def list_tools() -> dict[str, object]:
             assert "manifest_yaml_text" in properties
             assert ("manifest_yaml_file" in properties) is enabled
             assert "manifest_yaml" not in properties
+            assert (
+                cloud.MANIFEST_YAML_SOURCE_HINT
+                in (getattr(tool, "description", None) or "")
+            ) is enabled
+            if enabled:
+                assert (
+                    "manifest_yaml_text"
+                    not in properties["manifest_yaml_file"]["description"]
+                )
+
+
+@pytest.mark.parametrize("enabled", [False, True], ids=["untrusted", "trusted"])
+def test_publish_missing_manifest_error_depends_on_trusted_execution(
+    monkeypatch: MonkeyPatch,
+    enabled: bool,
+) -> None:
+    """The missing-manifest error only mentions the file argument when it is available."""
+    _set_trusted(monkeypatch, enabled=enabled)
+    app = FastMCP(f"manifest-error-{enabled}")
+    cloud.register_cloud_tools(app)
+
+    async def call_publish() -> object:
+        async with Client(app) as client:
+            return await client.call_tool(
+                "publish_custom_source_definition",
+                {"name": "x"},
+                raise_on_error=False,
+            )
+
+    result = asyncio.run(call_publish())
+    assert result.is_error  # type: ignore[attr-defined]
+    error_text = result.content[0].text  # type: ignore[attr-defined]
+    assert ("manifest_yaml_file" in error_text) is enabled
 
 
 _UNTRUSTED_LOCAL_HELPERS: list[Callable[[], object]] = [

From 7b00db94486bd3a9f576c3ded48b8092cc67e6d7 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 14 Oct 2026 04:51:25 +0000
Subject: [PATCH 4/8] fix(mcp): drop misleading omit guidance from update
 manifest args

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/cloud.py | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index d85060d0a..fe19e54c4 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -3259,10 +3259,9 @@ def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MC
         str | None,
         Field(
             description=(
-                "New manifest. Full YAML contents of the declarative (low-code) connector "
-                "manifest: the file's text, not a filename or path. If the manifest is in a "
-                "file, read the file and pass its full contents here. Omit to leave the "
-                "manifest unchanged."
+                "Optional. New manifest. Full YAML contents of the declarative (low-code) "
+                "connector manifest: the file's text, not a filename or path. If the manifest "
+                "is in a file, read the file and pass its full contents here."
             ),
             default=None,
         ),
@@ -3271,8 +3270,7 @@ def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MC
         str | None,
         Field(
             description=(
-                "Path to a manifest YAML file on the machine running this MCP server. "
-                "Omit to leave the manifest unchanged."
+                "Optional. Path to a manifest YAML file on the machine running this MCP server."
             ),
             default=None,
         ),

From 70c17eff9a2733182d098765b45ae1897682038c Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 14 Oct 2026 04:55:03 +0000
Subject: [PATCH 5/8] fix(mcp): keep manifest_yaml_text description
 self-contained

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/cloud.py | 13 +++++++------
 1 file changed, 7 insertions(+), 6 deletions(-)

diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index fe19e54c4..70d383b97 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -2996,8 +2996,7 @@ def publish_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror M
         Field(
             description=(
                 "Full YAML contents of the declarative (low-code) connector manifest: the "
-                "file's text, not a filename or path. If the manifest is in a file, read the "
-                "file and pass its full contents here."
+                "file's text, not a filename or path."
             ),
             default=None,
         ),
@@ -3005,7 +3004,9 @@ def publish_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror M
     manifest_yaml_file: Annotated[
         str | None,
         Field(
-            description="Path to a manifest YAML file on the machine running this MCP server.",
+            description=(
+                "Absolute path to a manifest YAML file on the machine running this MCP server."
+            ),
             default=None,
         ),
     ] = None,
@@ -3260,8 +3261,7 @@ def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MC
         Field(
             description=(
                 "Optional. New manifest. Full YAML contents of the declarative (low-code) "
-                "connector manifest: the file's text, not a filename or path. If the manifest "
-                "is in a file, read the file and pass its full contents here."
+                "connector manifest: the file's text, not a filename or path."
             ),
             default=None,
         ),
@@ -3270,7 +3270,8 @@ def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MC
         str | None,
         Field(
             description=(
-                "Optional. Path to a manifest YAML file on the machine running this MCP server."
+                "Optional. Absolute path to a manifest YAML file on the machine running this "
+                "MCP server."
             ),
             default=None,
         ),

From da49fc9d974c5231378906458b3ab4e9a2e73593 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 14 Oct 2026 05:03:13 +0000
Subject: [PATCH 6/8] refactor(mcp): use callable extra_help_text from
 fastmcp-extensions#150

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 airbyte/mcp/_tool_utils.py |  2 +-
 airbyte/mcp/cloud.py       | 57 +++++++-------------------------------
 pyproject.toml             |  2 +-
 uv.lock                    | 12 +++-----
 4 files changed, 16 insertions(+), 57 deletions(-)

diff --git a/airbyte/mcp/_tool_utils.py b/airbyte/mcp/_tool_utils.py
index 2473e0e92..9c950a087 100644
--- a/airbyte/mcp/_tool_utils.py
+++ b/airbyte/mcp/_tool_utils.py
@@ -453,7 +453,7 @@ def mcp_tool(  # noqa: PLR0913 - mirrors the upstream decorator's kwargs
     meta: Mapping[str, object] | None = None,
     app: AppConfig | None = None,
     required_capabilities: Iterable[Capability | str] | None = None,
-    extra_help_text: str | None = None,
+    extra_help_text: str | Callable[[], str | None] | None = None,
     pipeline_change: bool | None = None,
     external_access: bool = False,
     tracing: bool | Callable[[Mapping[str, object]], Mapping[str, object]] = True,
diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index 70d383b97..f760ddbe9 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -10,14 +10,12 @@
 # tool / helper definitions as a redundant "API Documentation" list.
 __all__: list[str] = []
 
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Annotated, Any, Final, Literal, TypeVar, cast
 
 import requests
 from fastmcp import Context, FastMCP
-from fastmcp.server.transforms import GetToolNext, Transform, VersionSpec
-from fastmcp.tools.base import Tool
 from fastmcp_extensions import TraceArg, get_mcp_config, register_mcp_tools
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -123,6 +121,12 @@
 MANIFEST_YAML_SOURCE_HINT = (
     "Provide the manifest as either `manifest_yaml_text` or `manifest_yaml_file`, not both."
 )
+
+
+def _manifest_yaml_source_hint() -> str | None:
+    return MANIFEST_YAML_SOURCE_HINT if is_trusted_execution_enabled() else None
+
+
 SKILL_DOCS_SECTION_HINT = """\
 Skill doc sections for direct API queries and actions are named with the pattern
 `actions.<entity_type>.<action>`, matching the `entity_type` and `action` arguments of the
@@ -2974,7 +2978,7 @@ def _get_custom_source_definition_description(
     )
 
 
-@mcp_tool()
+@mcp_tool(extra_help_text=_manifest_yaml_source_hint)
 def publish_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MCP tool schema.
     ctx: Context,
     name: Annotated[
@@ -3246,9 +3250,7 @@ def get_connector_builder_draft_manifest(
     }
 
 
-@mcp_tool(
-    destructive=True,
-)
+@mcp_tool(destructive=True, extra_help_text=_manifest_yaml_source_hint)
 def update_custom_source_definition(  # noqa: PLR0913  # Explicit args mirror MCP tool schema.
     ctx: Context,
     definition_id: Annotated[
@@ -4071,52 +4073,15 @@ def _add_defaults_for_exclude_args(
             func.__signature__ = sig.replace(parameters=new_params)  # type: ignore[attr-defined]
 
 
-class _ToolDescriptionSuffix(Transform):
-    """Append mode-specific guidance to selected tool descriptions."""
-
-    def __init__(self, suffixes: Mapping[str, str]) -> None:
-        self._suffixes = suffixes
-
-    def _apply(self, tool: Tool) -> Tool:
-        if tool.name not in self._suffixes:
-            return tool
-        suffix = self._suffixes[tool.name]
-        return tool.model_copy(
-            update={"description": f"{tool.description or ''}\n\n{suffix}".strip()}
-        )
-
-    async def list_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]:
-        return [self._apply(tool) for tool in tools]
-
-    async def get_tool(
-        self,
-        name: str,
-        call_next: GetToolNext,
-        *,
-        version: VersionSpec | None = None,
-    ) -> Tool | None:
-        tool = await call_next(name, version=version)
-        return self._apply(tool) if tool is not None else None
-
-
 def register_cloud_tools(app: FastMCP) -> None:
     """Register cloud tools with the FastMCP app.
 
     Args:
         app: FastMCP application instance
     """
-    trusted = is_trusted_execution_enabled()
     exclude_args = ["workspace_id"] if AIRBYTE_CLOUD_WORKSPACE_ID_IS_SET else []
-    description_suffix: _ToolDescriptionSuffix | None = None
-    if not trusted:
+    if not is_trusted_execution_enabled():
         exclude_args.append("manifest_yaml_file")
-    else:
-        description_suffix = _ToolDescriptionSuffix(
-            {
-                "publish_custom_source_definition": MANIFEST_YAML_SOURCE_HINT,
-                "update_custom_source_definition": MANIFEST_YAML_SOURCE_HINT,
-            }
-        )
     if exclude_args:
         _add_defaults_for_exclude_args(exclude_args)
     register_mcp_tools(
@@ -4124,5 +4089,3 @@ def register_cloud_tools(app: FastMCP) -> None:
         mcp_module=__name__,
         exclude_args=exclude_args or None,
     )
-    if description_suffix is not None:
-        app.add_transform(description_suffix)
diff --git a/pyproject.toml b/pyproject.toml
index 1573fa58f..0c4ddf51c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -46,7 +46,7 @@ dependencies = [
     "typing-extensions",
     "uuid7>=0.1.0,<1.0",
     "fastmcp>=4.0.9,<5.0",
-    "fastmcp-extensions>=0.34.2,<1.0.0",
+    "fastmcp-extensions @ git+https://github.com/airbytehq/fastmcp-extensions@devin/1791953245-callable-extra-help-text",  # TK-TODO: Revert to a PyPI version once https://github.com/airbytehq/fastmcp-extensions/pull/150 is released.
     "starlette",  # Transitive dependency of fastmcp, imported directly
     "httpx",  # Transitive dependency of fastmcp, imported directly
     "mcp",  # Transitive dependency of fastmcp, imported directly
diff --git a/uv.lock b/uv.lock
index 6a8507054..859dd5068 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1,5 +1,5 @@
 version = 1
-revision = 3
+revision = 5
 requires-python = ">=3.11, <3.15"
 resolution-markers = [
     "python_full_version >= '3.14'",
@@ -248,7 +248,7 @@ requires-dist = [
     { name = "duckdb", specifier = "==1.4.3" },
     { name = "duckdb-engine", specifier = "==0.17.0" },
     { name = "fastmcp", specifier = ">=4.0.9,<5.0" },
-    { name = "fastmcp-extensions", specifier = ">=0.34.2,<1.0.0" },
+    { name = "fastmcp-extensions", git = "https://github.com/airbytehq/fastmcp-extensions?rev=devin%2F1791953245-callable-extra-help-text" },
     { name = "google-auth", specifier = ">=2.27.0,<3.0" },
     { name = "google-cloud-bigquery", specifier = ">=3.12.0,<4.0" },
     { name = "google-cloud-bigquery-storage", specifier = ">=2.25.0,<3.0" },
@@ -1352,8 +1352,8 @@ wheels = [
 
 [[package]]
 name = "fastmcp-extensions"
-version = "0.34.2"
-source = { registry = "https://pypi.org/simple" }
+version = "0.34.2.post2.dev0+50cae05"
+source = { git = "https://github.com/airbytehq/fastmcp-extensions?rev=devin%2F1791953245-callable-extra-help-text#50cae051d86a547d2db03428c5c7284824c12f10" }
 dependencies = [
     { name = "fastmcp" },
     { name = "httpx" },
@@ -1367,10 +1367,6 @@ dependencies = [
     { name = "starlette" },
     { name = "uvicorn" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/ec/6d/379330f660ba2a4ed9fa7a6b0fe8c467f905469faa6aa26b6c55ca0a603a/fastmcp_extensions-0.34.2.tar.gz", hash = "sha256:b724514f60b65c3e73f0b4b6aaf2fc5b555bf30cd18f27bb6e8994497725e133", size = 356897, upload-time = "2026-10-10T06:22:20.759Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/77/5cc2e6de8ab239969811cdb1be44a59359c879e7494f8dcf71b037d04e4e/fastmcp_extensions-0.34.2-py3-none-any.whl", hash = "sha256:8628542fb17ea016519ae755c6aaa429b62ee042929009fc7d961cef3a0a7cbd", size = 156076, upload-time = "2026-10-10T06:22:19.108Z" },
-]
 
 [[package]]
 name = "fastmcp-slim"

From f4ecc0c49a31b4911fd3103b51bcb76338191e6b Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 14 Oct 2026 05:07:50 +0000
Subject: [PATCH 7/8] ci: suppress intentional dependency TK-TODO marker

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 pyproject.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pyproject.toml b/pyproject.toml
index 0c4ddf51c..eb0c0ed36 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -46,7 +46,7 @@ dependencies = [
     "typing-extensions",
     "uuid7>=0.1.0,<1.0",
     "fastmcp>=4.0.9,<5.0",
-    "fastmcp-extensions @ git+https://github.com/airbytehq/fastmcp-extensions@devin/1791953245-callable-extra-help-text",  # TK-TODO: Revert to a PyPI version once https://github.com/airbytehq/fastmcp-extensions/pull/150 is released.
+    "fastmcp-extensions @ git+https://github.com/airbytehq/fastmcp-extensions@devin/1791953245-callable-extra-help-text",  # TK-TODO: Revert to a PyPI version once https://github.com/airbytehq/fastmcp-extensions/pull/150 is released. IGNORE:TK
     "starlette",  # Transitive dependency of fastmcp, imported directly
     "httpx",  # Transitive dependency of fastmcp, imported directly
     "mcp",  # Transitive dependency of fastmcp, imported directly

From d3c8cc1275ff5172493ac3dcacd9bb4ed9149ee1 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 14 Oct 2026 05:20:37 +0000
Subject: [PATCH 8/8] chore(deps): bump fastmcp-extensions to 0.35.0

Co-Authored-By: AJ Steers <aj@airbyte.io>
---
 pyproject.toml |  2 +-
 uv.lock        | 10 +++++++---
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/pyproject.toml b/pyproject.toml
index eb0c0ed36..07f3c5cb5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -46,7 +46,7 @@ dependencies = [
     "typing-extensions",
     "uuid7>=0.1.0,<1.0",
     "fastmcp>=4.0.9,<5.0",
-    "fastmcp-extensions @ git+https://github.com/airbytehq/fastmcp-extensions@devin/1791953245-callable-extra-help-text",  # TK-TODO: Revert to a PyPI version once https://github.com/airbytehq/fastmcp-extensions/pull/150 is released. IGNORE:TK
+    "fastmcp-extensions>=0.35.0,<1.0.0",
     "starlette",  # Transitive dependency of fastmcp, imported directly
     "httpx",  # Transitive dependency of fastmcp, imported directly
     "mcp",  # Transitive dependency of fastmcp, imported directly
diff --git a/uv.lock b/uv.lock
index 859dd5068..27b1dbfbe 100644
--- a/uv.lock
+++ b/uv.lock
@@ -248,7 +248,7 @@ requires-dist = [
     { name = "duckdb", specifier = "==1.4.3" },
     { name = "duckdb-engine", specifier = "==0.17.0" },
     { name = "fastmcp", specifier = ">=4.0.9,<5.0" },
-    { name = "fastmcp-extensions", git = "https://github.com/airbytehq/fastmcp-extensions?rev=devin%2F1791953245-callable-extra-help-text" },
+    { name = "fastmcp-extensions", specifier = ">=0.35.0,<1.0.0" },
     { name = "google-auth", specifier = ">=2.27.0,<3.0" },
     { name = "google-cloud-bigquery", specifier = ">=3.12.0,<4.0" },
     { name = "google-cloud-bigquery-storage", specifier = ">=2.25.0,<3.0" },
@@ -1352,8 +1352,8 @@ wheels = [
 
 [[package]]
 name = "fastmcp-extensions"
-version = "0.34.2.post2.dev0+50cae05"
-source = { git = "https://github.com/airbytehq/fastmcp-extensions?rev=devin%2F1791953245-callable-extra-help-text#50cae051d86a547d2db03428c5c7284824c12f10" }
+version = "0.35.0"
+source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "fastmcp" },
     { name = "httpx" },
@@ -1367,6 +1367,10 @@ dependencies = [
     { name = "starlette" },
     { name = "uvicorn" },
 ]
+sdist = { url = "https://files.pythonhosted.org/packages/74/45/dcf55ab5ba27c238654c71d92af320eec2312ae3fad62572b78ea0b9f8fc/fastmcp_extensions-0.35.0.tar.gz", hash = "sha256:bad04d428433c424407277928bc44d77e356f9c2ff73b42aa9e058ad54173038", size = 357967, upload-time = "2026-10-14T05:17:25.382Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/1d/5f/3836a5aea53632be7a7629b7514003138377cfc5e37958308839cb627d7e/fastmcp_extensions-0.35.0-py3-none-any.whl", hash = "sha256:86d31337876faec30dcc336c68b7103d84f553ce8f1555540888c717ddad9e8f", size = 156435, upload-time = "2026-10-14T05:17:23.977Z" },
+]
 
 [[package]]
 name = "fastmcp-slim"