Creating Raw Container Tasks

When you need to execute non-Python binaries, pre-packaged scripts, or custom container images directly in Flyte without wrapping them inside standard Python @task functions, ContainerTask in flyte-sdk allows you to specify the container image, entrypoint command, arguments, inputs, and outputs explicitly.

Defining a ContainerTask

ContainerTask is exported from flyte.extras (implemented in extras._container.py) and subclasses TaskTemplate with task_type="raw-container". It allows you to run any container image and map task inputs and outputs to container arguments and filesystem paths.

from flyte.extras import ContainerTask

simple_shell_task = ContainerTask(
    name="simple-shell-task",
    image="alpine:latest",
    command=["/bin/sh", "-c"],
    arguments=["echo 'Hello from Flyte ContainerTask'"],
)

Constructor Parameters

The ContainerTask class accepts the following parameters:

Parameter	Type	Default	Description
name	str	Required	Name of the task template.
image	Union[str, Image]	Required	Container image reference string (e.g., "alpine:latest" or "auto" which resolves to Image.from_debian_base()) or a flyte.Image instance.
command	List[str]	Required	The entrypoint command to run inside the container.
arguments	Optional[List[str]]	None	Arguments passed to the command.
inputs	Optional[Dict[str, Type]]	None	Mapping of input argument names to their expected Python types.
outputs	Optional[Dict[str, Type]]	None	Mapping of output names to their expected Python types.
input_data_dir	str | pathlib.Path	"/var/inputs"	Target filesystem directory inside the container where input files/directories are staged.
output_data_dir	str | pathlib.Path	"/var/outputs"	Directory inside the container where the task command must write its output files.
metadata_format	Literal["JSON", "YAML", "PROTO"]	"JSON"	Format enum for Flyte data loading metadata.
local_logs	bool	True	Whether to stream container stdout/stderr to the console during local execution.

Handling Scalar Inputs with Template Syntax

For primitive scalar inputs (such as str, int, or float), ContainerTask supports templated substitution inside command and arguments using the {{.inputs.<name>}} placeholder syntax.

from flyte.extras import ContainerTask

calculate_task = ContainerTask(
    name="calculate-sum",
    image="alpine:latest",
    command=["/bin/sh", "-c"],
    arguments=[
        "echo $(( {{.inputs.a}} + {{.inputs.b}} )) > /var/outputs/result"
    ],
    inputs={"a": int, "b": int},
    outputs={"result": int},
)

During execution, _render_command_and_volume_binding identifies placeholders matching the regular expression \{\{\.inputs\.([a-zA-Z0-9_]+)\}\} and replaces each key with the string representation of its corresponding input value.

Handling File and Directory Inputs

When working with flyte.io.File or flyte.io.Dir, do not use the {{.inputs.<name>}} template syntax. Instead, reference the absolute path under input_data_dir (default: /var/inputs/).

from flyte.extras import ContainerTask
from flyte.io import File

line_counter = ContainerTask(
    name="count-lines",
    image="alpine:latest",
    command=["/bin/sh", "-c"],
    arguments=[
        "wc -l < /var/inputs/input_file > /var/outputs/line_count"
    ],
    inputs={"input_file": File},
    outputs={"line_count": int},
    input_data_dir="/var/inputs",
    output_data_dir="/var/outputs",
)

Path Matching and Volume Binding

When executing locally:

1. _extract_path_command_key matches arguments starting with input_data_dir/<name>.
2. ContainerTask extracts the key name (input_file) from /var/inputs/input_file.
3. The local path of the File or Dir (input_val.path) is bound to /var/inputs/input_file inside the container in read-write mode ({"bind": "/var/inputs/input_file", "mode": "rw"}).

If a File or Dir is passed using the template syntax {{.inputs.input_file}}, ContainerTask raises an AssertionError:

AssertionError: File and Directory commands should not use the template syntax like this: {{.inputs.infile}}
Please use a path-like syntax, such as: /var/inputs/infile.
This requirement is due to how Flyte Propeller processes template syntax inputs.

Producing and Parsing Outputs

ContainerTask expects the container process to write task outputs as individual files directly into output_data_dir (default: /var/outputs), where each filename corresponds to the key defined in the outputs dictionary.

import datetime
from flyte.extras import ContainerTask
from flyte.io import File

process_data = ContainerTask(
    name="generate-metadata",
    image="ubuntu:22.04",
    command=["/bin/bash", "-c"],
    arguments=[
        "echo '2026-03-30T12:00:00' > /var/outputs/timestamp && "
        "echo 'true' > /var/outputs/is_valid && "
        "echo '100' > /var/outputs/count && "
        "echo 'task payload' > /var/outputs/artifact"
    ],
    outputs={
        "timestamp": datetime.datetime,
        "is_valid": bool,
        "count": int,
        "artifact": File,
    },
)

When reading output files via _convert_output_val_to_correct_type, the values are converted based on their target Python type:

• bool: Evaluates output_val.lower() != "false".
• datetime.datetime: Parsed via datetime.datetime.fromisoformat(output_val).
• datetime.timedelta: Parsed using regex supporting formats such as 1 days, 02:30:00.500000 or 01:15:30.
• File: Loaded via await File.from_local(output_path).
• Dir: Loaded via await Dir.from_local(output_path).
• Other types (e.g., int, float, str): Instantiated directly using the type constructor output_type(output_val).

If an expected output file does not exist when the container finishes, output_val defaults to None.

Local Execution and Cluster Serialization

Local Execution with Docker

Calling await task.execute(**kwargs) runs the task inside your local Docker daemon:

import asyncio
from flyte.extras import ContainerTask

task = ContainerTask(
    name="local-echo",
    image="alpine:latest",
    command=["echo"],
    arguments=["Local container execution test"],
)

asyncio.run(task.execute())

During local execution:

1. docker.from_env() connects to the local Docker engine (requires the docker Python package).
2. The image is pulled if not present locally via _pull_image_if_not_exists.
3. Input volumes and a temporary local output directory (storage.get_random_local_directory()) are mounted to the container.
4. The container is executed (remove=True, detach=True), container logs are streamed to stdout if local_logs=True, and output files are collected upon completion.

Serialization for Remote Cluster Execution

When serialized for Flyte Propeller, ContainerTask implements two serialization hooks:

• container_args(sctx): Returns the concatenated command and arguments list (self._cmd + (self._args or [])).
• data_loading_config(sctx): Returns a flyteidl.core.tasks_pb2.DataLoadingConfig protobuf object containing:
  • input_path: str(self._input_data_dir)
  • output_path: str(self._output_data_dir)
  • enabled: True
  • format: Protobuf enum mapping for "JSON", "YAML", or "PROTO".

Troubleshooting and Common Gotchas

Missing Docker Package for Local Runs

If you call execute() locally without having the Docker SDK installed, an ImportError is raised:

ImportError: Docker is not installed. Please install Docker by running `pip install docker`.

Fix: Install the docker client in your Python environment:

pip install docker

Template Substitution on Files/Directories

Using {{.inputs.myfile}} when myfile is a flyte.io.File or flyte.io.Dir results in an explicit AssertionError. Fix: Change the argument to /var/inputs/myfile (or <input_data_dir>/myfile).

Missing Output Files

If the container command fails to create a file matching a key in outputs under /var/outputs, the SDK receives None for that output and passes it to the type converter, which may raise a TypeError or ValueError (e.g., int(None)). Fix: Ensure every branch in your container command creates all files specified in the outputs dictionary.