feat(mcp): deploy sources from custom YAML definitions via deploy_connector_to_cloud

airbyte/cloud/connectors.py

@@ -59,11 +59,13 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol
 
+import jsonschema
 import requests
 import yaml
 from pydantic import ValidationError
 
 from airbyte import exceptions as exc
+from airbyte._connector_base import _json_path_from_absolute_path
 from airbyte._direct_connectors import api_util as agents_api_util
 from airbyte._direct_connectors import connector_docs
 from airbyte._direct_connectors.actions import (
@@ -98,6 +100,7 @@
     _SourceResponseLike,
 )
 from airbyte.registry import ApiDocsUrl, get_connector_api_docs_urls
+from airbyte.secrets.hydration import hydrate_secrets
 
 
 if TYPE_CHECKING:
@@ -1274,7 +1277,7 @@ def _unwrap_builder_manifest(value: object) -> dict[str, Any] | None:
     return value
 
 
-class CustomCloudSourceDefinition:
+class CustomCloudSourceDefinition:  # noqa: PLR0904
     """A custom source connector definition in Airbyte Cloud.
 
     This represents either a YAML (declarative) or Docker-based custom source definition.
@@ -1339,6 +1342,40 @@ def manifest(self) -> dict[str, Any] | None:
             self._definition_info = self._fetch_definition_info()
         return self._definition_info.manifest
 
+    @property
+    def config_spec_jsonschema(self) -> dict[str, Any] | None:
+        """Get the JSON schema for this custom source's configuration."""
+        manifest = self.manifest
+        if manifest is None:
+            return None
+        manifest_spec = manifest.get("spec")
+        if not isinstance(manifest_spec, dict):
+            return None
+        config_spec_jsonschema = manifest_spec.get("connection_specification")
+        return config_spec_jsonschema if isinstance(config_spec_jsonschema, dict) else None
+
+    def validate_config(self, config: dict[str, Any]) -> None:
+        """Validate the config against the custom source's connection specification."""
+        specification = self.config_spec_jsonschema
+        if specification is None:
+            return
+
+        try:
+            jsonschema.validate(hydrate_secrets(config), specification)
+        except jsonschema.ValidationError as ex:
+            raise exc.AirbyteConnectorValidationFailedError(
+                connector_name=self.name,
+                message=(
+                    f"The provided config is not valid at "
+                    f"`{_json_path_from_absolute_path(ex.absolute_path)}` "
+                    f"(failed the `{ex.validator}` rule)."
+                ),
+                context={
+                    "error_path": ex.path,
+                    "error_schema": ex.schema,
+                },
+            ) from ex
+
     @property
     def version(self) -> str | None:
         """Get the manifest version. Only present for YAML connectors."""
@@ -1810,7 +1847,7 @@ def set_testing_values(
             if manifest_spec:
                 spec = {
                     "documentationUrl": manifest_spec.get("documentation_url"),
-                    "connectionSpecification": manifest_spec.get("connection_specification", {}),
+                    "connectionSpecification": self.config_spec_jsonschema or {},
                     "advancedAuth": manifest_spec.get("advanced_auth"),
                 }
 

airbyte/cloud/workspaces.py

@@ -70,7 +70,7 @@
 from airbyte.exceptions import AirbyteCloudError
 from airbyte.registry import _get_connector_name_by_definition_id
 from airbyte.secrets.base import SecretString
-from airbyte.secrets.hydration import detect_hardcoded_secrets
+from airbyte.secrets.hydration import detect_hardcoded_secrets, hydrate_secrets
 
 
 if TYPE_CHECKING:
@@ -667,12 +667,14 @@ def deploy_source(
 
         Args:
             name: The name to use when deploying.
-            source: The source object to deploy, or (with `defer_credentials=True`) a
+            source: The source object to deploy, or a dictionary of configuration values
+                when `definition_id` is provided. With `defer_credentials=True`, pass a
                 dictionary of non-secret configuration values.
             unique: Whether to require a unique name. If `True`, duplicate names
                 are not allowed. Defaults to `True`.
             random_name_suffix: Whether to append a random suffix to the name.
-            definition_id: The source definition ID. Required with `defer_credentials=True`.
+            definition_id: The source definition ID. Required for a dictionary config (for
+                example, a custom definition ID) and with `defer_credentials=True`.
             defer_credentials: Save a draft with partial configuration. A person completes
                 credentials and other missing settings at the returned source's `connector_url`.
                 A successful connection check promotes the draft. Raises
@@ -691,12 +693,17 @@ def deploy_source(
                 ),
             )
         if isinstance(source, dict):
-            raise exc.AirbyteLibInputError(
-                message="`source` must be a `Source` object unless `defer_credentials=True`.",
-            )
-
-        source_config_dict = source._hydrated_config.copy()  # noqa: SLF001 (non-public API)
-        source_config_dict["sourceType"] = source.name.replace("source-", "")
+            if definition_id is None:
+                raise exc.AirbyteLibInputError(
+                    message=(
+                        "`source` must be a `Source` object unless `definition_id` or "
+                        "`defer_credentials=True` is given."
+                    ),
+                )
+            source_config_dict = hydrate_secrets(source)
+        else:
+            source_config_dict = source._hydrated_config.copy()  # noqa: SLF001 (non-public API)
+            source_config_dict["sourceType"] = source.name.replace("source-", "")
 
         if random_name_suffix:
             name += f" (ID: {text_util.generate_random_suffix()})"
@@ -810,10 +817,18 @@ def _deploy_deferred(
         random_name_suffix: bool,
     ) -> str:
         """Create a connector with deferred credentials on the Config API and return its ID."""
+        spec_json_schema = _get_deferred_spec(definition_id) if definition_id else None
+        if spec_json_schema is None and connector_type == "source" and definition_id:
+            with contextlib.suppress(Exception):
+                spec_json_schema = self.get_custom_source_definition(
+                    definition_id,
+                    definition_type="yaml",
+                ).config_spec_jsonschema
+
         config_dict, definition_id = _deferred_credentials_config(
             config,
             definition_id=definition_id,
-            spec_json_schema=_get_deferred_spec(definition_id) if definition_id else None,
+            spec_json_schema=spec_json_schema,
         )
 
         if random_name_suffix:

airbyte/mcp/cloud.py

@@ -610,26 +610,41 @@ def deploy_connector_to_cloud(  # noqa: PLR0913  # Mirrors the API surface.
         TraceArg.FINGERPRINT,
     ],
     connector_name: Annotated[
-        str,
+        str | None,
         Field(
             description=(
                 "The canonical name of the connector (e.g., 'source-faker' or "
                 "'destination-postgres')."
             ),
+            default=None,
         ),
         TraceArg.FINGERPRINT,
-    ],
+    ] = None,
     *,
     connector_type: Annotated[
         ConnectorType | None,
         Field(
             description=(
                 "Optional: `source` or `destination`. When omitted, inferred from the "
-                "`source-`/`destination-` prefix of `connector_name`."
+                "`source-`/`destination-` prefix of `connector_name`. Custom definitions "
+                "are sources."
             ),
             default=None,
         ),
     ] = None,
+    custom_definition_id: Annotated[
+        str | None,
+        Field(
+            description=(
+                "ID of a custom (YAML / Connector Builder) source definition in this "
+                "workspace, e.g. from `publish_custom_source_definition` or "
+                "`list_custom_source_definitions`. Pass this instead of `connector_name` "
+                "to deploy a source from a custom definition."
+            ),
+            default=None,
+        ),
+        TraceArg.HASH,
+    ] = None,
     workspace_id: Annotated[
         str | None,
         Field(
@@ -671,40 +686,81 @@ def deploy_connector_to_cloud(  # noqa: PLR0913  # Mirrors the API surface.
 ) -> str:
     """Deploy a source or destination connector to Airbyte Cloud.
 
+    Use `custom_definition_id` to deploy a custom YAML source definition.
+
     With `defer_credentials=True`, returns a JSON `DeferredDeployResult` whose `settings_url`
     is where a person completes the credentials.
     """
-    resolved_type = connector_type or _infer_connector_type_from_name(connector_name)
+    if (connector_name is None) == (custom_definition_id is None):
+        raise AirbyteLibInputError(
+            message="Provide exactly one of `connector_name` or `custom_definition_id`.",
+            guidance=(
+                "Use `connector_name` for a registry connector or "
+                "`custom_definition_id` for a custom source."
+            ),
+        )
+    if custom_definition_id is not None and connector_type == ConnectorType.DESTINATION:
+        raise AirbyteLibInputError(
+            message="Custom source definitions can only be deployed as sources.",
+            guidance="Pass `connector_type='source'` or omit `connector_type`.",
+        )
+
+    if custom_definition_id is not None:
+        resolved_type = ConnectorType.SOURCE
+    else:
+        assert connector_name is not None
+        resolved_type = connector_type or _infer_connector_type_from_name(connector_name)
     if defer_credentials:
         return _deploy_deferred_to_cloud(
             ctx,
             connector_type=resolved_type,
             name=name,
             connector_name=connector_name,
+            custom_definition_id=custom_definition_id,
             workspace_id=workspace_id,
             config=config,
             config_secret_name=config_secret_name,
             unique=unique,
         ).model_dump_json(indent=2)
 
-    connector: Source | Destination = (
-        get_source(connector_name, no_executor=True)
-        if resolved_type == ConnectorType.SOURCE
-        else get_destination(connector_name, no_executor=True)
-    )
-    config_dict = resolve_connector_config(
-        config=config,
-        config_secret_name=config_secret_name,
-        config_spec_jsonschema=connector.config_spec,
-    )
-    connector.set_config(config_dict, validate=True)
+    if custom_definition_id is not None:
+        workspace = _get_cloud_workspace(ctx, workspace_id)
+        definition = workspace.get_custom_source_definition(
+            custom_definition_id,
+            definition_type="yaml",
+        )
+        config_dict = resolve_connector_config(
+            config=config,
+            config_secret_name=config_secret_name,
+            config_spec_jsonschema=definition.config_spec_jsonschema,
+        )
+        definition.validate_config(config_dict)
+        deployed: CloudConnector = workspace.deploy_source(
+            name=name,
+            source=config_dict,
+            unique=unique,
+            definition_id=custom_definition_id,
+        )
+    else:
+        assert connector_name is not None
+        connector: Source | Destination = (
+            get_source(connector_name, no_executor=True)
+            if resolved_type == ConnectorType.SOURCE
+            else get_destination(connector_name, no_executor=True)
+        )
+        config_dict = resolve_connector_config(
+            config=config,
+            config_secret_name=config_secret_name,
+            config_spec_jsonschema=connector.config_spec,
+        )
+        connector.set_config(config_dict, validate=True)
 
-    workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
-    deployed: CloudConnector = (
-        workspace.deploy_source(name=name, source=connector, unique=unique)
-        if isinstance(connector, Source)
-        else workspace.deploy_destination(name=name, destination=connector, unique=unique)
-    )
+        workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
+        deployed = (
+            workspace.deploy_source(name=name, source=connector, unique=unique)
+            if isinstance(connector, Source)
+            else workspace.deploy_destination(name=name, destination=connector, unique=unique)
+        )
 
     register_guid_created_in_session(deployed.connector_id)
     return (
@@ -713,12 +769,13 @@ def deploy_connector_to_cloud(  # noqa: PLR0913  # Mirrors the API surface.
     )
 
 
-def _deploy_deferred_to_cloud(
+def _deploy_deferred_to_cloud(  # noqa: PLR0913
     ctx: Context,
     *,
     connector_type: ConnectorType,
     name: str,
-    connector_name: str,
+    connector_name: str | None,
+    custom_definition_id: str | None,
     workspace_id: str | None,
     config: dict | str | None,
     config_secret_name: str | None,
@@ -730,14 +787,19 @@ def _deploy_deferred_to_cloud(
             message="`config_secret_name` cannot be used with `defer_credentials=True`.",
             guidance="Pass non-secret configuration in `config`; credentials are entered in Cloud.",
         )
-    metadata = get_connector_metadata(connector_name)
-    if metadata is None or metadata.definition_id is None:
-        raise AirbyteConnectorNotRegisteredError(connector_name=connector_name)
-    if metadata.connector_type != connector_type:
-        raise AirbyteLibInputError(
-            message=f"`{connector_name}` is not a {connector_type} connector.",
-            guidance=f"Pass a `{connector_type}-*` connector name.",
-        )
+    if custom_definition_id is not None:
+        definition_id = custom_definition_id
+    else:
+        assert connector_name is not None
+        metadata = get_connector_metadata(connector_name)
+        if metadata is None or metadata.definition_id is None:
+            raise AirbyteConnectorNotRegisteredError(connector_name=connector_name)
+        if metadata.connector_type != connector_type:
+            raise AirbyteLibInputError(
+                message=f"`{connector_name}` is not a {connector_type} connector.",
+                guidance=f"Pass a `{connector_type}-*` connector name.",
+            )
+        definition_id = metadata.definition_id
     config_dict = resolve_connector_config(config=config)
 
     workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
@@ -748,15 +810,15 @@ def _deploy_deferred_to_cloud(
                 name=name,
                 source=config_dict,
                 unique=unique,
-                definition_id=metadata.definition_id,
+                definition_id=definition_id,
                 defer_credentials=True,
             )
         else:
             deployed = workspace.deploy_destination(
                 name=name,
                 destination=config_dict,
                 unique=unique,
-                definition_id=metadata.definition_id,
+                definition_id=definition_id,
                 defer_credentials=True,
             )
     except AirbyteDeferredSetupError as ex:

tests/integration_tests/cloud/test_custom_definitions.py

@@ -100,6 +100,41 @@ def test_publish_custom_yaml_source(
         fetched.permanently_delete(safe_mode=False)
 
 
+@pytest.mark.requires_creds
+def test_deploy_source_from_custom_yaml_definition(
+    cloud_workspace: CloudWorkspace,
+) -> None:
+    """Deploy a custom YAML source definition by passing its ID and a config dictionary."""
+    from airbyte._util import text_util
+
+    name = f"delete-me-custom-yaml-source-{text_util.generate_random_suffix()}"
+    definition = cloud_workspace.publish_custom_source_definition(
+        name=name,
+        manifest_yaml=TEST_YAML_MANIFEST,
+        unique=True,
+        pre_validate=True,
+    )
+    deployed_source = None
+
+    try:
+        try:
+            deployed_source = cloud_workspace.deploy_source(
+                name=name,
+                source={},
+                definition_id=definition.definition_id,
+            )
+
+            assert deployed_source.definition_id == definition.definition_id
+        finally:
+            if deployed_source is not None:
+                cloud_workspace.permanently_delete_source(
+                    deployed_source,
+                    safe_mode=False,
+                )
+    finally:
+        definition.permanently_delete(safe_mode=False)
+
+
 @pytest.mark.requires_creds
 def test_yaml_validation_error(
     cloud_workspace: CloudWorkspace,