Skip to main content

Task authoring and execution

Declaring a Task​

Flyte tasks are just typed Python functions. The smallest working example is:

from flytekit import task

@task
def my_task(x: int, y: typing.Dict[str, str]) -> str:
    ...

When you apply @task (defined in flytekit/core/task.py), flytekit inspects the function's type annotations to infer an input/output Interface, builds a TaskMetadata from the decorator arguments, looks up a registered task plugin by the type of task_config, and instantiates that plugin class — by default PythonFunctionTask from flytekit/core/python_function_task.py. Coroutine functions are automatically routed to AsyncPythonFunctionTask instead.

@task(task_config=Spark(), retries=3)
def my_task(x: int, y: typing.Dict[str, str]) -> str:
    ...

If you pass an unrecognized keyword, the decorator fails fast: raise ValueError(f"Unrecognized argument(s) for task: {kwargs.keys()}"). After construction, update_wrapper(task_instance, decorated_fn) copies the wrapped function's metadata onto the returned task object, so my_task behaves like the original function from Python's perspective.

One thing worth knowing: task construction has a side effect. Task.__init__ appends every instance to FlyteEntities.entities, which is how flytekit later finds all tasks in a module for serialization.

Configuring behavior with TaskMetadata​

TaskMetadata (in flytekit/core/base_task.py) is the dataclass that carries execution semantics: cache, cache_serialize, cache_version, cache_ignore_input_vars, interruptible, deprecated, retries, timeout, pod_template_name, generates_deck, and is_eager. Its __post_init__ enforces the rules so misconfiguration fails at definition time, not mid-run:

  • timeout may be an int (seconds, converted to datetime.timedelta) or a timedelta; anything else truthy raises ValueError.
  • cache=True requires a non-empty cache_version.
  • cache_serialize=True or cache_ignore_input_vars require cache=True.

The modern authoring path is the Cache object. When you pass cache=True without a cache_version, the decorator builds Cache(serialize=..., ignored_inputs=...) and computes a version from the function and container settings via cache.get_version(VersionParameters(...)). Mixing a Cache object with the deprecated cache_serialize/cache_version/cache_ignore_input_vars arguments raises ValueError.

Caching locally​

Task.local_execute also honors caching in your local runs: if metadata.cache is set and caching is enabled in the local config, it consults LocalTaskCache.get(...) and stores outputs with LocalTaskCache.set(...), with cache_overwrite bypassing the read.

The Task Abstraction Hierarchy​

flytekit's task classes form a layered hierarchy, each layer adding one capability:

Task                       base_task.py — FlyteIDL-oriented base: metadata, interface,
 │                         registration, __call__, local_execute
 └─ PythonTask             base_task.py — adds a native Python Interface, task_config,
    │                      environment, decks, compile()
    └─ PythonAutoContainerTask          — container/image/command serialization
       ├─ PythonFunctionTask            python_function_task.py — wraps a user function
       │  └─ AsyncPythonFunctionTask    — coroutine tasks
       │     └─ EagerAsyncPythonFunctionTask  — behind @eager
       └─ PythonInstanceTask            — tasks with no user function body

Task is the layer closest to the FlyteIDL TaskTemplate. It stores the task_type, name, TypedInterface, TaskMetadata, task_type_version, security context, and docs, and defines the execution contract via three abstract methods: pre_execute, execute, and dispatch_execute. Calling a task (t1(a=5)) goes through Task.__call__, which delegates to flyte_entity_call_handler — the same entry point that distinguishes "being invoked inside a workflow during compilation" from "being run for real."

PythonTask adds the native Python interface on top. Its constructor accepts a task_config (the plugin-specific configuration object), environment variables, and deck settings; note that disable_deck is deprecated and that supplying both disable_deck and enable_deck raises ValueError. Deck fields are validated against the DeckField enum, and docstrings become Documentation automatically. get_input_types() returns the native {name: type} mapping, and compile() calls create_and_link_node to attach the task as a node when a workflow is being built.

How a Task Actually Runs​

The local execution path in Task.local_execute shows the full pipeline:

  1. Inputs → literals. translate_inputs_to_literals converts incoming values (which may be raw Python values, Promise objects from upstream tasks, or nested lists/dicts of either) into a LiteralMap, using the task's Flyte interface and native types. On failure, the error is re-raised with task context: Failed to convert inputs of task '{self.name}'.
  2. Optional local cache (described above).
  3. sandbox_execute, which builds a sandboxed ExecutionParameters (with_task_sandbox()) and calls dispatch_execute — the method also used at real runtime.
  4. Outputs → Promises. The resulting LiteralMap is re-wrapped one output per declared interface output as Promise(var, literal), or as a single VoidPromise(self.name) when the task declares no outputs. A length mismatch raises AssertionError.

Inside PythonTask.dispatch_execute — the method invoked both locally and remotely on the cluster — the flow is:

  1. pre_execute(user_params) runs (the default returns params unchanged; plugins like Spark override it to set up a session before type conversion).
  2. self._literal_map_to_python_input(...) converts the input LiteralMap back to Python native kwargs via TypeEngine.literal_map_to_kwargs.
  3. self.execute(**native_inputs) runs your function. Locally, the original exception is re-raised with a message like Error encountered while executing '<task name>'; remotely, it's wrapped in FlyteUserRuntimeException.
  4. post_execute(...) runs (a no-op by default; it's also where IgnoreOutputs — the exception you can raise to signal outputs may be safely discarded, e.g. in distributed training — gets bubbled up).
  5. _output_to_literal_map converts native outputs back through TypeEngine.async_to_literal, attaching any output metadata (partitions, etc.). If the raw result is already a LiteralMap or a DynamicJobSpec — as happens with dynamic tasks — it's returned as-is without conversion.
  6. _write_decks emits the HTML decks if enable_deck is set and the run is local.

Dynamic Tasks​

What if the shape of your workflow isn't known until runtime — for example, you need for i in range(a) where a is an input? Ordinary workflows can't do that because they only run at compile time. @dynamic solves it:

@dynamic
def my_dynamic_subwf(a: int) -> (typing.List[str], int):
    s = []
    for i in range(a):
        s.append(t1(a=i))
    return s, 5

dynamic in flytekit/core/dynamic_workflow_task.py is literally functools.partial(task.task, execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC). At execution time, PythonFunctionTask.execute routes to dynamic_execute, which builds and caches a PythonFunctionWorkflow from your function (compile_into_workflow):

  • Remotely (execution state TASK_EXECUTION), the generated workflow is serialized and returned as a DynamicJobSpec containing the discovered task templates, nodes, outputs, and subworkflows. The Flyte engine then runs it like a subworkflow. A caveat: ReferenceTask inside a dynamic task raises ValueError("Reference tasks are currently unsupported within dynamic tasks"), because resolving reference tasks requires a network call to flyteadmin.
  • Locally, the generated workflow is executed directly and outputs are converted to a LiteralMap, mimicking a workflow's local execute.

Keep dynamic workflows small: the module docstring warns that a loop can easily produce thousands of nodes and recommends staying under roughly fifty tasks, using the map task for large identical runs.

If your dynamic body invokes a launch plan (which must be registered on flyteadmin before it can run), pass hints so it gets registered alongside the dynamic task:

@workflow
def workflow0():
    ...

launchplan0 = LaunchPlan.get_or_create(workflow0)

@dynamic(node_dependency_hints=[launchplan0])
def launch_dynamically():
    return [launchplan0]*10

Passing node_dependency_hints to a static task raises ValueError — flyte finds node dependencies automatically there.

The default resolver also can't rehydrate nested or local functions, so PythonFunctionTask rejects them with ValueError("TaskFunction cannot be a nested/inner or local function..."). Test modules beginning with test_ are exempt, and custom decorators are fine if they use functools.wraps/functools.update_wrapper.

Async and Eager Tasks​

If the decorated function is a coroutine, @task instantiates AsyncPythonFunctionTask. Its __call__ is await async_flyte_entity_call_handler(self, ...), and async_execute simply awaits the underlying function in DEFAULT mode. Dynamic mode is deliberately unsupported there and raises NotImplementedError — eager and dynamic don't mix.

EagerAsyncPythonFunctionTask (what @eager produces) goes further: instead of building a static graph, Python itself becomes the scheduler — every nested entity invocation creates an execution on the Flyte cluster. The constructor forces TaskMetadata(is_eager=True) and ExecutionBehavior.EAGER, deleting any execution_mode you supplied. A minimal runnable example:

from flytekit import task, eager

@task
def add_one(x: int) -> int:
    return x + 1

@task
def double(x: int) -> int:
    return x * 2

@eager
async def eager_workflow(x: int) -> int:
    out = add_one(x=x)
    return double(x=out)

if __name__ == "__main__":
    import asyncio
    result = asyncio.run(eager_workflow(x=1))
    print(f"Result: {result}")  # "Result: 4"

For a real backend run, point @eager at a remote:

@eager(
    remote=FlyteRemote(config=Config.auto(config_file="config.yaml")),
    client_secret_group="my_client_secret_group",
    client_secret_key="my_client_secret_key",
)
async def eager_workflow(x: int) -> int:
    out = await add_one(x)
    return await double(out)

Mechanically, remote eager execution builds a Controller backed by FlyteRemote (obtained via get_plugin().get_remote(...)), installs a signal handler for SIGINT/SIGTERM on the main thread, and puts the worker queue into the FlyteContext. Nested entity calls enqueue work and await results; afterward render_html() produces an "Eager Executions" deck, and an EagerException causes the task to raise FlyteNonRecoverableSystemException so the run doesn't show as succeeded.

Two supporting pieces handle cleanup. EagerFailureTaskResolver is a TaskResolverMixin whose load_task always reconstructs an EagerFailureHandlerTask and whose loader args are fixed to ["eager", "failure", "handler"]. EagerFailureHandlerTask.dispatch_execute (remote-only) repeatedly lists unfinished executions tagged eager-exec matching the parent's execution name and terminates them — an on-failure handler wired in by get_as_workflow, which wraps the eager task in an ImperativeWorkflow with add_on_failure_handler(cleanup).

Custom Tasks and Resolvers​

Two extension paths avoid writing a Python function body altogether:

  • PythonInstanceTask is for platform-defined tasks with no user function — you override execute. Use it as x = MyInstanceTask(name="x", ...) and then call x(a=5) per its declared interface. The class exists specifically so the module loader rehydrates the right instance at runtime by capturing the module name and variable name.

  • TaskResolverMixin defines how a containerized task is serialized and rehydrated. The docstring shows what a serialized task's container command looks like:

    pyflyte-execute --inputs s3://path/inputs.pb --output-prefix s3://outputs/location \
    --raw-output-data-prefix /tmp/data \
    --resolver flytekit.core.python_auto_container.default_task_resolver \
    -- \
    task-module repo_root.workflows.example task-name t1

    After the boilerplate arguments, flytekit appends the resolver's location followed by whatever loader_args(settings, t) returns. The default resolver emits the task's module and name, and at runtime load_task does importlib.import_module on the module and looks up the task by that key. Implement your own resolver (providing location, name, load_task, loader_args, and optionally get_all_tasks/task_name) to support tasks that don't live at module level — notebooks, for instance.

Plugin registration is the other half of custom tasks: TaskPlugins.register_pythontask_plugin(config_type, plugin_type) associates a task_config type with a PythonFunctionTask subclass. @task(task_config=MyConfig()) then instantiates your plugin; registering the same config type twice with different plugin classes raises TypeError, and async functions require the plugin to derive from AsyncPythonFunctionTask.

Wrapping Tasks: Map Task Restrictions​

map_task wraps an existing task to run it over a list, and it's picky about what it accepts:

@task
def my_mappable_task(a: int) -> typing.Optional[str]:
    return str(a)

@workflow
def my_wf(x: typing.List[int]) -> typing.List[typing.Optional[str]]:
    return map_task(
        my_mappable_task,
        metadata=TaskMetadata(retries=1),
        concurrency=10,
        min_success_ratio=0.75,
    )(a=x).with_overrides(requests=Resources(cpu="10M"))

Only default-mode PythonFunctionTask or PythonInstanceTask are accepted — @dynamic and @eager tasks are rejected — and the underlying task may declare at most one output.

Quick Limitations Reference​

  • Caching combos are validated at construction (TaskMetadata.__post_init__): no cache_version, no serialize/ignore-inputs.
  • timeout=0 is treated as unset; integer timeouts become timedelta seconds.
  • disable_deck is deprecated; specifying it and enable_deck raises ValueError. Decks are off by default.
  • Untyped outputs require pickle_untyped=True, which the source explicitly flags as not recommended for production.
  • Nested functions fail with the default resolver; use module-level definitions, functools.wraps, or a custom TaskResolverMixin.
  • No ReferenceTask inside dynamic tasks; no conditionals in eager tasks (use Python if instead).
  • When you need a programmatic interface without a Python function — e.g. for reference entities — use kwtypes: kwtypes(a=str, b=int) returns an ordered mapping usable as inputs=kwtypes(a=str, b=int) in get_reference_entity(...).