Azure · Chakradhar886 · Aug 10, 2026 · Aug 10, 2026 · Aug 10, 2026 · Aug 10, 2026
@@ -5,6 +5,7 @@
 ### Features Added
 
 ### Bugs Fixed
+- `load_component` now accepts a `default` value for asset-type inputs (`uri_file`, `uri_folder`, `mltable`, `mlflow_model`, `custom_model`), matching the public CLI v2 YAML schema. Previously this raised `UserErrorException: Non-primitive type Input has no default value.`
 - Fixed internal pipeline `Command` node dropping node-level interactive `services` (SSH, JupyterLab, TensorBoard, VS Code, etc.) during serialization, which prevented interactive endpoints from being created for Singularity jobs. The `services` are now serialized into the pipeline REST request and round-tripped on deserialization, matching the public `Command` node behavior.
 - Fixed `MLClient.jobs.create_or_update`, `archive`, and `restore` failing for previously-fetched jobs across all job types by routing metadata-only edits through the RunHistory PATCH endpoint.
 - Fixed `DeploymentTemplate.creation_context` always being `None` when retrieved via `get()` or `list()`. The created/modified timestamps and identity returned by the service (as `createdTime` / `modifiedTime` / `createdBy`) are now populated on `creation_context`, making `DeploymentTemplate` consistent with `Model` and `Environment`.

@@ -108,6 +108,7 @@ class ComponentParameterTypes:
 
 
 class IOConstants:
+    ASSET_INPUT_TYPES = ("uri_file", "uri_folder", "mltable", "mlflow_model", "custom_model")
     PRIMITIVE_STR_2_TYPE = {
         ComponentParameterTypes.INTEGER: int,
         ComponentParameterTypes.STRING: str,
@@ -133,11 +134,11 @@ class IOConstants:
     }
     # For validation, indicates specific parameters combination for each type
     INPUT_TYPE_COMBINATION = {
-        "uri_folder": ["path", "mode"],
-        "uri_file": ["path", "mode"],
-        "mltable": ["path", "mode"],
-        "mlflow_model": ["path", "mode"],
-        "custom_model": ["path", "mode"],
+        "uri_folder": ["path", "mode", "default"],
+        "uri_file": ["path", "mode", "default"],
+        "mltable": ["path", "mode", "default"],
+        "mlflow_model": ["path", "mode", "default"],
+        "custom_model": ["path", "mode", "default"],
         "integer": ["default", "min", "max"],
         "number": ["default", "min", "max"],
         "string": ["default"],

@@ -6,10 +6,12 @@
 
 from typing import Any, Callable, List, Mapping
 
+from azure.ai.ml.constants._component import IOConstants
 from azure.ai.ml.dsl._dynamic import KwParameter, create_kw_function_from_parameters
 from azure.ai.ml.entities import Component as ComponentEntity
 from azure.ai.ml.entities._builders import Command
 from azure.ai.ml.entities._component.datatransfer_component import DataTransferImportComponent
+from azure.ai.ml.entities._inputs_outputs import Input
 
 
 def get_dynamic_input_parameter(inputs: Mapping) -> List:
@@ -24,8 +26,13 @@ def get_dynamic_input_parameter(inputs: Mapping) -> List:
         KwParameter(
             name=name,
             annotation=input._get_python_builtin_type_str(),
-            default=None,
+            default=(
+                Input(type=input.type, path=input.default, mode=input.mode)
+                if input.type in IOConstants.ASSET_INPUT_TYPES and input.default is not None
+                else None
+            ),
             _type=input._get_python_builtin_type_str(),
+            _copy_default=input.type in IOConstants.ASSET_INPUT_TYPES and input.default is not None,
         )
         for name, input in inputs.items()
     ]

@@ -1,6 +1,7 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
+import copy
 import logging
 import types
 from inspect import Parameter, Signature
@@ -24,14 +25,23 @@ class KwParameter(Parameter):
     :type _type: str
     :param _optional: Indicates if the parameter is optional, defaults to False.
     :type _optional: bool
+    :param _copy_default: Indicates if the default should be copied for each call, defaults to False.
+    :type _copy_default: bool
     """
 
     def __init__(
-        self, name: str, default: Any, annotation: Any = Parameter.empty, _type: str = "str", _optional: bool = False
+        self,
+        name: str,
+        default: Any,
+        annotation: Any = Parameter.empty,
+        _type: str = "str",
+        _optional: bool = False,
+        _copy_default: bool = False,
     ) -> None:
         super().__init__(name, Parameter.KEYWORD_ONLY, default=default, annotation=annotation)
         self._type = _type
         self._optional = _optional
+        self._copy_default = _copy_default
 
 
 def _replace_function_name(func: types.FunctionType, new_name: str) -> types.FunctionType:
@@ -165,12 +175,18 @@ def create_kw_function_from_parameters(
             target=ErrorTarget.COMPONENT,
         )
     default_kwargs = {p.name: p.default for p in parameters}
+    copied_default_keys = {
+        p.name for p in parameters if isinstance(p, KwParameter) and p._copy_default  # pylint: disable=protected-access
+    }
 
     def f(**kwargs: Any) -> Any:
         # We need to make sure all keys of kwargs are valid.
         # Merge valid group keys with original keys.
         _assert_arg_valid(kwargs, [*list(default_kwargs.keys()), *flattened_group_keys], func_name=func_name)
         # We need to put the default args to the kwargs before invoking the original function.
+        for key in copied_default_keys:
+            if key not in kwargs:
+                kwargs[key] = copy.deepcopy(default_kwargs[key])
         _update_dct_if_not_exist(kwargs, default_kwargs)
         return func(**kwargs)
 

@@ -21,7 +21,7 @@
 from azure.ai.ml._utils._asset_utils import get_object_hash
 from azure.ai.ml._utils.utils import hash_dict, is_data_binding_expression
 from azure.ai.ml.constants._common import ARM_ID_PREFIX, ASSET_ARM_ID_REGEX_FORMAT, COMPONENT_TYPE
-from azure.ai.ml.constants._component import ComponentSource, NodeType
+from azure.ai.ml.constants._component import ComponentSource, IOConstants, NodeType
 from azure.ai.ml.constants._job.pipeline import ValidationErrorCode
 from azure.ai.ml.entities._builders import BaseNode, Command
 from azure.ai.ml.entities._builders.control_flow_node import ControlFlowNode, LoopNode
@@ -36,6 +36,9 @@
 module_logger = logging.getLogger(__name__)
 
 
+_ASSET_INPUT_DEFAULTS_PROPERTY = "_azureml_asset_input_defaults"
+
+
 class PipelineComponent(Component):
     """Pipeline component, currently used to store components in an azure.ai.ml.dsl.pipeline.
 
@@ -452,6 +455,18 @@ def _get_telemetry_values(self, *args: Any, **kwargs: Any) -> Dict:
 
     @classmethod
     def _from_rest_object_to_init_params(cls, obj: ComponentVersion) -> Dict:
+        serialized_defaults = (obj.properties.properties or {}).get(_ASSET_INPUT_DEFAULTS_PROPERTY)
+        if serialized_defaults:
+            try:
+                asset_input_defaults = json.loads(serialized_defaults)
+                if not isinstance(asset_input_defaults, dict):
+                    raise ValueError("Asset input defaults must be a mapping.")
+                for input_name, default in asset_input_defaults.items():
+                    input_spec = obj.properties.component_spec.get("inputs", {}).get(input_name)
+                    if input_spec is not None and input_spec.get("default") is None:
+                        input_spec["default"] = default
+            except (TypeError, ValueError):
+                module_logger.warning("Failed to restore pipeline component asset input defaults.")
         # Pop jobs to avoid it goes with schema load
         jobs = obj.properties.component_spec.pop("jobs", None)
         init_params_dict: dict = super()._from_rest_object_to_init_params(obj)
@@ -509,11 +524,21 @@ def _to_rest_object(self) -> ComponentVersion:
             # hack while full pass through supported is worked on for IPP fields
             component.pop("intellectual_property")
             component["intellectualProperty"] = self._intellectual_property._to_rest_object()
+        properties = dict(self.properties) if self.properties else {}
+        asset_input_defaults = {
+            input_name: component_input.default
+            for input_name, component_input in self.inputs.items()
+            if component_input.type in IOConstants.ASSET_INPUT_TYPES and component_input.default is not None
+        }
+        if asset_input_defaults:
+            properties[_ASSET_INPUT_DEFAULTS_PROPERTY] = json.dumps(asset_input_defaults)
+        else:
+            properties.pop(_ASSET_INPUT_DEFAULTS_PROPERTY, None)
         properties = ComponentVersionProperties(
             component_spec=component,
             description=self.description,
             is_anonymous=self._is_anonymous,
-            properties=self.properties,
+            properties=properties,
             tags=self.tags,
         )
         result = ComponentVersion(properties=properties)

@@ -373,6 +373,17 @@ def _update_default(self, default_value: Any) -> None:
         msg_prefix = f"Default value of Input {name}"
 
         if not self._is_primitive_type and default_value is not None:
+            # Asset-type inputs accept a string default value (e.g. "azureml:", "https://",
+            # local path) matching the public CLI v2 YAML schema.
+            if not self._multiple_types and self.type in IOConstants.ASSET_INPUT_TYPES:
+                if isinstance(default_value, str):
+                    self.default: Any = default_value
+                    return
+                msg = (
+                    f"{msg_prefix}cannot be set: default for type '{self.type}' must be a "
+                    f"string asset reference, got '{type(default_value)}'."
+                )
+                raise UserErrorException(msg)
             msg = f"{msg_prefix}cannot be set: Non-primitive type Input has no default value."
             raise UserErrorException(msg)
         if isinstance(default_value, float) and not math.isfinite(default_value):

@@ -12,7 +12,7 @@
 from azure.ai.ml._restclient.arm_ml_service.models import (
     JobOutput as RestJobOutput,
 )
-from azure.ai.ml.constants._component import ComponentJobConstants
+from azure.ai.ml.constants._component import ComponentJobConstants, IOConstants
 from azure.ai.ml.entities._inputs_outputs import GroupInput, Input, Output
 from azure.ai.ml.entities._util import copy_output_setting
 from azure.ai.ml.exceptions import ErrorTarget, ValidationErrorType, ValidationException
@@ -118,6 +118,8 @@ def _build_output(self, name: str, meta: Optional[Output], data: Optional[Union[
 
     # pylint: disable=unused-argument
     def _get_default_input_val(self, val: Any):  # type: ignore
+        if isinstance(val, Input) and val.type in IOConstants.ASSET_INPUT_TYPES and val.default is not None:
+            return val.default
         # use None value as data placeholder for unfilled inputs.
         # server side will fill the default value
         return None
@@ -644,6 +646,8 @@ def _get_default_input_val(self, val: Any):  # type: ignore
         if isinstance(val, GroupInput):
             # Copy default value dict for group
             return copy.deepcopy(val.default)
+        if isinstance(val, Input) and val.type in IOConstants.ASSET_INPUT_TYPES and val.default is not None:
+            return Input(type=val.type, path=val.default, mode=val.mode)
         return val.default
 
     def _update_output_types(self, rest_data_outputs: Dict) -> None:

@@ -673,9 +673,21 @@ def create_or_update(
         )
 
         if not result:
-            component = self.get(name=name, version=version)
+            created_component = self.get(name=name, version=version)
         else:
-            component = Component._from_rest_object(result)
+            created_component = Component._from_rest_object(result)
+
+        if isinstance(component, PipelineComponent) and isinstance(created_component, PipelineComponent):
+            for input_name, source_input in component.inputs.items():
+                created_input = created_component.inputs.get(input_name)
+                if (
+                    created_input
+                    and not source_input._is_primitive_type
+                    and source_input.default is not None
+                    and created_input.default is None
+                ):
+                    created_input.default = source_input.default
+        component = created_component
 
         self._resolve_azureml_id(
             component=component,

@@ -3,6 +3,7 @@
 
 import pytest
 
+from azure.ai.ml import Input
 from azure.ai.ml._restclient.arm_ml_service.models import (
     ComponentContainer as ComponentContainerData,
     ComponentContainerProperties as ComponentContainerDetails,
@@ -11,6 +12,7 @@
     SystemData,
 )
 from azure.ai.ml._scope_dependent_operations import OperationConfig, OperationScope
+from azure.ai.ml.entities import PipelineComponent
 from azure.ai.ml.entities._component.command_component import CommandComponent
 from azure.ai.ml.entities._component.component import Component
 from azure.ai.ml.operations import ComponentOperations
@@ -84,6 +86,84 @@ def test_create_skip_validation(self, mock_component_operation: ComponentOperati
             mock_component_operation.create_or_update(component)
             mock_thing.assert_called_once()
 
+    def test_create_preserves_pipeline_asset_input_default(self, mock_component_operation: ComponentOperations) -> None:
+        component = PipelineComponent(
+            name="pipeline_component",
+            version="1",
+            inputs={"default_data": Input(type="uri_file", default="azureml:test_data:1")},
+            jobs={},
+        )
+        service_component = PipelineComponent(
+            name="pipeline_component",
+            version="1",
+            inputs={"default_data": Input(type="uri_file")},
+            jobs={},
+        )
+
+        with patch.object(ComponentOperations, "_resolve_arm_id_or_upload_dependencies"), patch(
+            "azure.ai.ml.operations._component_operations.Component._from_rest_object",
+            return_value=service_component,
+        ):
+            created_component = mock_component_operation.create_or_update(component)
+
+        assert created_component.inputs["default_data"].default == "azureml:test_data:1"
+        node = created_component()
+        built_input = node._build_inputs()["default_data"]
+        assert built_input.type == "uri_file"
+        assert built_input.path == "azureml:test_data:1"
+        assert node._to_rest_inputs()["default_data"] == {
+            "uri": "azureml:test_data:1",
+            "job_input_type": "uri_file",
+        }
+        assert node._to_job()._to_rest_object().properties.inputs["default_data"].as_dict() == {
+            "uri": "azureml:test_data:1",
+            "jobInputType": "uri_file",
+        }
+
+        second_node = created_component()
+        assert node._build_inputs()["default_data"] is not second_node._build_inputs()["default_data"]
+        node._build_inputs()["default_data"].path = "azureml:mutated:1"
+        assert second_node._build_inputs()["default_data"].path == "azureml:test_data:1"
+
+    def test_create_preserves_pipeline_asset_input_default_after_get(
+        self, mock_component_operation: ComponentOperations
+    ) -> None:
+        component = PipelineComponent(
+            name="pipeline_component",
+            version="1",
+            inputs={"default_data": Input(type="uri_file", default="azureml:test_data:1")},
+            jobs={},
+        )
+        service_component = PipelineComponent(
+            name="pipeline_component",
+            version="1",
+            inputs={"default_data": Input(type="uri_file")},
+            jobs={},
+        )
+
+        with patch.object(ComponentOperations, "_resolve_arm_id_or_upload_dependencies"), patch.object(
+            ComponentOperations, "_create_or_update_component_version", return_value=None
+        ), patch.object(ComponentOperations, "get", return_value=service_component):
+            created_component = mock_component_operation.create_or_update(component)
+
+        assert created_component.inputs["default_data"].default == "azureml:test_data:1"
+
+    def test_get_restores_pipeline_asset_input_default(self, mock_component_operation: ComponentOperations) -> None:
+        component = PipelineComponent(
+            name="pipeline_component",
+            version="1",
+            inputs={"default_data": Input(type="uri_file", default="azureml:test_data:1")},
+            jobs={},
+        )
+        rest_component = component._to_rest_object()
+        rest_component.properties.component_spec["inputs"]["default_data"].pop("default")
+        mock_component_operation._version_operation.get.return_value = rest_component
+
+        fetched_component = mock_component_operation.get(name="pipeline_component", version="1")
+
+        assert fetched_component.inputs["default_data"].default == "azureml:test_data:1"
+        assert fetched_component()._build_inputs()["default_data"].path == "azureml:test_data:1"
+
     def test_create_autoincrement(
         self, mock_component_operation: ComponentOperations, mock_component_from_rest
     ) -> None: