feat(mcp)!: split manifest_yaml into manifest_yaml_text and manifest_yaml_file

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

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,

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,
@@ -118,6 +118,15 @@
     "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."
+)
+
+
+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
@@ -2969,8 +2978,8 @@ def _get_custom_source_definition_description(
     )
 
 
-@mcp_tool()
-def publish_custom_source_definition(
+@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[
         str,
@@ -2986,13 +2995,21 @@ 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."
+            ),
+            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."
+                "Absolute path to a manifest YAML file on the machine running this MCP server."
             ),
             default=None,
         ),
@@ -3044,7 +3061,20 @@ 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:
+        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=error_message,
+        )
 
     # Resolve testing values from inline config and/or secret
     testing_values_dict: dict[str, Any] | None = None
@@ -3220,27 +3250,34 @@ def get_connector_builder_draft_manifest(
     }
 
 
-@mcp_tool(
-    destructive=True,
-)
-def update_custom_source_definition(
+@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[
         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."
+                "Optional. New manifest. Full YAML contents of the declarative (low-code) "
+                "connector manifest: the file's text, not a filename or path."
+            ),
+            default=None,
+        ),
+    ] = None,
+    manifest_yaml_file: Annotated[
+        str | None,
+        Field(
+            description=(
+                "Optional. Absolute path to a manifest YAML file on the machine running this "
+                "MCP server."
             ),
             default=None,
         ),
     ] = None,
-    *,
     workspace_id: Annotated[
         str | None,
         Field(
@@ -3300,31 +3337,38 @@ 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.
+    Provide at least one value to update.
     """
     check_guid_created_in_session(definition_id)
 
     workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
 
     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
     ):
+        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, 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,
                 "workspace_id": workspace.workspace_id,
             },
         )
 
-    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 +4079,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,
     )

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>=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

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",

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