Launch plans, schedules, and fixed inputs

Choose the launch plan that represents the execution

A workflow is registered with a bare, cached launch plan when you do not need execution-specific defaults, fixed values, schedules, or other associations. Retrieve it with LaunchPlan.get_or_create(workflow=...):

@workflow
def wf(a: int, c: str) -> str:
    ...

LaunchPlan.get_or_create(workflow=wf)

get_or_create() treats an omitted name as a request for that workflow's default launch plan. Supplying any configuration while omitting the name raises ValueError; configured launch plans therefore need a unique name:

launch_plan = LaunchPlan.get_or_create(
    workflow=wf,
    name="wf-with-inputs",
    default_inputs={"a": 10},
    fixed_inputs={"c": "production"},
)

The cache is keyed by the launch-plan name, or by the workflow name for the default plan. Repeating a named request returns the cached object only when its workflow and configuration agree. A different workflow or a different schedule, input set, label, security setting, or other cached option raises AssertionError; create() rejects a duplicate name directly.

What the default plan preserves

LaunchPlan.get_default_launch_plan() derives a ParameterMap from workflow.python_interface and creates an empty Flyte LiteralMap for fixed inputs. It also copies defaults declared in the workflow signature into the plan's saved inputs. Thus the default plan uses the workflow's own signature defaults, while a named plan can add or override them.

Parameterize inputs with defaults and fixed values

Use default_inputs for values that should be used when the launch plan is called without an override. Use fixed_inputs for values that must be supplied by the launch plan and cannot be changed at launch time:

launch_plan = LaunchPlan.get_or_create(
    workflow=wf,
    name="wf-with-inputs",
    default_inputs={"a": 10},
    fixed_inputs={"c": "production"},
)

LaunchPlan.create() first transforms the workflow interface into parameters. It then transforms default_inputs through a temporary interface and updates the workflow-derived parameters, so explicit launch-plan defaults take precedence over defaults from the workflow signature. Fixed values are converted with translate_inputs_to_literals() using both the workflow's Flyte interface and Python interface. The resulting LiteralMap is stored as fixed_inputs.

During LaunchPlan.__init__(), fixed-input names are removed from the parameter map. In source terms, it filters the parameter entries with if k not in fixed_inputs.literals before constructing the new ParameterMap.

The resulting views have different purposes:

• launch_plan.parameters exposes launch-time parameters, excluding fixed-input names.
• launch_plan.fixed_inputs contains the serialized Flyte literals.
• launch_plan.saved_inputs contains the original Python values used for local invocation. It returns a copy, so updating it does not mutate the plan.

create() combines explicit defaults and fixed values in _saved_inputs with default_inputs.update(fixed_inputs). This means fixed values are absent from the launch interface but are still available when the plan forwards a local call. Values must be compatible with the workflow input types because fixed values are translated against the workflow interfaces.

Call a launch plan with keyword inputs

Launch-plan execution is keyword-only:

launch_plan(a=20)

LaunchPlan.__call__() rejects positional arguments with AssertionError. It starts from saved_inputs and overlays the keyword arguments. With a compilation state, it passes those inputs to create_and_link_node(ctx, entity=self, **inputs), making the launch plan a node in the compiled graph. Without compilation, it forwards the inputs to the underlying workflow:

LaunchPlan.__call__
  ├─ compiling: create_and_link_node(..., entity=launch_plan, ...)
  └─ local:     workflow(..., saved inputs overridden by keyword inputs)

This behavior also lets a launch plan participate where flytekit accepts callable workflow entities. For a dynamic task, list the plan in node_dependency_hints so it is registered before the dynamic task attempts to run it:

@workflow
def workflow0():
    ...

launchplan0 = LaunchPlan.get_or_create(workflow0)

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

map_task() recognizes LaunchPlan and routes it through array_node. The array-node integration excludes the launch plan's fixed-input names from the mapped interface, so a fixed value is not treated as a value to map over.

Attach a schedule or trigger

Pass a CronSchedule when the native scheduler should use a cron expression or alias:

from flytekit import CronSchedule

schedule = CronSchedule(
    schedule="*/1 * * * *",
)

launch_plan = LaunchPlan.get_or_create(
    workflow=wf,
    name="wf-every-minute",
    schedule=schedule,
)

The schedule argument is stored on LaunchPlan.schedule. CronSchedule validates aliases from its fixed alias list, or validates a schedule expression with croniter. The schedule form is the five-field native-scheduler form. The older cron_expression argument is rejected immediately as deprecated. An optional offset is checked against CronSchedule._OFFSET_PATTERN, which accepts the module's ISO-8601-like duration syntax.

To use a fixed interval, supply a datetime.timedelta to FixedRate:

from datetime import timedelta
from flytekit import FixedRate

schedule = FixedRate(duration=timedelta(minutes=10))

launch_plan = LaunchPlan.get_or_create(
    workflow=wf,
    name="wf-every-ten-minutes",
    schedule=schedule,
)

FixedRate._translate_duration() represents an exact duration as days first, then hours, otherwise minutes. Microseconds and intervals with a sub-minute remainder raise AssertionError; fixed rates therefore have whole-minute granularity.

Both schedule types accept kickoff_time_input_arg. Set it to the name of a workflow input when the workflow needs the scheduled kickoff time:

from datetime import datetime

@workflow
def my_wf(kickoff_time: datetime):
    ...

schedule = CronSchedule(
    schedule="*/1 * * * *",
    kickoff_time_input_arg="kickoff_time",
)

The source notes that the actual kickoff can be a few seconds different from the nominal schedule time because Flyte does not have an atomic clock.

Use the trigger adapter

The newer trigger form wraps either schedule type in OnSchedule:

from flytekit import CronSchedule, OnSchedule

trigger = OnSchedule(CronSchedule(schedule="*/1 * * * *"))

launch_plan = LaunchPlan.get_or_create(
    workflow=wf,
    name="wf-triggered",
    trigger=trigger,
)

LaunchPlanTriggerBase is a protocol requiring to_flyte_idl(). OnSchedule.to_flyte_idl() simply delegates to its wrapped CronSchedule or FixedRate. LaunchPlan.create() and get_or_create() retain both the model-level schedule argument and the newer trigger argument; OnSchedule itself performs no additional schedule transformation.

Reference an externally registered launch plan

Use ReferenceLaunchPlan when the launch plan already exists in Flyte and you need a local pointer for compilation. The reference is identified by project, domain, name, and version, while the caller supplies the expected interface:

@reference_launch_plan(
    project="my-project",
    domain="development",
    name="external-plan",
    version="v1",
)
def external_plan(input_value: int) -> str:
    ...

The decorator calls transform_function_to_interface(..., is_reference_entity=True) and constructs ReferenceLaunchPlan from the annotated inputs and outputs. ReferenceLaunchPlan does not make a network call to Admin to discover that interface, so the annotations must match the registered launch plan; an incorrect interface is reported later during compilation or registration.

Other launch-plan execution settings

Named launch plans can also carry notifications, labels, annotations, raw_output_data_config, max_parallelism, security_context, overwrite_cache, and auto_activate. The corresponding properties expose these values, and should_auto_activate reports the activation flag. max_parallelism controls the maximum number of task nodes that can run in parallel for the workflow; the get_or_create() documentation explicitly excludes MapTasks from that parallelism accounting.

For compatibility, auth_role is converted to a SecurityContext containing an Identity. Supplying auth_role together with security_context raises ValueError, and the source marks AuthRole usage as deprecated.

Every constructed LaunchPlan is appended to FlyteEntities.entities, allowing it to be discovered during later serialization and registration. clone_with() creates another plan for the same workflow while reusing existing values for omitted options; replacements should be chosen carefully because several fallbacks use truthiness (value or existing), and trigger is passed directly rather than inherited.

Troubleshooting common failures

Symptom	Cause in flytekit	Correction
ValueError when creating an unnamed plan	A default plan cannot have defaults, fixed inputs, schedules, triggers, or other associations.	Supply a unique name.
AssertionError for a repeated name	The cached name belongs to another workflow or was requested with different configuration.	Reuse the original configuration or choose another name.
Positional-call AssertionError	LaunchPlan.__call__() accepts keyword arguments only.	Call launch_plan(input_name=value).
Cron schedule rejected	schedule is neither a supported alias nor a croniter-parseable expression; cron_expression is also deprecated and rejected.	Use the five-field schedule form or a supported alias.
Fixed-rate AssertionError	The duration contains microseconds or is not divisible by one minute.	Use a whole-minute, hour, or day interval.
Reference plan fails during compilation or registration	The annotated reference interface does not match the registered launch plan.	Supply the registered plan's actual input and output types.