Built-in Type Support

How built-in values move through Flyte

When a task input or output is a Python value such as list[int], Dict[str, float], an enum, or a model, you normally declare that type on the task interface; you do not instantiate a transformer yourself. flyte-sdk routes the annotation through TypeEngine.to_literal_type() to build a Flyte LiteralType, and routes values through TypeEngine.to_literal() and TypeEngine.to_python_value() to produce or consume Flyte Literal values.

The runtime preserves the declared annotations when it converts native task arguments. This matters for collections: a runtime list does not carry its element type, but the task interface does. _internal/runtime/convert.py passes those interface types into TypeEngine.dict_to_literal_map():

async def convert_from_native_to_inputs(interface: NativeInterface, *args, **kwargs) -> Inputs:
    kwargs = interface.convert_to_kwargs(*args, **kwargs)

    missing = [key for key in interface.required_inputs() if key not in kwargs]
    if missing:
        raise ValueError(f"Missing required inputs: {', '.join(missing)}")

    if len(interface.inputs) == 0:
        return Inputs.empty()

    type_hints: Dict[str, type] = {}
    already_converted_kwargs: Dict[str, literals_pb2.Literal] = {}
    for input_name, (input_type, default_value) in interface.inputs.items():
        if input_name in kwargs:
            type_hints[input_name] = input_type
        elif (
            (default_value is not None and default_value is not inspect.Signature.empty)
            or (default_value is None and is_optional_type(input_type))
            or input_type is None
        ):
            if default_value == NativeInterface.has_default:
                if interface._remote_defaults is None or input_name not in interface._remote_defaults:
                    raise ValueError(f"Input '{input_name}' has a default value but it is not set in the interface.")
                already_converted_kwargs[input_name] = interface._remote_defaults[input_name]
            elif input_type is None:
                kwargs[input_name] = None
                type_hints[input_name] = NoneType
            else:
                kwargs[input_name] = default_value
                type_hints[input_name] = input_type

    literal_map = await TypeEngine.dict_to_literal_map(kwargs, type_hints)

Outputs follow the same annotation-first rule. convert_from_native_to_outputs() obtains the declared type and its LiteralType for every result, and wraps a TypeTransformerFailedError as RuntimeDataValidationError for that output:

async def convert_from_native_to_outputs(o: Any, interface: NativeInterface, task_name: str = "") -> Outputs:
    if not isinstance(o, tuple):
        o = (o,)

    named = []
    for (output_name, python_type), v in zip(interface.outputs.items(), o):
        try:
            lit = await TypeEngine.to_literal(v, python_type, TypeEngine.to_literal_type(python_type))
            named.append(run_definition_pb2.NamedLiteral(name=output_name, value=lit))
        except TypeTransformerFailedError as e:
            raise flyte.errors.RuntimeDataValidationError(output_name, e, task_name)

    return Outputs(proto_outputs=run_definition_pb2.Outputs(literals=named))

At module initialization, _register_default_type_transformers() registers the built-in scalar, collection, union, dictionary, enum, and Pydantic handlers. Dataclass handling is supplied by the object transformer in the same type engine:

TypeEngine.register(IntTransformer)
TypeEngine.register(FloatTransformer)
TypeEngine.register(StrTransformer)
TypeEngine.register(DatetimeTransformer)
TypeEngine.register(DateTransformer)
TypeEngine.register(TimedeltaTransformer)
TypeEngine.register(BoolTransformer)
TypeEngine.register(NoneTransformer, [None])
TypeEngine.register(ListTransformer())
TypeEngine.register(UnionTransformer(), [UnionType])
TypeEngine.register(DictTransformer())
TypeEngine.register(EnumTransformer())
TypeEngine.register(ProtobufTransformer())
TypeEngine.register(PydanticTransformer())

Primitive values

SimpleTransformer is the reusable adapter for scalar types whose conversion is expressed by a pair of callables. Its constructor receives a name, an exact Python type, a Flyte LiteralType, and the to/from-literal callables. get_literal_type() always returns the configured literal type; to_literal() delegates only after checking the runtime type exactly:

if type(python_val) is not self._type:
    raise TypeTransformerFailedError(
        f"Expected value of type {self._type} but got '{python_val}' of type {type(python_val)}"
    )
return self._to_literal_transformer(python_val)

That is an exact type(...) is ... check rather than an isinstance check. A subclass can therefore be rejected even when it is related to the registered type. Deserialization similarly verifies the exact result type after invoking the configured decoder. For MessagePack binary literals, SimpleTransformer.from_binary_idl() uses Mashumaro's MessagePackDecoder for datetime.date, datetime.datetime, and datetime.timedelta; ordinary scalar binary values use msgpack.loads() and are checked against the expected type. The source notes that the date transformer represents dates using Flyte's DATETIME primitive and converts them at midnight.

The default registrations cover int, float, str, datetime, date, timedelta, bool, and None (NoneType). You get these through type annotations and TypeEngine; the registered SimpleTransformer instances are not generally user-created.

Homogeneous lists

Use a parameterized list when the element type must be carried into the Flyte interface—for example, typing.List[int] or the equivalent built-in generic. ListTransformer supports one generic subtype. get_sub_type_or_none() extracts the first argument, including from an annotated list alias; get_sub_type() raises ValueError("Only generic univariate typing.List[T] type is supported.") when it cannot find one.

The resulting literal type is a Flyte collection whose collection_type is recursively obtained from TypeEngine for the element type. Serialization requires an actual list, then converts each element with the declared subtype and the expected collection subtype:

if type(python_val) is not list:
    raise TypeTransformerFailedError("Expected a list")

t = self.get_sub_type(python_type)
lit_list = [TypeEngine.to_literal(x, t, expected.collection_type) for x in python_val]

lit_list = await _run_coros_in_chunks(lit_list, batch_size=_TYPE_ENGINE_COROS_BATCH_SIZE)

return Literal(collection=LiteralCollection(literals=lit_list))

On the way back, ListTransformer.to_python_value() reads lv.collection.literals, recursively calls TypeEngine.to_python_value() for the subtype, and batches those conversions as well. A binary scalar is handled through the transformer's binary path. Untyped lists, tuples, and other non-univariate forms do not provide the subtype information that this transformer requires; tuples are explicitly registered as restricted types in _register_default_type_transformers().

_TYPE_ENGINE_COROS_BATCH_SIZE is read from _F_TE_MAX_COROS when the module loads and defaults to 10. It controls the batches used by list and dictionary element conversions; changing the environment after module import does not change the already-read value.

Dictionaries: maps versus MessagePack structs

DictTransformer has two deliberately different representations:

• Dict[str, T] becomes a Flyte map, with the value type recursively represented as map_value_type.
• An untyped or otherwise non-map dictionary becomes a STRUCT literal and is transported as MessagePack binary.

get_literal_type() makes this split. It calls extract_types() to unwrap Annotated and inspect dictionary arguments. Only a dictionary whose first type argument is exactly str takes the map path:

def get_literal_type(self, t: Type[dict]) -> LiteralType:
    tp = DictTransformer.extract_types(t)

    if tp:
        if tp[0] is str:
            sub_type = TypeEngine.to_literal_type(cast(type, tp[1]))
            return types_pb2.LiteralType(map_value_type=sub_type)
    return types_pb2.LiteralType(
        simple=types_pb2.SimpleType.STRUCT,
        annotation=TypeAnnotation(annotations={CACHE_KEY_METADATA: {SERIALIZATION_FORMAT: MESSAGEPACK}}),
    )

For the map path, every key must be an actual str; otherwise to_literal() raises ValueError("Flyte MapType expects all keys to be strings"). Values are recursively passed to TypeEngine using the declared dictionary value type. Consequently, Dict[int, T] is not represented as a Flyte map; it follows the STRUCT/MessagePack path instead.

The STRUCT path uses MessagePackEncoder and emits a binary literal tagged MESSAGEPACK. MessagePack decoding uses strict_map_key=False in the Mashumaro decoder, which allows dictionaries with non-string keys to be reconstructed when the expected Python annotation supplies the key type. A protobuf generic Struct is also accepted as a compatibility input and is converted through JSON, MessagePack, and a cached Mashumaro decoder.

Pickle is not the default. If an Annotated dictionary carries metadata containing an OrderedDict with allow_pickle=True, a MessagePack TypeError can fall back to FlytePickle.to_pickle(). Deserialization recognizes the resulting metadata["format"] == "pickle" and calls FlytePickle.from_pickle(). This fallback is opt-in and should only be used where the matching Python/runtime compatibility is available.

For a typed map literal, the destination annotation must also provide matching subtype hints. If those hints are absent, or the destination key type is not str, to_python_value() raises a type-mismatch error rather than implicitly converting the map.

Enums

EnumTransformer maps an enum.Enum class to a Flyte EnumType. The values in the class must be strings:

values = [v.value for v in t]
if not isinstance(values[0], str):
    raise TypeTransformerFailedError("Only EnumTypes with value of string are supported")
return LiteralType(enum_type=types_pb2.EnumType(values=values))

An Annotated enum is explicitly rejected because Flyte annotations are not supported for enums. Serialization also checks that the value is an enum and that its .value is exactly a string, then stores that value as a primitive string. Deserialization calls the declared enum type with the literal string. assert_type() accepts an enum instance or raw value only when that value occurs among the declared members.

When only a Flyte EnumType is available, guess_python_type() creates a dynamic enum named DynamicEnum, with generated member names based on the enum values. An empty enum is an edge case: the implementation indexes values[0] while constructing its literal type.

Dataclasses

DataclassTransformer publishes user dataclasses as STRUCT types, but its primary v2 value transport is MessagePack binary tagged MESSAGEPACK. get_literal_type() also attempts to add a Mashumaro JSON schema as literal metadata and adds cache-key metadata identifying the serialization format as msgpack. It strips Annotated wrappers before processing the dataclass itself.

A dataclass instance is encoded with a cached MessagePackEncoder for its Python type. The transformer also retains a compatibility path for dictionaries:

async def to_literal(self, python_val: T, python_type: Type[T], expected: LiteralType) -> Literal:
    if isinstance(python_val, dict):
        msgpack_bytes = msgpack.dumps(python_val)
        return Literal(scalar=Scalar(binary=Binary(value=msgpack_bytes, tag=MESSAGEPACK)))

    if not dataclasses.is_dataclass(python_val):
        raise TypeTransformerFailedError(
            f"{type(python_val)} is not of type @dataclass, only Dataclasses are supported for "
            f"user defined datatypes in Flytekit"
        )

    try:
        encoder = self._msgpack_encoder[python_type]
    except KeyError:
        encoder = MessagePackEncoder(python_type)
        self._msgpack_encoder[python_type] = encoder

    try:
        msgpack_bytes = encoder.encode(python_val)
    except NotImplementedError:
        raise NotImplementedError(
            f"{python_type} should inherit from mashumaro.types.SerializableType"
            f" and implement _serialize and _deserialize methods."
        )

    return Literal(scalar=Scalar(binary=Binary(value=msgpack_bytes, tag=MESSAGEPACK)))

Fields such as FlyteFile, FlyteDirectory, and StructuredDataset therefore need Mashumaro-compatible serialization hooks when their encoding is not built in. The error explicitly directs a custom type to inherit mashumaro.types.SerializableType and implement _serialize and _deserialize.

On decode, from_binary_idl() uses DataClassJSONMixin.from_json() for dataclasses that provide that Mashumaro mixin; otherwise it uses a cached MessagePackDecoder. to_python_value() also accepts a generic protobuf Struct compatibility value and chooses either from_json() or a cached JSONDecoder.

Schema extraction is best-effort. get_literal_type() catches schema-generation exceptions, logs an error, and continues without usable schema metadata. The source message specifically suggests removing DataClassJsonMixin and dataclass_json if schema generation fails. Separately, assert_type() checks dataclass fields, including nested dataclasses and Optional fields, and rejects missing or extra dictionary keys and mismatched field types.

The reverse path can construct a Python dataclass from schema metadata. guess_python_type() uses the schema title and convert_mashumaro_json_schema_to_python_class(). The transformer keeps generated results stable for repeated calls so remote and CLI flows do not receive a new, merely equivalent dataclass class each time.

Flyte's dataframe integration follows the same serialization contract: io/_dataframe/dataframe.py defines DataFrame as a SerializableType and DataClassJSONMixin, and its transformer calls TypeEngine.to_literal_type(DataFrame). This is a concrete example of a richer Flyte type participating in the type engine rather than bypassing it.

Pydantic models

PydanticTransformer handles Pydantic v2 BaseModel subclasses. It publishes simple=STRUCT, uses model_json_schema() as metadata, and adds the same MessagePack serialization annotation used by dataclasses. To serialize, it calls model_dump_json(), parses the JSON into a dictionary, and encodes that dictionary as MessagePack binary.

The transformer disables TypeEngine-level type assertions (enable_type_assertions=False in its constructor), leaving validation to Pydantic. Binary deserialization decodes MessagePack, converts the result to JSON, and calls:

python_val = expected_python_type.model_validate_json(
    json_data=json_str, strict=False, context={"deserialize": True}
)

Generic protobuf Struct values are also accepted by to_python_value(), which converts them with MessageToJson before calling model_validate_json(..., strict=False, context={"deserialize": True}). The deserialize context is therefore present on both model reconstruction paths.

Unions and Optional values

UnionTransformer handles typing.Union and the Python 3.10 union form registered as types.UnionType. It creates a Flyte union type containing one recursively generated variant per member. Each variant receives a structure tag containing the selected transformer name.

Serialization first prefers an exact runtime match. If type(python_val) is one of the declared union members, that member's transformer is used immediately. This explicit preference matters for overlapping Python relationships such as bool and int. If there is no exact match, the transformer probes the variants in declaration order. More than one successful probe is treated as ambiguous and raises TypeError, rather than silently selecting a structural match.

The source includes this real Union example inside SimpleTransformer because MessagePack decoding must respect the expected subtype:

@dataclass
class DC:
    a: Union[int, bool, str, float]
    b: Union[int, bool, str, float]

@task(container_image=custom_image)
def add(a: Union[int, bool, str, float],
    b: Union[int, bool, str, float]) -> Union[int, bool, str, float]:
    return a + b

@workflow
def wf(dc: DC) -> Union[int, bool, str, float]:
    return add(dc.a, dc.b)

wf(DC(1, 1))

On deserialization, UnionTransformer reads the stored transformer tag and checks whether the corresponding literal type is castable to the expected variant. If no tag is available, it tries every variant; multiple successful conversions raise an ambiguity error. guess_python_type() reconstructs a typing.Union from the variant literal types.

is_optional_type() identifies a union containing None, including Optional-shaped unions. get_sub_type_in_optional() returns get_args(t)[0]; it is used for Optional handling in field checks and is not a general-purpose union simplifier. Runtime input conversion also recognizes an optional annotation when filling a missing default whose value is None.

Troubleshooting common type failures

Symptom	What flyte-sdk is checking	What to change
A list annotation cannot produce a literal type	ListTransformer requires one generic subtype	Declare a homogeneous parameterized list such as List[T]; untyped lists are not supported as typed lists.
A dictionary with integer keys is rejected as a map	Flyte map literals require keys whose declared and runtime type is str	Use Dict[str, T] for a map, or use the untyped/STRUCT MessagePack path for non-string keys.
A typed map cannot be converted back	The destination dictionary lacks matching subtype hints or does not accept string keys	Preserve the Dict[str, T] annotation on the destination; the transformer does not perform implicit conversions.
An enum fails during interface construction or serialization	Enum values must be strings; Annotated enums are rejected	Use a non-empty string-valued enum without Annotated metadata.
A dataclass field cannot be encoded	Mashumaro cannot serialize a custom field	Implement Mashumaro SerializableType hooks (_serialize and _deserialize) for the custom Flyte datatype.
Dataclass schema metadata is absent	Schema generation is caught and logged as an error	Inspect the logged exception; the source specifically suggests removing DataClassJsonMixin and dataclass_json decorators when they cause schema extraction to fail.
A union reports an ambiguous variant	Multiple union transformers successfully accept the value or literal	Use non-overlapping variants or a more specific declared type; ambiguity is raised by design of the conversion behavior.
Dictionary MessagePack encoding fails	Pickle fallback is disabled by default	Add Annotated metadata containing an OrderedDict with allow_pickle=True only when the FlytePickle compatibility constraints are acceptable.

The examples above are source/runtime examples from types/_type_engine.py and _internal/runtime/convert.py. The repository search used for this reference found no separate test, demo, or README examples matching these patterns; they should therefore be read as demonstrations of the implemented call paths, not as copied test guarantees.