Short answer: you can’t. In Prefect 3.x,

prefect.yaml

only supports built‑in time schedules (cron, interval, rrule). There’s no way to attach a custom Python function in YAML to compute “next run.” What you can do instead: - Encode complex timing with RRule in YAML - Example:

deployments:
      - name: reports
        entrypoint: flows/report.py:generate
        schedule:
          rrule: "FREQ=MONTHLY;BYDAY=1FR"   # first Friday each month
          timezone: "America/New_York"

- Compute the schedule in Python, then deploy - This is the usual pattern if you truly need code to decide the cadence:

from datetime import timedelta
    from prefect import flow
    from prefect.schedules import Cron, Interval, RRule

    def build_schedule():
        # Your custom logic here (DB lookup, config, holidays, etc.)
        return Cron("0 9 * * 1-5", timezone="America/New_York")  # example

    @flow
    def my_flow():
        ...

    my_flow.deploy(
        name="dynamic-schedule",
        work_pool_name="your-work-pool",
        schedule=build_schedule(),
    )

- If your code lives in a repo, use:

from prefect import flow
    from prefect.schedules import Cron

    schedule = Cron("0 * * * *", timezone="UTC")
    flow.from_source(
        source="<https://github.com/your-org/your-repo>",
        entrypoint="flows.py:my_flow",
    ).deploy(
        name="repo-deployment",
        work_pool_name="your-work-pool",
        schedule=schedule,
    )

- Go event-driven instead of time-based - If your “custom function” really decides based on external conditions, consider Automations/Triggers to kick off runs on events or states rather than trying to encode it in a cron expression. Docs for reference: - Schedules concept: Concepts: Schedules - YAML deployments (schedule fields): How to: prefect.yaml - Python schedule helpers (Cron, Interval, RRule): API: prefect.schedules If you share the custom logic you want (e.g., business days, holiday exclusions, “every Nth weekday”, etc.), I can help translate it to an RRule or show a small Python helper that builds the schedule before deploying.

Yes. In Prefect 3.x you can check it from the flow run context. -

flow_run.auto_scheduled

is True when the run was created by a deployment schedule, False for manual/adhoc. -

flow_run.created_by

gives more detail (e.g.,

type == "SCHEDULE"

for scheduled runs,

"AUTOMATION"

for automations), plus an id and display name. - As a secondary signal, scheduled runs include the

"auto-scheduled"

tag. Example:

from prefect import flow, get_run_logger
from prefect.context import get_run_context

@flow
def my_flow():
    logger = get_run_logger()
    flow_run = get_run_context().flow_run

    if flow_run.auto_scheduled:
        <http://logger.info|logger.info>("This run was created by a schedule.")
        if flow_run.created_by:
            <http://logger.info|logger.info>(f"type={flow_run.created_by.type} "
                        f"name={flow_run.created_by.display_value} "
                        f"id={flow_run.created_by.id}")
    else:
        <http://logger.info|logger.info>("This run was started manually or by something else.")
        if flow_run.created_by:
            <http://logger.info|logger.info>(f"type={flow_run.created_by.type}")

    # Optional secondary check via tags
    if "auto-scheduled" in (flow_run.tags or []):
        <http://logger.info|logger.info>("Confirmed scheduled via tag.")

Docs: - Flow run context: API: prefect.context - FlowRun schema (fields like auto_scheduled, created_by): API: client schemas – objects

Short answer: not directly. In a subflow,

flow_run.auto_scheduled

will be False because subflows are created by the parent flow engine, not by the scheduler. So the “check `auto_scheduled`” trick only tells you the truth on the top-level flow run. Two good patterns to use inside a subflow: - Look up the parent’s flow run and inspect it (authoritative) - Async example:

from prefect import flow
    from prefect.context import get_run_context
    from prefect.client.orchestration import get_client

    @flow
    async def child():
        ctx = get_run_context()
        parent_task_run_id = ctx.flow_run.parent_task_run_id

        async with get_client() as client:
            parent_task = await client.read_task_run(parent_task_run_id)
            parent_flow = await client.read_flow_run(parent_task.flow_run_id)

        is_scheduled = bool(parent_flow.auto_scheduled)
        trigger_type = getattr(parent_flow.created_by, "type", None)
        return is_scheduled, trigger_type

- If you prefer sync, use

SyncPrefectClient

similarly. - Pass the trigger down from the parent (simplest)

from prefect import flow
  from prefect.context import get_run_context

  @flow
  def parent():
    top = get_run_context().flow_run
    run_mode = "scheduled" if top.auto_scheduled else "manual"
    child(run_mode)

  @flow
  def child(run_mode: str):
    print(f"child run_mode={run_mode}")

Notes: - Subflows may inherit tags from the parent; you might see

"auto-scheduled"

in

ctx.flow_run.tags

if the parent was scheduled, but treat the parent’s

auto_scheduled

flag as the source of truth. - You can also inspect the parent’s

created_by.type

(

"SCHEDULE"

for scheduler-triggered runs). Docs: - Runtime context: API: prefect.context - Prefect client (orchestration): API: orchestration client

• there's

on_completion

hooks where you can call run_deployment with whatever schedule time you want • there's automations so you can trigger a run when a set or combination of other things happen