Yes — from the Python SDK you can list automations and delete them one by one. There does not appear to be a Python SDK bulk-delete method like

delete_automations()

. In Prefect 3.x, the relevant client methods are: -

read_automations()

→ list all automations -

delete_automation(automation_id)

→ delete one automation - optionally

read_automations_by_name(name)

or

find_automation(...)

Async example: delete all automations

python
import asyncio
from prefect.client.orchestration import get_client

async def delete_all_automations():
    async with get_client() as client:
        automations = await client.read_automations()

        for automation in automations:
            await client.delete_automation(automation.id)
            print(f"Deleted: {automation.name} ({automation.id})")

asyncio.run(delete_all_automations())

If you want it a little faster, you can delete in parallel:

python
import asyncio
from prefect.client.orchestration import get_client

async def delete_all_automations():
    async with get_client() as client:
        automations = await client.read_automations()

        semaphore = asyncio.Semaphore(10)

        async def delete_one(automation):
            async with semaphore:
                await client.delete_automation(automation.id)
                print(f"Deleted: {automation.name} ({automation.id})")

        await asyncio.gather(*(delete_one(a) for a in automations))

asyncio.run(delete_all_automations())

Sync example

python
from prefect.client.orchestration import get_client

with get_client(sync_client=True) as client:
    automations = client.read_automations()

    for automation in automations:
        client.delete_automation(automation.id)
        print(f"Deleted: {automation.name} ({automation.id})")

If you only want to delete automations with a specific name:

python
import asyncio
from prefect.client.orchestration import get_client

async def delete_named_automations(name: str):
    async with get_client() as client:
        automations = await client.read_automations_by_name(name)
        for automation in automations:
            await client.delete_automation(automation.id)

asyncio.run(delete_named_automations("my-automation"))

Also, if CLI is acceptable, Prefect does have an

--all

delete flow there. I can verify the exact command syntax for your version if you want.

Absolutely — here is the synchronous example extended to: 1. list existing automations 2. delete them one by one 3. create a new automation

python
from datetime import timedelta

from prefect.client.orchestration import get_client
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.actions import DoNothing

# 1) Delete all existing automations
with get_client(sync_client=True) as client:
    automations = client.read_automations()

    for automation in automations:
        client.delete_automation(automation.id)
        print(f"Deleted: {automation.name} ({automation.id})")

# 2) Create a new automation
new_automation = Automation(
    name="example-automation",
    description="Example automation created from synchronous Python code",
    enabled=True,
    trigger=EventTrigger(
        expect={"prefect.flow-run.Failed"},
        posture=Posture.Reactive,
        threshold=1,
        within=timedelta(minutes=5),
    ),
    actions=[DoNothing()],
).create()

print(f"Created automation: {new_automation.name} ({new_automation.id})")

A few notes: -

client.read_automations()

and

client.delete_automation(...)

are the sync client methods. - For creation, the easiest SDK pattern is using

Automation(...).create()

. -

DoNothing()

is a simple placeholder action, useful if you just want to verify creation works. - You can replace

DoNothing()

with another action like

RunDeployment

,

CancelFlowRun

, or

SendNotification

depending on what you want the automation to do. For example, if you want to create a flow-failure automation with a more specific filter:

python
from datetime import timedelta

from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.actions import DoNothing

automation = Automation(
    name="failed-flow-alert",
    description="Trigger when a specific flow fails",
    enabled=True,
    trigger=EventTrigger(
        expect={"prefect.flow-run.Failed"},
        posture=Posture.Reactive,
        threshold=1,
        within=timedelta(minutes=10),
        match={
            "prefect.resource.name": "my-flow"
        },
    ),
    actions=[DoNothing()],
).create()

print(f"Created automation: {automation.id}")

If you want, I can also give you a version that: - deletes all automations and recreates a specific set, or - creates a real automation action like

RunDeployment

or

SendNotification

.