Type System and Type Engine

Every Python value that crosses a task boundary in Flyte — an int passed as input, a List[str] returned as output, a dataclass threaded through a workflow — must be serialized to Flyte's wire format (Literal) on one side and deserialized back on the other. Flytekit's type engine is the machinery that performs these conversions, and its design lets you extend it for your own types.

TypeEngine: The Central Registry

TypeEngine (in flytekit/core/type_engine.py) is the dispatch hub. It maintains a class-level dictionary _REGISTRY that maps Python types to TypeTransformer instances. Every transformer that ships with flytekit — and every transformer added by a plugin — ends up in this registry.

The three most-called methods are:

• TypeEngine.to_literal_type(python_type) — converts a Python type annotation (e.g., List[int]) to a Flyte LiteralType
• TypeEngine.to_literal(ctx, python_val, python_type, expected) — serializes a runtime value to a Literal
• TypeEngine.to_python_value(ctx, lv, expected_python_type) — deserializes a Literal back to a Python value

from flytekit.core.type_engine import TypeEngine
from flytekit.core.context_manager import FlyteContext
import typing

ctx = FlyteContext.current_context()

# Convert a Python type to a Flyte LiteralType
lt = TypeEngine.to_literal_type(int)
# lt.simple == SimpleType.INTEGER

# Nested generics work the same way
lt = TypeEngine.to_literal_type(typing.Dict[str, typing.List[typing.Dict[str, int]]])
# lt.map_value_type.collection_type.map_value_type.simple == SimpleType.INTEGER

Transformer Lookup Order

TypeEngine.get_transformer(python_type) implements a layered lookup:

1. Direct registry match — checks _REGISTRY[python_type] exactly
2. Annotated unwrapping — if the type is Annotated[T, ...] and any annotation is itself a TypeTransformer instance, that transformer is returned directly (bypassing the registry entirely)
3. Generic origin match — for List[int], checks _REGISTRY[list]
4. MRO traversal — walks the class hierarchy with inspect.getmro() to find a transformer for any base class
5. Dataclass fallback — if dataclasses.is_dataclass(python_type) is true, returns DataclassTransformer (stored separately as TypeEngine._DATACLASS_TRANSFORMER, not in _REGISTRY)
6. Pickle fallback — anything unrecognized goes through FlytePickleTransformer with a warning

from flytekit.core.type_engine import TypeEngine, ListTransformer, DictTransformer, SimpleTransformer
import typing, os

assert type(TypeEngine.get_transformer(typing.List[int])) == ListTransformer
assert type(TypeEngine.get_transformer(list)) == ListTransformer

assert type(TypeEngine.get_transformer(typing.Dict[str, int])) == DictTransformer
assert type(TypeEngine.get_transformer(dict)) == DictTransformer

assert type(TypeEngine.get_transformer(int)) == SimpleTransformer
assert type(TypeEngine.get_transformer(os.PathLike)) != SimpleTransformer  # FlyteFilePathTransformer

Lazy-Loaded Optional Transformers

When get_transformer is first called, TypeEngine.lazy_import_transformers() runs — but only once, protected by a threading.Lock. This method checks which optional libraries are already imported and registers their transformers on demand:

@classmethod
def lazy_import_transformers(cls):
    with cls.lazy_import_lock:
        if is_imported("tensorflow"):
            from flytekit.extras import tensorflow  # registers TensorFlowModelTransformer
        if is_imported("torch"):
            from flytekit.extras import pytorch    # registers PyTorchTensorTransformer
        if is_imported("pydantic"):
            from flytekit.extras import pydantic_transformer  # registers PydanticTransformer
        # ... pandas, numpy, PIL, sklearn, StructuredDataset

This means importing torch before calling any type engine method is sufficient — the PyTorch transformer registers itself automatically.

TypeTransformer: The Base Contract

Every type transformer subclasses the abstract TypeTransformer[T] class. Three methods are mandatory:

from flytekit.core.type_engine import TypeTransformer
from flytekit.models.types import LiteralType
from flytekit.models.literals import Literal

class TypeTransformer(typing.Generic[T]):
    @abstractmethod
    def get_literal_type(self, t: Type[T]) -> LiteralType:
        """Python type → Flyte LiteralType (used at compile/registration time)"""

    @abstractmethod
    def to_literal(self, ctx, python_val, python_type, expected) -> Literal:
        """Python value → Flyte Literal (serialization)"""

    @abstractmethod
    def to_python_value(self, ctx, lv, expected_python_type) -> Optional[T]:
        """Flyte Literal → Python value (deserialization)"""

Optional overrides add capability:

• assert_type(t, v) — validates a runtime value before serialization; enabled/disabled by type_assertions_enabled
• guess_python_type(literal_type) — reverse-maps a LiteralType to a Python type; used when type information is not available at call time
• from_binary_idl(binary_idl_object, expected_python_type) — handles the MessagePack Binary scalar format; the base class implementation uses mashumaro's MessagePackDecoder
• to_html(ctx, python_val, expected_python_type) — returns an HTML string for Flyte Deck rendering

AsyncTypeTransformer

Types that require I/O — file uploads, dataset writes — use AsyncTypeTransformer[T], which replaces to_literal and to_python_value with async counterparts:

class AsyncTypeTransformer(TypeTransformer[T]):
    @abstractmethod
    async def async_to_literal(self, ctx, python_val, python_type, expected) -> Literal: ...

    @abstractmethod
    async def async_to_python_value(self, ctx, lv, expected_python_type) -> Optional[T]: ...

The synchronous to_literal and to_python_value methods on AsyncTypeTransformer exist as wrappers — they call the async versions through loop_manager.synced(), allowing synchronous callers to invoke async-capable transformers. ListTransformer, DictTransformer, UnionTransformer, FlyteFile, FlyteDirectory, and StructuredDataset all use this pattern.

Primitive Types: SimpleTransformer

For scalars, flytekit uses SimpleTransformer — a concrete subclass that takes lambda functions instead of requiring a full class definition. All the built-in primitive transformers are SimpleTransformer instances:

# From flytekit/core/type_engine.py

IntTransformer = SimpleTransformer(
    "int",
    int,
    _type_models.LiteralType(simple=_type_models.SimpleType.INTEGER),
    lambda x: Literal(scalar=Scalar(primitive=Primitive(integer=x))),
    _handle_flyte_console_float_input_to_int,
)

FloatTransformer = SimpleTransformer(
    "float",
    float,
    _type_models.LiteralType(simple=_type_models.SimpleType.FLOAT),
    lambda x: Literal(scalar=Scalar(primitive=Primitive(float_value=x))),
    _check_and_covert_float,
)

StrTransformer = SimpleTransformer(
    "str",
    str,
    _type_models.LiteralType(simple=_type_models.SimpleType.STRING),
    lambda x: Literal(scalar=Scalar(primitive=Primitive(string_value=x))),
    lambda x: x.scalar.primitive.string_value,
)

DatetimeTransformer = SimpleTransformer(
    "datetime",
    datetime.datetime,
    _type_models.LiteralType(simple=_type_models.SimpleType.DATETIME),
    lambda x: Literal(scalar=Scalar(primitive=Primitive(datetime=x))),
    lambda x: x.scalar.primitive.datetime,
)

DateTransformer maps datetime.date to SimpleType.DATETIME by combining the date with datetime.time.min on serialization and calling .date() on deserialization. TimedeltaTransformer maps to SimpleType.DURATION.

All of these are registered at module import time by _register_default_type_transformers(), which is called at the bottom of type_engine.py.

One notable detail in IntTransformer: the deserialization lambda is _handle_flyte_console_float_input_to_int instead of a simple lambda x: x.scalar.primitive.integer. Flyte Console is written in JavaScript, which has a single number type, so integer values can arrive as floats. The helper handles this conversion with a logged warning about potential precision loss.

Container Transformers

Lists

ListTransformer maps typing.List[T] to Flyte's LiteralCollection. Since list serialization can involve many items — each potentially requiring a remote file operation — it implements AsyncTypeTransformer and processes items in batches:

# Batch size is controlled by the _F_TE_MAX_COROS environment variable (default 10)
_TYPE_ENGINE_COROS_BATCH_SIZE = int(os.environ.get("_F_TE_MAX_COROS", "10"))

During serialization, it calls TypeEngine.async_to_literal for each element and gathers them with _run_coros_in_chunks. During deserialization, it reads the element type from expected_python_type via get_sub_type():

ctx = FlyteContext.current_context()

# Deserializing a LiteralCollection back to a Python list
from flytekit.models.literals import Literal, LiteralCollection, Scalar, Primitive

l0 = Literal(scalar=Scalar(primitive=Primitive(integer=3)))
l1 = Literal(scalar=Scalar(primitive=Primitive(integer=4)))
lit = Literal(collection=LiteralCollection(literals=[l0, l1]))

result = TypeEngine.to_python_value(ctx, lit, typing.List[int])
# result == [3, 4]

ListTransformer.get_sub_type(t) extracts the element type from List[T]; a bare list or unparameterized List without __args__ results in get_sub_type_or_none() returning None and ultimately a ValueError.

Dicts

DictTransformer takes two distinct paths depending on the key type:

• Dict[str, T] → Flyte MapType (LiteralMap), where each key is a string and each value is a Literal for type T
• Dict[int, T], dict, or untyped dict → Flyte STRUCT with MessagePack Binary IDL encoding

# Dict[str, T] produces a map type
lt = TypeEngine.to_literal_type(typing.Dict[str, int])
assert lt.map_value_type.simple == SimpleType.INTEGER

# dict or Dict[int, str] produces STRUCT
lt = TypeEngine.to_literal_type(dict)
assert lt.simple == SimpleType.STRUCT

For map-typed dicts, keys must be strings during serialization — a non-string key raises ValueError("Flyte MapType expects all keys to be strings"). The STRUCT path uses mashumaro's MessagePackEncoder.

You can annotate a dict with allow_pickle=True to fall back to FlytePickle when MessagePack encoding fails:

from flytekit import kwtypes
from typing_extensions import Annotated

# Falls back to pickle if dict is not MessagePack-serializable
MyDict = Annotated[dict, kwtypes(allow_pickle=True)]

Unions and Optional

UnionTransformer handles typing.Union[T1, T2, ...] (and Optional[T], which is Union[T, None]). The Python 3.10+ union syntax str | int is also supported by registering the transformer for types.UnionType.

Serialization tries each variant's transformer in order, collecting the first success into a Literal with the matched type stored as a tag:

pt = typing.Union[str, int]
lt = TypeEngine.to_literal_type(pt)
# lt.union_type.variants == [
#     LiteralType(simple=STRING, structure=TypeStructure(tag="str")),
#     LiteralType(simple=INTEGER, structure=TypeStructure(tag="int")),
# ]

ctx = FlyteContext.current_context()
lv = TypeEngine.to_literal(ctx, 3, pt, lt)
# lv.scalar.union.stored_type.structure.tag == "int"
# lv.scalar.union.value.scalar.primitive.integer == 3

v = TypeEngine.to_python_value(ctx, lv, pt)
# v == 3

During deserialization, UnionTransformer reads stored_type.structure.tag from the literal and skips transformers whose name doesn't match, making reverse-mapping unambiguous.

Ambiguity errors: If two transformers both successfully serialize a value (e.g., a plain Literal with no union wrapper passed to Union[int, MyInt] where both have SimpleType.INTEGER), UnionTransformer raises TypeError("Ambiguous choice of variant for union type"). The tag stored during serialization prevents this during deserialization, but it can appear if you construct a bare literal manually and try to deserialize it into a Union type.

DataclassTransformer

The DataclassTransformer handles Python @dataclass types. It's the most feature-rich transformer in flytekit and is stored separately from _REGISTRY as TypeEngine._DATACLASS_TRANSFORMER — it's applied as a fallback after MRO traversal, so user-registered transformers for dataclass-like objects take precedence.

Serialization Format

Since flytekit 1.14.0, the default serialization format is MessagePack Binary IDL using mashumaro's MessagePackEncoder:

@dataclasses.dataclass
class MyData:
    x: int
    y: str

# Serializes to Literal(scalar=Scalar(binary=Binary(value=<msgpack bytes>, tag="msgpack")))

The old format (JSON → protobuf Struct) is still available via the FLYTE_USE_OLD_DC_FORMAT environment variable:

FLYTE_USE_OLD_DC_FORMAT=true pyflyte run my_workflow.py wf

Two mashumaro mixin protocols are supported:

• DataClassJSONMixin (from mashumaro.mixins.json) — uses python_val.to_json() / expected_python_type.from_json(), allowing mashumaro-level customization of serialization hooks
• Plain dataclasses — uses mashumaro's MessagePackEncoder/MessagePackDecoder objects cached per type in self._msgpack_encoder and self._msgpack_decoder dicts

JSON Schema Extraction

get_literal_type() attempts to extract a JSON Schema from the dataclass definition and stores it as LiteralType.metadata. This metadata powers autocomplete in the Flyte Console. It first tries mashumaro's build_json_schema, then falls back to marshmallow-jsonschema for classes mixing in DataClassJsonMixin (from dataclasses_json):

# The resulting LiteralType carries schema metadata for UI/CLI tooling
lt = TypeEngine.to_literal_type(MyData)
assert lt.simple == SimpleType.STRUCT
assert lt.metadata is not None  # JSON Schema dict

Nested Flyte Types

The _make_dataclass_serializable() method walks the dataclass fields before encoding and converts string paths to FlyteFile or FlyteDirectory objects. This means a dataclass field of type FlyteFile can be assigned a plain string path ("s3://my-bucket/file.txt"), and flytekit will silently convert it — but this behavior is deprecated and will be removed in a future version. Assign FlyteFile("s3://my-bucket/file.txt") directly.

Int Preservation After JSON Round-Trip

Protobuf Struct (used in the old format) does not support explicit integer types — all numbers are upcast to double. _fix_dataclass_int() walks the dataclass tree after JSON deserialization and recasts float values back to int wherever the field type annotation says int:

@dataclasses.dataclass
class InnerStruct(DataClassJsonMixin):
    a: int
    b: typing.Optional[str]
    c: typing.List[int]

ctx = FlyteContext.current_context()
tf = DataclassTransformer()
o = InnerStruct(a=5, b=None, c=[1, 2, 3])
lv = tf.to_literal(ctx, o, InnerStruct, tf.get_literal_type(InnerStruct))
ot = tf.to_python_value(ctx, lv=lv, expected_python_type=InnerStruct)
assert ot == o  # int fields preserved correctly

The new MessagePack format avoids this issue entirely since MessagePack preserves integer types natively.

EnumTransformer

EnumTransformer handles enum.Enum subclasses, serializing to a Primitive(string_value=...). Only string-valued enums are supported. Attempting to use an integer-valued enum raises TypeTransformerFailedError:

from enum import Enum

class Color(Enum):
    RED = "red"
    GREEN = "green"
    BLUE = "blue"

lt = TypeEngine.to_literal_type(Color)
# lt.enum_type.values == ["red", "green", "blue"]

ctx = FlyteContext.current_context()
lv = TypeEngine.to_literal(ctx, Color.RED, Color, lt)
# lv.scalar.primitive.string_value == "red"

v = TypeEngine.to_python_value(ctx, lv, Color)
# v == Color.RED

# This raises TypeTransformerFailedError:
class BadEnum(Enum):
    A = 1  # integer value — not supported

Multi-inheritance enums like class Color(str, Enum) are correctly routed to EnumTransformer rather than StrTransformer. This works because TypeEngine._get_transformer() checks issubclass(python_type, enum.Enum) before performing the MRO traversal that would otherwise find str first:

class MultiInheritanceColor(str, Enum):
    RED = auto()

assert isinstance(TypeEngine.get_transformer(MultiInheritanceColor), EnumTransformer)

ProtobufTransformer

ProtobufTransformer handles google.protobuf.Message subclasses. It serializes protobuf messages to Flyte's Scalar.generic (a protobuf Struct) and embeds the full module.classname as type metadata under the key "pb_type":

from flyteidl.core import errors_pb2

pb = errors_pb2.ContainerError(code="code", message="message")
lt = TypeEngine.to_literal_type(errors_pb2.ContainerError)
# lt.simple == SimpleType.STRUCT
# lt.metadata["pb_type"] == "flyteidl.core.errors_pb2.ContainerError"

lv = TypeEngine.to_literal(ctx, pb, errors_pb2.ContainerError, lt)
recovered = TypeEngine.to_python_value(ctx, lv, errors_pb2.ContainerError)
assert recovered == pb

guess_python_type() uses load_type_from_tag() to reconstruct the class from the "pb_type" metadata string, enabling round-trip type inference without the original Python type annotation.

Restricted Types

tuple, typing.Tuple, and NamedTuple are registered with RestrictedTypeTransformer via TypeEngine.register_restricted_type(). Any attempt to use them as a task input or output — calling get_literal_type(), to_literal(), or to_python_value() — raises RestrictedTypeError. Task functions can return NamedTuple at the workflow level, but NamedTuple cannot be nested inside another type.

LiteralsResolver

When working with execution results through FlyteRemote, the raw output is a LiteralMap — a dict-like structure of str → Literal. LiteralsResolver wraps this map and provides lazy, type-aware conversion:

from flytekit.core.type_engine import LiteralsResolver
from flytekit.models.literals import Literal, LiteralCollection, Scalar, Primitive
import typing

# Simple usage with explicit type hints
lit_dict = {
    "result": Literal(
        collection=LiteralCollection(literals=[
            Literal(scalar=Scalar(primitive=Primitive(integer=1))),
            Literal(scalar=Scalar(primitive=Primitive(integer=2))),
        ])
    )
}
lr = LiteralsResolver(lit_dict)
out = lr.get("result", typing.List[int])
# out == [1, 2]

If you provide a variable_map (the VariableMap from a task's interface), LiteralsResolver can call TypeEngine.guess_python_type() to infer types when no explicit hint is given:

lr = LiteralsResolver(lm, variable_map=variable_map, ctx=ctx)

# [] operator triggers guess_python_type() from variable_map
result = lr["my_list"]  # guesses List[int] from the LiteralType

# Explicit type overrides guessing
result = lr.get("my_list", as_type=typing.List[int])

Converted values are cached in _native_values, so repeated access to the same key doesn't re-invoke the type engine. Use update_type_hints() to add explicit type annotations before accessing — useful when you want column-specific StructuredDataset types rather than the unconstrained guessed version.

Extending the Type System

To add support for a new Python type, subclass TypeTransformer and register it with TypeEngine.register(). The PydanticTransformer in flytekit/extras/pydantic_transformer/transformer.py is a complete, production example:

import json, msgpack
from pydantic import BaseModel
from flytekit.core.type_engine import TypeEngine, TypeTransformer, TypeTransformerFailedError
from flytekit.models.literals import Binary, Literal, Scalar
from flytekit.models.types import LiteralType, TypeStructure
from flytekit.models import types
from flytekit.models.annotation import TypeAnnotation as TypeAnnotationModel
from flytekit.core.constants import CACHE_KEY_METADATA, MESSAGEPACK, SERIALIZATION_FORMAT

class PydanticTransformer(TypeTransformer[BaseModel]):
    def __init__(self):
        super().__init__("Pydantic Transformer", BaseModel, enable_type_assertions=False)

    def get_literal_type(self, t: Type[BaseModel]) -> LiteralType:
        schema = t.model_json_schema()
        literal_type = {}
        for name, python_type in t.__annotations__.items():
            try:
                literal_type[name] = TypeEngine.to_literal_type(python_type)
            except Exception:
                pass
        ts = TypeStructure(tag="", dataclass_type=literal_type)
        return types.LiteralType(
            simple=types.SimpleType.STRUCT,
            metadata=schema,
            structure=ts,
            annotation=TypeAnnotationModel({CACHE_KEY_METADATA: {SERIALIZATION_FORMAT: MESSAGEPACK}}),
        )

    def to_literal(self, ctx, python_val, python_type, expected):
        json_str = python_val.model_dump_json()
        dict_obj = json.loads(json_str)
        msgpack_bytes = msgpack.dumps(dict_obj)
        return Literal(scalar=Scalar(binary=Binary(value=msgpack_bytes, tag=MESSAGEPACK)))

    def to_python_value(self, ctx, lv, expected_python_type):
        if lv and lv.scalar and lv.scalar.binary is not None:
            return self.from_binary_idl(lv.scalar.binary, expected_python_type)
        json_str = _json_format.MessageToJson(lv.scalar.generic)
        return expected_python_type.model_validate_json(json_str, strict=False)

TypeEngine.register(PydanticTransformer())

A few things to note from this example:

• enable_type_assertions=False disables the base class isinstance check, which would fail because BaseModel subclasses don't match isinstance(v, BaseModel) the way plain classes do
• get_literal_type() delegates field types to TypeEngine.to_literal_type() recursively, so nested Flyte types are represented correctly
• to_python_value() handles both the new Binary IDL path (via from_binary_idl()) and the old protobuf Struct path for backward compatibility with Flyte Console inputs

Inline Transformers with Annotated

Rather than registering globally, you can attach a TypeTransformer instance directly as an annotation. TypeEngine.get_transformer() extracts it from get_args(python_type) before consulting the registry:

from typing_extensions import Annotated

class JsonTypeTransformer(TypeTransformer[T]):
    def get_literal_type(self, t):
        return LiteralType(simple=SimpleType.STRING)

    def to_literal(self, ctx, python_val, python_type, expected):
        return Literal(scalar=Scalar(primitive=Primitive(string_value=json.dumps(python_val))))

    def to_python_value(self, ctx, lv, expected_python_type):
        return json.loads(lv.scalar.primitive.string_value)

# Attach per-type, no global registration needed
MyJsonDict = Annotated[typing.Dict[str, int], JsonTypeTransformer(name="json[dict]", t=dict)]

assert TypeEngine.get_transformer(MyJsonDict) is the_transformer_instance

This pattern is useful when you want different serialization behavior for the same underlying Python type depending on context (e.g., JSON encoding vs. MessagePack encoding for the same dict).

Using SimpleTransformer for Quick Registration

For simple mappings that don't need custom logic, SimpleTransformer takes the boilerplate away:

from flytekit.core.type_engine import TypeEngine, SimpleTransformer
from flytekit.models.literals import Literal, Scalar, Primitive
from flytekit.models.types import LiteralType, SimpleType

class MyInt:
    def __init__(self, x: int):
        self.val = x

TypeEngine.register(
    SimpleTransformer(
        "MyInt",
        MyInt,
        LiteralType(simple=SimpleType.INTEGER),
        lambda x: Literal(scalar=Scalar(primitive=Primitive(integer=x.val))),
        lambda x: MyInt(x.scalar.primitive.integer),
    )
)

Common Pitfalls

Registry is global and permanent. TypeEngine.register() raises ValueError if the type is already registered. There is no unregister method. Plugin tests that register custom types must clean up manually (del TypeEngine._REGISTRY[MyType]). To override an existing registration, use TypeEngine.register_additional_type(transformer, type, override=True).

DataclassTransformer is not in _REGISTRY. It lives as TypeEngine._DATACLASS_TRANSFORMER and is applied as a fallback. This means TypeEngine.guess_python_type() must also check it separately — the method does this explicitly after exhausting the registry.

FlyteAnnotation cannot use "cache-key-metadata" as a key. The CACHE_KEY_METADATA constant is reserved for the internal serialization format annotation that DataclassTransformer and PydanticTransformer embed. TypeEngine.to_literal_type() raises AssertionError if a FlyteAnnotation contains it.

TypeTransformerFailedError inherits from TypeError, AssertionError, and ValueError. This is intentional — UnionTransformer's try/except loops need to catch conversion failures from any transformer without knowing which exception type each raises.

TypeEngine.guess_python_type() is best-effort. It iterates all registered transformers in insertion order and returns the first match. Use explicit as_type arguments (e.g., LiteralsResolver.get(key, as_type=MyClass)) wherever the Python type is known.

FLYTE_USE_OLD_DC_FORMAT migration. Workflows compiled and executed with the old JSON/Struct format will fail to deserialize if the worker runs a newer flytekit that expects Binary IDL, and vice versa. During upgrades, set FLYTE_USE_OLD_DC_FORMAT=true to maintain backward compatibility until all workers are updated.