Launch plans, schedules, and fixed inputs

## Launch Plans

Every workflow in Flyte has an execution interface — the inputs it accepts, the outputs it produces, and the constraints around when and how it runs. That interface is the launch plan. When you register a workflow, Flyte creates a default launch plan automatically, but you can also create custom launch plans with specific default values, fixed inputs, schedules, and notifications.

### The Default Launch Plan

You get the default launch plan for any workflow by calling `LaunchPlan.get_or_create()` with just the workflow:

```python
from flytekit import workflow, LaunchPlan

@workflow
def my_wf(a: int, c: str) -> str:
    return f"\{a\} \{c\}"

default_lp = LaunchPlan.get_or_create(workflow=my_wf)

The default launch plan takes its name from the workflow (my_wf in this case) and inherits whatever default values are declared in the workflow signature. It has no schedule, no fixed inputs, and no notifications — those must be configured on a named launch plan.

One constraint: you cannot add default inputs, fixed inputs, a schedule, or any other configuration to the default launch plan. The get_or_create() method enforces this at line 231–247 of flytekit/core/launch_plan.py:

if name is None and (
    default_inputs is not None
    or fixed_inputs is not None
    or schedule is not None
):
    raise ValueError(
        "Only named launchplans can be created that have other properties. "
        "Drop the name if you want to create a default launchplan."
    )

If you need any of those features, you must create a named launch plan instead.

Named Launch Plans with Default and Fixed Inputs

Pass a name argument to create a launch plan with additional configuration. The name must be unique across your Flyte deployment (project, domain, version, and name together form the primary key).

named_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="my_scheduled_lp",
    default_inputs=\{"a": 10\},
    fixed_inputs=\{"c": "fixed_value"\},
)

default_inputs are pre-populated values that callers can override at execution time. In the example above, a defaults to 10 but you can still pass a different value when launching. The precedence order is: explicit call arguments override defaults, which override any signature-level defaults.

fixed_inputs are locked at launch plan creation time and cannot be changed when launching. In the example above, c is always "fixed_value" — if you try to pass a different value, the launch plan rejects it. This distinction matters because fixed inputs are stripped from the parameters map at construction time (see line 338 of flytekit/core/launch_plan.py):

# Ensure fixed inputs are not in parameter map
param_map = \{k: v for k, v in parameters.parameters.items() if k not in fixed_inputs.literals\}

The _saved_inputs property on LaunchPlan (line 419) merges both default and fixed inputs together, so calling the launch plan locally automatically supplies them:

result = named_lp()  # Uses default_inputs['a']=10 and fixed_inputs['c']='fixed_value'
result = named_lp(a=42)  # Override the default, still uses fixed 'c'

One gotcha: if you pass the same argument in both default_inputs and fixed_inputs, the fixed value wins and the default is removed during construction.

Launch Plan Caching

LaunchPlan.get_or_create() caches its results. The CACHE is a class-level dictionary keyed by name (or by workflow.name for the default launch plan). Calling get_or_create() twice with the same name returns the cached object:

lp1 = LaunchPlan.get_or_create(workflow=my_wf, name="my_lp")
lp2 = LaunchPlan.get_or_create(workflow=my_wf, name="my_lp")
assert lp1 is lp2  # Same object

If you call get_or_create() with the same name but different configuration values, Flyte raises an AssertionError comparing the old and new values for each parameter (lines 271–287). This prevents accidentally creating two launch plans with the same name but different behavior.

Scheduling with CronSchedule and FixedRate

Attach a schedule to a launch plan to make it run automatically. flytekit provides two schedule types in flytekit/core/schedule.py.

CronSchedule

Use CronSchedule for cron-based schedules. Pass a cron expression using the standard 5-field format:

from flytekit.core.schedule import CronSchedule

hourly_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="hourly_training",
    schedule=CronSchedule(schedule="0 * * * *"),
)

CronSchedule also accepts human-readable aliases defined in _VALID_CRON_ALIASES (lines 37–55 of flytekit/core/schedule.py):

# All equivalent aliases
CronSchedule(schedule="daily")
CronSchedule(schedule="@daily")
CronSchedule(schedule="days")

The aliases cover hourly, daily, weekly, monthly, and yearly with their plural forms and @ prefixes.

Note: The cron_expression parameter is deprecated. Always use schedule instead.

You can add an offset to shift the schedule relative to UTC using ISO 8601 duration format:

# Run daily at 5am UTC instead of midnight
morning_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="morning_training",
    schedule=CronSchedule(schedule="daily", offset="PT5H"),
)

FixedRate

Use FixedRate for interval-based scheduling. Pass a timedelta duration:

from datetime import timedelta
from flytekit.core.schedule import FixedRate

every_10_min_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="frequent_sync",
    schedule=FixedRate(duration=timedelta(minutes=10)),
)

FixedRate._translate_duration() (lines 178–206 of flytekit/core/schedule.py) automatically converts the duration to the appropriate unit:

• If the duration is a multiple of days, it uses FixedRateUnit.DAY
• If it's a multiple of hours, it uses FixedRateUnit.HOUR
• Otherwise, it uses FixedRateUnit.MINUTE

Minimum granularity is 1 minute. Sub-minute durations raise an AssertionError at line 188–191:

if duration.microseconds != 0 or duration.seconds % _SECONDS_TO_MINUTES != 0:
    raise AssertionError(
        f"Granularity of less than a minute is not supported for FixedRate schedules."
    )

Passing Kickoff Time to the Workflow

Both CronSchedule and FixedRate accept a kickoff_time_input_arg parameter that injects the scheduled kickoff time as a workflow input:

import datetime
from flytekit import workflow
from flytekit.core.schedule import CronSchedule

@workflow
def daily_report(kickoff_time: datetime.datetime) -> str:
    return f"Report generated at \{kickoff_time\}"

daily_lp = LaunchPlan.get_or_create(
    workflow=daily_report,
    name="daily_email",
    schedule=CronSchedule(
        schedule="0 9 * * *",
        kickoff_time_input_arg="kickoff_time",
    ),
)

The kickoff_time_input_arg docstring (lines 73–83 of flytekit/core/schedule.py) notes that Flyte does not have an atomic clock, so there may be a few seconds of drift between the intended kickoff time and the actual one.

Notifications for Workflow Completion

Attach notifications to a launch plan so Flyte emails you when the workflow reaches terminal phases. The notification classes in flytekit/core/notification.py are Email, PagerDuty, and Slack — all three ultimately send emails through their respective integrations.

from flytekit.core.notification import Email, PagerDuty, Slack
from flytekit.models.core.execution import WorkflowExecutionPhase

email_on_complete = Email(
    phases=[WorkflowExecutionPhase.SUCCEEDED, WorkflowExecutionPhase.FAILED],
    recipients_email=["team@company.com"],
)

pagerduty_on_failure = PagerDuty(
    phases=[WorkflowExecutionPhase.FAILED],
    recipients_email=["oncall@company.com"],
)

notified_lp = LaunchPlan.get_or_create(
    workflow=my_wf,
    name="monitored_training",
    notifications=[email_on_complete, pagerduty_on_failure],
)

Notifications can only fire for terminal phases. The Notification.VALID_PHASES at lines 19–24 of flytekit/core/notification.py defines the allowed set:

VALID_PHASES = \{
    WorkflowExecutionPhase.ABORTED,
    WorkflowExecutionPhase.FAILED,
    WorkflowExecutionPhase.SUCCEEDED,
    WorkflowExecutionPhase.TIMED_OUT,
\}

Any other phase raises an AssertionError in _validate_phases() (line 47–48).

Reference Launch Plans

A ReferenceLaunchPlan points to a launch plan that already exists on your Flyte deployment — it does not initiate network calls to Admin at creation time. Use it when you want to reference a launch plan from another project or domain.

from flytekit.core.launch_plan import reference_launch_plan

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="core.basic.lp.my_lp",
    version="abc123",
)
def my_lp_ref(a: str, b: int) -> str:
    """
    This function's signature defines the expected interface.
    """
    ...

The decorated function's type annotations define the expected inputs and outputs. Interface validation happens at registration time, not at creation time. If the interface doesn't match the actual launch plan, you'll get an error during compilation or registration.

Using Launch Plans in Dynamic Tasks

When launching launch plans from within @dynamic tasks, you must use node_dependency_hints to ensure the launch plan is registered on flyteadmin before execution. Dynamic tasks cannot determine dependencies statically, so this hint tells Flyte to register the launch plan during the registration phase.

from flytekit import workflow, dynamic, LaunchPlan

@workflow
def workflow0():
    pass

launchplan0 = LaunchPlan.get_or_create(workflow0)

@dynamic(node_dependency_hints=[launchplan0])
def launch_dynamically():
    # To run a sub-launchplan it must have previously been registered on flyteadmin.
    return [launchplan0] * 10

This pattern is documented in flytekit/core/task.py lines 319–335.

Using Launch Plans with ArrayNode (Map Tasks)

When you use a LaunchPlan as the target of an ArrayNode (for map tasks), the fixed inputs are automatically excluded from the mapped inputs. The ArrayNode constructor at flytekit/core/array_node.py lines 84–85 handles this:

if isinstance(target, (LaunchPlan, FlyteLaunchPlan)) and not isinstance(target, ReferenceLaunchPlan):
    self._excluded_inputs = set(target.fixed_inputs.literals)

This means fixed inputs don't get mapped across task instances — they're applied once per launch.

Runtime Options with the Options Dataclass

The Options dataclass in flytekit/core/options.py configures runtime behavior that applies during both registration and execution. You can override labels, annotations, security context, output storage location, parallelism, notifications, and cache behavior:

from flytekit.core.options import Options
from flytekit.models import common, security

options = Options(
    labels=common.Labels(\{"env": "production"\}),
    annotations=common.Annotations(\{"owner": "data-team"\}),
    security_context=security.SecurityContext(
        run_as=security.Identity(k8s_service_account="ml-pod-sa")
    ),
    raw_output_data_config=common.RawOutputDataConfig(
        output_location_prefix="s3://my-bucket/outputs"
    ),
    max_parallelism=10,
    overwrite_cache=True,
)

The Options.default_from() factory method provides a shorthand for common configurations:

options = Options.default_from(
    k8s_service_account="ml-pod-sa",
    raw_data_prefix="s3://my-bucket/outputs",
)

Eager Workflows with LaunchPlan as RunnableEntity

In eager workflow scenarios, LaunchPlan is one of the valid RunnableEntity types that the Controller class can execute. The RunnableEntity type alias in flytekit/core/worker_queue.py line 33 includes LaunchPlan:

RunnableEntity = typing.Union[WorkflowBase, LaunchPlan, PythonTask, ReferenceEntity, RemoteEntity]

The Controller.add() method accepts a RunnableEntity and its inputs, then submits it to Admin for execution.

Launch Plan Properties Reference

Property	Type	Description
name	str	Unique identifier (combined with project/domain/version)
parameters	ParameterMap	Workflow inputs with default values (excludes fixed inputs)
fixed_inputs	LiteralMap	Locked inputs that cannot be overridden at launch time
schedule	Schedule	CronSchedule or FixedRate for automated execution
notifications	List[Notification]	Email/PagerDuty/Slack alerts on terminal phases
labels	Labels	Custom labels for execution resources
annotations	Annotations	Custom annotations for execution resources
raw_output_data_config	RawOutputDataConfig	Offloaded data location (S3, GCS, etc.)
max_parallelism	int	Maximum parallel task nodes across the workflow
security_context	SecurityContext	IAM role and Kubernetes service account for execution
auto_activate	bool	Whether to auto-activate on registration (default False)
overwrite_cache	bool	Whether to ignore cache and re-execute (default None)

Deprecation Notes

• auth_role parameter: Deprecated in favor of security_context. You cannot use both together — doing so raises a ValueError at line 145–146 of flytekit/core/launch_plan.py.
• cron_expression parameter in CronSchedule: Deprecated. Use schedule instead, which supports both aliases and cron expressions.

Key Gotchas

1. Default launch plans cannot have additional properties — you must use a named launch plan to set default_inputs, fixed_inputs, schedule, or notifications.
2. Launch plan names must be unique — creating two with the same name but different properties raises AssertionError.
3. Fixed inputs are excluded from the parameters map — you cannot override them at execution time, and they are excluded from ArrayNode mappings.
4. FixedRate minimum granularity is 1 minute — sub-minute durations raise AssertionError.
5. ReferenceLaunchPlan does not make network calls — interface validation happens at registration time, not at creation time.
6. Notifications only fire for terminal phases — ABORTED, FAILED, SUCCEEDED, and TIMED_OUT.</parameter> </invoke>