Launch plans, schedules, and fixed inputs

Suppose you've written a workflow and registered it — but you want a second execution flavor of the same workflow: same code, different inputs, run every day at noon. You don't fork the workflow; you attach a launch plan to it. In flytekit, a LaunchPlan is a named, registrable wrapper around a WorkflowBase that carries default inputs, fixed inputs, a schedule, notifications, security settings, and more. Every workflow implicitly gets a default launch plan with none of those extras.

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

LaunchPlan.get_or_create(workflow=wf)

The default launch plan

When you call LaunchPlan.get_or_create() with only a workflow and no name, flytekit builds the default launch plan via LaunchPlan.get_default_launch_plan(). This plan:

• Derives its ParameterMap from the workflow's Python interface using transform_inputs_to_parameters() — so signature defaults like def wf(a: int = 3) become parameter defaults.
• Inherits labels and annotations from workflow.default_options if present.
• Has an empty fixed_inputs (LiteralMap(literals={})) and no schedule or notifications.
• Saves the workflow's signature defaults into saved_inputs so that calling the plan locally fills them in.

The result is cached in the process-global LaunchPlan.CACHE dict keyed by workflow name, since (per the source comment) "users may get the default launch plan twice for a single Workflow. We don't want to create two defaults, could be confusing."

Customized, named launch plans

The moment you want defaults, fixed inputs, or a schedule, you must supply a unique name:

from flytekit import LaunchPlan

my_lp = LaunchPlan.get_or_create(
    name="wf_daily",
    workflow=wf,
    fixed_inputs={"c": "production"},
    default_inputs={"a": 10},
)

Trying to attach properties to the unnamed plan fails loudly:

# ValueError: Only named launchplans can be created that have other properties.
LaunchPlan.get_or_create(workflow=wf, schedule=some_schedule)

get_or_create() checks this up front and raises ValueError if name is None and any other property is set — "Default launchplans cannot have any other associations."

How LaunchPlan.create() builds the plan

LaunchPlan.create() (called by get_or_create() when a name is given) does four things worth understanding:

1. Workflow-signature parameters first. wf_signature_parameters = transform_inputs_to_parameters(ctx, workflow.python_interface) turns the workflow's signature defaults into Flyte parameters.
2. Explicit defaults overlay the signature. A temporary Interface is constructed from just the default_inputs dict, transformed again, and merged in via wf_signature_parameters._parameters.update(...) — so default_inputs have higher precedence than function-signature defaults.
3. Fixed inputs become literals. translate_inputs_to_literals() converts the native Python values into a LiteralMap, which is stored on the plan.
4. Fixed inputs disappear from the parameter map. In __init__ (source in launch_plan.py):

# Ensure fixed inputs are not in parameter map
filtered = {
    k: v
    for k, v in parameters_map.parameters.items()
    if k not in fixed_inputs.literals
}
self._parameters = _interface_models.ParameterMap(parameters=filtered)

(parameters_map here stands in for the parameters argument of LaunchPlan.__init__; the original source shadows that name, but the logic is identical.)

This is why a fixed input can't be supplied at launch time — it simply isn't part of the plan's exposed interface. If the same name appears in both default_inputs and fixed_inputs, the constructor strips it from defaults, and the fixed value wins.

For local execution and serialization convenience, create() also merges fixed inputs into a native-Python dict: default_inputs.update(fixed_inputs); lp._saved_inputs = default_inputs. Note two gotchas here: create() mutates the dict you passed in as default_inputs, and saved_inputs returns a copy (so callers updating it don't corrupt the plan).

Calling a launch plan

Launch plan calls accept keyword arguments only — __call__ raises AssertionError("Only Keyword Arguments are supported for launch plan executions") on any positional args. Two behaviors depending on context:

• During compilation (when ctx.compilation_state is not None): the call merges saved_inputs with the keyword arguments and delegates to create_and_link_node(), wiring a launch-plan node into the workflow graph.
• Local execution: the same merge happens, then the call simply forwards to self.workflow(*args, **inputs).

This is also why passing a fixed input at call time fails on the promise-building path with Fixed inputs cannot be specified — the value belongs to the plan, not the call site.

Caching and uniqueness

LaunchPlan.CACHE is process-global, and flytekit enforces consistency within it. If get_or_create() is called twice with the same name:

• For a different workflow under the same name: AssertionError — "please ensure unique names."
• For the same workflow but a different schedule, notifications, default inputs, labels, annotations, raw-output config, max parallelism, security context, overwrite-cache flag, or auto-activate flag: AssertionError — "please use different launch plan names."
• Identical configuration: the cached plan is returned.

There's also a clone_with() method that produces a new plan for the same workflow, inheriting parameters, fixed inputs, schedule, notifications, labels, annotations, raw-output config, parallelism, and security context unless overridden. Be aware of its truthiness-based fallback (schedule or self.schedule, etc.): you can't use it to clear an inherited option, and the trigger is not inherited — you must pass one explicitly.

Schedules

Two schedule types live in flytekit's schedule module, both subclassing flytekit.models.schedule.Schedule.

CronSchedule — cron-based runs via the native scheduler:

from flytekit import CronSchedule

CronSchedule(
    schedule="*/1 * * * *",  # runs every minute
)

The schedule argument accepts either a cron alias from _VALID_CRON_ALIASES (@hourly, @daily, @weekly, @monthly, @yearly, and a few non-@ variants) or a croniter-parseable five-field cron expression; _validate_schedule() falls back to croniter.croniter(schedule) for anything that isn't an alias and raises ValueError if that fails too.

An important compatibility note: the constructor still accepts a cron_expression parameter, but it immediately raises AssertionError — "cron_expression is deprecated and should not be used. Use schedule instead." The old AWS/CloudWatch-style six-field validation (_validate_expression) exists but is not the supported path; the error message explicitly points you to schedule for five-field expressions. An optional offset is validated against an ISO-8601-duration regex (_OFFSET_PATTERN).

FixedRate — interval-based runs:

from datetime import timedelta
from flytekit import FixedRate

FixedRate(duration=timedelta(minutes=10))

_translate_duration() rejects sub-minute granularity (microseconds or a non-zero seconds remainder raise AssertionError), then converts to the largest clean unit: whole days → FixedRateUnit.DAY, whole hours → FixedRateUnit.HOUR, otherwise minutes.

OnSchedule — the newer trigger syntax. LaunchPlan accepts either the older schedule= argument or the newer trigger= argument (marked "[alpha] This is a new syntax for specifying schedules" in the docstring). OnSchedule implements the LaunchPlanTriggerBase protocol and simply forwards to_flyte_idl() to the wrapped CronSchedule or FixedRate:

from flytekit import LaunchPlan, CronSchedule, OnSchedule

lp = LaunchPlan.get_or_create(
    name="wf_scheduled",
    workflow=wf,
    fixed_inputs={"c": "prod"},
    trigger=OnSchedule(CronSchedule(schedule="0 3 * * *")),  # 3am daily
)

Passing the kickoff time to your workflow

If your code needs to know when a run was kicked off, both schedule types accept kickoff_time_input_arg, naming an input of your workflow that Flyte will populate with the scheduled time:

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

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

The docstring is candid about accuracy: "until Flyte has an atomic clock, there could be a few seconds here or there" — a 3pm-UTC-Wednesday run may actually start at 15:00:02.

Launch plans inside workflows and dynamic tasks

Launch plans participate in graph construction like other flytekit entities:

• WorkflowBase.add_launch_plan() delegates to add_entity(), which calls create_node(entity=entity, **kwargs); create_node() in node_creation.py accepts LaunchPlan alongside tasks and workflows.
• Every constructed LaunchPlan appends itself to FlyteEntities.entities in __init__, which is how it gets picked up during serialization.
• Dynamic tasks that invoke a launch plan must declare it as a registration-order dependency, using node_dependency_hints — the source example in task.py shows exactly this pattern:

@workflow
def workflow0():
    ...

launchplan0 = LaunchPlan.get_or_create(workflow0)

# Specify node_dependency_hints so that launchplan0 will be registered on flyteadmin, despite this being a
# dynamic task.
@dynamic(node_dependency_hints=[launchplan0])
def launch_dynamically():
    # To run a sub-launchplan it must have previously been registered on flyteadmin.
    return [launchplan0] * 10

• When a launch plan is used as the target of a map (array) execution, ArrayNode excludes target.fixed_inputs.literals from the mapped interface — fixed inputs can't vary per mapped invocation, so they're removed from what each invocation must supply.

Pointing at already-registered plans: ReferenceLaunchPlan

Sometimes you need to compile against a launch plan that already exists on your Flyte installation — say, a sub-launchplan owned by another team. ReferenceLaunchPlan (a ReferenceEntity and a LaunchPlan) is a pointer that does not contact Flyte Admin; you must supply the expected interface yourself:

ReferenceLaunchPlan(
    project="flytesnacks",
    domain="development",
    name="my.other.launchplan",
    version="abc123",
    inputs={"a": int},
    outputs={},
)

The friendlier path is the reference_launch_plan() decorator, which derives the interface from an annotated function signature via transform_function_to_interface():

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="my.other.launchplan",
    version="abc123",
)
def other_lp(a: int) -> int:
    ...

Two constraints: if the declared interface doesn't match what's actually registered, you'll get an error at registration/compilation time; and reference entities cannot execute locally — they exist for compilation against remote entities.

Quick reference

Goal	Code
Default plan	LaunchPlan.get_or_create(workflow=wf)
Named plan with defaults	LaunchPlan.get_or_create(name=..., workflow=wf, default_inputs={...})
Unchangeable inputs	fixed_inputs={"x": 1} (removed from the parameter map)
Cron schedule (old arg)	schedule=CronSchedule(schedule="@daily")
Schedule via trigger (new syntax)	trigger=OnSchedule(FixedRate(duration=timedelta(minutes=10)))
Kickoff time as input	kickoff_time_input_arg="kickoff_time"
Remote plan reference	reference_launch_plan(project=..., domain=..., name=..., version=...)