Here are the quickest ways to create a cron schedule in Prefect 3.x (code and CLI). Default timezone is UTC unless you set one. In code (using Cron)

from prefect import flow
from prefect.schedules import Cron

@flow
def my_flow():
    print("Hello!")

# Every weekday at 9 AM New York time
my_flow.serve(
    name="weekday-9am",
    schedule=Cron("0 9 * * 1-5", timezone="America/New_York")
)

In code (shorthand on serve)

my_flow.serve(
    name="weekday-9am",
    cron="0 9 * * 1-5",
    timezone="America/New_York"
)

CLI when creating a deployment

prefect deploy flows.py:my_flow \
  --name weekday-9am \
  --pool my-work-pool \
  --cron "0 9 * * 1-5" \
  --timezone "America/New_York"

CLI to add a schedule to an existing deployment

# Use the full "flow-name/deployment-name" if needed
prefect deployment schedule create "my_flow/weekday-9am" \
  --cron "0 9 * * 1-5" \
  --timezone "America/New_York"

Cron quick reference - Format: minute hour day_of_month month day_of_week - Examples: - Every day at midnight:

0 0 * * *

- Every 15 minutes:

*/15 * * * *

- Weekdays at 9 AM:

0 9 * * 1-5

Docs and source - How to create schedules - CLI: deployment commands - prefect.schedules (Cron, Interval, RRule) If you share the exact cadence and timezone you want, I can give you the precise cron and command/snippet.

Yes to both in Prefect 3.x. Temporarily stop (pause) a schedule - CLI

# Pause all schedules on a deployment
  prefect deployment schedule pause "flow-name/deployment-name" --all

  # Pause a specific schedule by ID
  prefect deployment schedule pause "flow-name/deployment-name" <schedule_id>

  # Resume later
  prefect deployment schedule resume "flow-name/deployment-name" --all

Tip: List schedules and IDs first

prefect deployment schedule ls "flow-name/deployment-name"

- Python SDK (toggle the active flag)

from prefect import get_client
  from uuid import UUID

  deployment_id = UUID("your-deployment-uuid")
  schedule_id = UUID("your-schedule-uuid")

  async def pause_schedule():
      async with get_client() as client:
          await client.update_deployment_schedule(
              deployment_id=deployment_id,
              schedule_id=schedule_id,
              active=False,  # pause
          )

  async def resume_schedule():
      async with get_client() as client:
          await client.update_deployment_schedule(
              deployment_id=deployment_id,
              schedule_id=schedule_id,
              active=True,  # resume
          )

You can discover schedule IDs via:

async with get_client() as client:
      schedules = await client.read_deployment_schedules(deployment_id)
      for s in schedules:
          print(s.id, s.slug, s.active)

Specify start and end times with the Python SDK - Best option: use an RRule schedule with DTSTART (start) and UNTIL (end). Prefect accepts an RFC 5545 string for RRule.

from prefect import flow
  from prefect.schedules import RRule

  @flow
  def my_flow():
      ...

  my_flow.serve(
      name="windowed-schedule",
      schedule=RRule(
          """DTSTART:20240401T090000Z
RRULE:FREQ=DAILY;UNTIL=20240430T090000Z"""
      ),
      timezone="UTC",
  )

Notes: - Put UNTIL inside the RRULE line (as shown). - Use an IANA timezone where needed (e.g., "America/New_York"). - Cron/Interval notes: - Cron doesn’t include start/end fields; use RRule if you need a fixed window. - Interval supports an anchor date (phase), not an end. You can still programmatically pause/resume at your desired end. Start inactive, enable later - You can create any schedule initially paused and flip it on later:

from prefect.schedules import Cron

  my_flow.serve(
      name="paused-at-start",
      schedule=Cron("0 9 * * 1-5", timezone="America/New_York", active=False),
  )
  # Later: set active=True via CLI or client.update_deployment_schedule(...)

Docs and references - How to create schedules - CLI: deployment schedule commands - prefect.schedules (Cron, Interval, RRule) If you share the exact time window you need (start/end and timezone), I can provide the exact RRule string or a ready-to-run snippet.

Great question. In Prefect 3.x you can programmatically create a deployment with the Python client’s create_deployment method. Here’s a minimal end‑to‑end example using the async client, including a cron schedule. Async example (recommended)

import asyncio
from datetime import timedelta
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import DeploymentScheduleCreate
from prefect.client.schemas.schedules import CronSchedule  # or IntervalSchedule, RRuleSchedule

# 1) Define your flow
@flow(name="my-flow")
def my_flow(name: str = "world"):
    print(f"Hello {name}")

async def main():
    async with get_client() as client:
        # 2) Register the flow (creates/updates the flow in Prefect)
        flow_id = await client.create_flow(my_flow)

        # 3) Define a schedule (every weekday at 9 AM New York time)
        schedule = DeploymentScheduleCreate(
            schedule=CronSchedule(cron="0 9 * * 1-5", timezone="America/New_York"),
            active=True,
        )

        # 4) Create the deployment
        # - entrypoint must be importable on your worker(s), e.g. "flows.py:my_flow"
        # - path is the working directory root for that entrypoint
        deployment_id = await client.create_deployment(
            flow_id=flow_id,
            name="weekday-9am",
            entrypoint="flows.py:my_flow",
            path=".",  # directory where flows.py lives on the worker
            work_pool_name="default",  # must exist
            parameters={"name": "Prefect"},
            tags=["example"],
            schedules=[schedule],
            description="Runs my_flow on weekdays at 9 AM ET",
            paused=False,
            enforce_parameter_schema=True,
        )
        print(f"Created deployment: {deployment_id}")

        # 5) (Optional) Trigger a manual run
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id=deployment_id,
            parameters={"name": "Manual run"},
        )
        print(f"Created flow run: {flow_run.id}")

if __name__ == "__main__":
    asyncio.run(main())

Sync version

from prefect import flow
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.actions import DeploymentScheduleCreate
from prefect.client.schemas.schedules import CronSchedule

@flow
def my_flow():
    print("Hello")

with SyncPrefectClient() as client:
    flow_id = client.create_flow(my_flow)
    schedule = DeploymentScheduleCreate(
        schedule=CronSchedule(cron="0 0 * * *", timezone="UTC"),
        active=True,
    )
    deployment_id = client.create_deployment(
        flow_id=flow_id,
        name="midnight-utc",
        entrypoint="flows.py:my_flow",
        path=".",
        work_pool_name="default",
        schedules=[schedule],
    )
    print(deployment_id)

Notes - entrypoint must be something your workers can import/execute (for example "flows.py:my_flow"), and path should point to that code’s root directory on the worker. - For remote code (Git, etc.), the high-level API is usually simpler: use flow.from_source(...).deploy(...). If you want to stick with client.create_deployment, you’ll need to ensure your workers have access to the code at runtime. - Schedules can be CronSchedule, IntervalSchedule, or RRuleSchedule (from prefect.client.schemas.schedules) wrapped in DeploymentScheduleCreate. Related docs - How to deploy flows - How to create schedules If you share your code layout (where flows.py lives and how workers access it), I can tailor the entrypoint/path values or suggest a from_source example.

You’re on the right track. A few best-practice tweaks will make this more robust and easier to maintain. Key suggestions - Include entrypoint and path: Unless your workers have your flow importable on PYTHONPATH already, set

entrypoint="...:flow_func"

and

path="."

(or your project root) on the deployment. This avoids “can’t load flow code” issues at run time. - Validate inputs early: - Validate cron strings and timezones up front to fail fast. - Consider exposing

day_or

behavior for cron if you need precise control (cron day vs. weekday logic). - Give the schedule a slug: Use

slug=

in

DeploymentScheduleCreate

so you can find/update/pause this schedule later without guessing IDs. - Handle idempotency: Before creating, check if a deployment with the same flow/name exists and update schedules instead of creating duplicates. - Control schedule activation: If you sometimes want to create it paused, wire

active

to your input and toggle with the client later. - Add work_queue_name (optional): If you use non-default queues inside the pool, set it here. - Error handling and logging: Catch known exceptions and log helpful messages. An improved version ``` import asyncio import uuid from typing import Optional from prefect.client.orchestration import get_client from prefect.client.schemas.actions import DeploymentScheduleCreate from prefect.client.schemas.schedules import CronSchedule from prefect.exceptions import ObjectAlreadyExists, ObjectNotFound # Optional: quick validators (fail fast) from prefect.client.schedules import is_valid_timezone async def create_schedule( schedule_create, # your DTO with cron_expression, timezone, workflow_definition_code, etc. created_by: str, ) -> str: """ Create or update a deployment with a cron schedule for an existing flow. Returns the deployment UUID as a string. """ # 0) Validate timezone (fail fast) if schedule_create.timezone and not is_valid_timezone(schedule_create.timezone): raise ValueError(f"Invalid timezone: {schedule_create.timezone} (use IANA tz, e.g. 'America/New_York')") deployment_name = f"ks-schedule-{schedule_create.workflow_definition_code}" async with get_client() as client: # 1) Ensure the flow exists (you’re providing a flow_id from settings) flow_id = uuid.UUID(settings.prefect_dag_flow_id) # 2) Try to reuse existing deployment (idempotency) existing_deployment_id: Optional[uuid.UUID] = None try: existing = await client.read_deployment_by_name(f"{schedule_create.flow_name}/{deployment_name}") # if you know the flow name existing_deployment_id = existing.id except Exception: # If you don’t know flow name at runtime, you can skip this and let create_deployment handle it pass # 3) Build the schedule with a slug for easy lifecycle management later cron_sched = DeploymentScheduleCreate( schedule=CronSchedule( cron=schedule_create.cron_expression, timezone=schedule_create.timezone or "UTC", # day_or=True, # uncomment/control if you need specific day vs weekday behavior ), active=True, # or wire this from schedule_create if you want to create it paused slug=f"cron-{schedule_create.workflow_definition_code}", ) # 4) Create or update the deployment try: deployment_id = await client.create_deployment( flow_id=flow_id, name=deployment_name, work_pool_name=settings.prefect_dag_worker_pool, # Optional but recommended: make code import explicit for workers # entrypoint="app/flows.py:my_flow", # path=".", schedules=[cron_sched], tags=[str(schedule_create.workflow_definition_code)], description=f"Created by {created_by}",

enforce_parameter_schema=True, paused=False, ) except ObjectAlreadyExists: # If you got here, a deployment with the same flow+name likely exists. # Update its schedule instead of creating a new one. if not existing_deployment_id: # Fallback: read by name now, if possible existing = await client.read_deployment_by_name(f"{schedule_create.flow_name}/{deployment_name}") existing_deployment_id = existing.id # If you want to replace the schedule, first find any existing schedules and pause/delete/update as needed schedules = await client.read_deployment_schedules(existing_deployment_id) # Option A: Update the first schedule (or the one with your slug) target = next((s for s in schedules if s.slug == f"cron-{schedule_create.workflow_definition_code}"), None) if target: await client.update_deployment_schedule( deployment_id=existing_deployment_id, schedule_id=target.id, schedule=cron_sched.schedule, # updates cron/timezone active=cron_sched.active, slug=cron_sched.slug, ) deployment_id = existing_deployment_id else: # Option B: Add a new schedule to the existing deployment await client.create_deployment_schedule( deployment_id=existing_deployment_id, schedule=cron_sched.schedule, active=cron_sched.active, slug=cron_sched.slug, ) deployment_id = existing_deployment_id except ObjectNotFound as e: raise RuntimeError(f"Flow not found: {flow_id}") from e return str(deployment_id) ``` A few optional improvements - Start paused: Set

active=False

on the schedule at creation, then flip it on later (CLI or client API). - Use RRule for windows: If you need a schedule with a fixed start/end window, use

RRuleSchedule

with `DTSTART`/`UNTIL`. - Work queue targeting: If your pool has multiple queues, pass

work_queue_name=

to aim runs precisely. - Observability: Add a

slug

per schedule to make pause/resume/update operations deterministic. Helpful docs - Create schedules (Cron/Interval/RRule) - Deployments how-to - CLI: deployment/schedule commands - Client schedule schemas (Cron/Interval/RRule) If you share how your workers access code (local path vs. Git), I can tailor the entrypoint/path or suggest a from_source flow for a fully-managed code delivery.

Short answer: Yes, but only with RRule. Prefect does not have a generic “active time range” switch on Cron/Interval schedules; instead: - Use RRule with DTSTART and UNTIL to bound a schedule between specific start/end datetimes. - Or toggle a schedule’s active flag (pause/resume) programmatically or via CLI at your desired times. Bounded schedule window with RRule

from prefect import flow
from prefect.schedules import RRule

@flow
def my_flow():
    ...

# Active only between 2024-06-01 09:00 UTC and 2024-06-30 17:00 UTC, every weekday at 9:00
my_flow.serve(
    name="bounded-window",
    schedule=RRule("""DTSTART:20240601T090000Z
RRULE:FREQ=DAILY;BYDAY=MO,TU,WE,TH,FR;UNTIL=20240630T170000Z"""),
    timezone="UTC",
)

Notes: - Put UNTIL inside the RRULE line (as shown), not as a separate line. - You can also add EXDATE/RDATE lines to exclude/include specific dates. Daily “hours of operation” with cron If you mean “only run during certain hours each day,” encode that in cron rather than a start/end window:

# Every 15 minutes between 09:00–17:00, Mon–Fri, New York time
my_flow.serve(
    name="business-hours",
    cron="*/15 9-17 * * 1-5",
    timezone="America/New_York",
)

Toggle active state (pause/resume) at specific times - CLI

prefect deployment schedule pause "flow-name/deployment-name" --all
  prefect deployment schedule resume "flow-name/deployment-name" --all

- Python

from prefect import get_client
  from uuid import UUID

  async def set_active(deployment_id: UUID, schedule_id: UUID, active: bool):
      async with get_client() as client:
          await client.update_deployment_schedule(
              deployment_id=deployment_id,
              schedule_id=schedule_id,
              active=active,
          )

You can run a small script or use your own scheduler to call this at desired times. Docs - How to create schedules - prefect.schedules (Cron, Interval, RRule) If you share the exact window you need (start, end, cadence, timezone), I can craft the precise RRule string or a ready-to-run snippet.