Assuming Prefect 3.x: there isn’t a built-in CLI command to bulk-cancel runs for a specific deployment. The easiest reliable way is to use the Python client with a filter by deployment and state, either via the bulk REST endpoint or by iterating. Option A (fastest): bulk-cancel via REST endpoint with filters - Uses the server’s bulk_set_state endpoint to set all matching runs to Cancelling in one request (up to 500 at a time). - Links: - Bulk set flow run state (REST) - FlowRunFilter schema Example:

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter, FlowRunFilterDeploymentId, FlowRunFilterState, FlowRunFilterStateName,
)
from prefect.states import Cancelling, to_state_create

async def cancel_late_runs_for_deployment(flow_name: str, deployment_name: str, limit: int = 500):
    async with get_client() as client:
        # Look up the deployment id
        dep = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")

        # Build a filter: deployment_id + state name "Late"
        filt = FlowRunFilter(
            deployment_id=FlowRunFilterDeploymentId(any_=[dep.id]),
            state=FlowRunFilterState(name=FlowRunFilterStateName(any_=["Late"])),
        )

        payload = {
            "flow_runs": filt.model_dump(mode="json", exclude_unset=True),
            "state": to_state_create(Cancelling()).model_dump(mode="json", serialize_as_any=True),
            "force": False,
            "limit": limit,  # max 500 per call
        }

        # Bulk set to Cancelling
        resp = await client.request("POST", "/flow_runs/bulk_set_state", json=payload)
        print(resp.json())  # contains per-run orchestration results

# Example call
asyncio.run(cancel_late_runs_for_deployment("my_flow", "my_deployment"))

Tip: If you have more than 500 matches, call it in a loop until no more runs are returned in the response. Option B: iterate and cancel each run - Uses regular client methods if you’d prefer not to hit the raw REST path. - Links: - Read flow runs (filters) - Set flow run state Example:

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter, FlowRunFilterDeploymentId, FlowRunFilterState, FlowRunFilterStateName,
)
from prefect.states import Cancelling

async def cancel_late_runs_for_deployment_iter(flow_name: str, deployment_name: str, page_size: int = 500):
    async with get_client() as client:
        dep = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")
        filt = FlowRunFilter(
            deployment_id=FlowRunFilterDeploymentId(any_=[dep.id]),
            state=FlowRunFilterState(name=FlowRunFilterStateName(any_=["Late"])),
        )
        runs = await client.read_flow_runs(flow_run_filter=filt, limit=page_size)
        for fr in runs:
            await client.set_flow_run_state(flow_run_id=fr.id, state=Cancelling())

asyncio.run(cancel_late_runs_for_deployment_iter("my_flow", "my_deployment"))

Notes - CLI: there’s no bulk cancel by deployment, and

prefect flow-run ls

cannot filter by deployment; you’d have to script it yourself. CLI refs: prefect flow-run - States you can filter by include `LATE`; cancel state to set is

Cancelling()

. If you’d like, tell me your flow/deployment names and whether you prefer bulk REST or iterative, and I can tailor a ready-to-run snippet for you.

Thanks for the error details — that helps. You’re right: building the payload with

to_state_create(Cancelling())

isn’t needed for this endpoint, and in 3.4.x it can trigger a Pydantic forward-ref issue. The bulk endpoint accepts a minimal JSON for

state

, so you can just send

{"type": "CANCELLING"}

(or

{"type": "CANCELLED"}

). Here’s a corrected, working version of Option A for Prefect 3.4.1:

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter, FlowRunFilterDeploymentId, FlowRunFilterState, FlowRunFilterStateName,
)

async def cancel_late_runs_for_deployment(flow_name: str, deployment_name: str, limit: int = 500, finalize: bool = False):
    """
    finalize=False -> set state to CANCELLING (good if some could be running)
    finalize=True  -> set state to CANCELLED (fine for SCHEDULED/Late runs)
    """
    async with get_client() as client:
        # Look up the deployment id
        dep = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")

        # Filter: this deployment + state name "Late"
        filt = FlowRunFilter(
            deployment_id=FlowRunFilterDeploymentId(any_=[dep.id]),
            state=FlowRunFilterState(name=FlowRunFilterStateName(any_=["Late"])),
        )

        payload = {
            "flow_runs": filt.model_dump(mode="json", exclude_unset=True),
            "state": {
                "type": "CANCELLED" if finalize else "CANCELLING"
            },
            "force": False,   # set True to bypass orchestration rules if needed
            "limit": limit,   # max 500 per call
        }

        resp = await client.request("POST", "/flow_runs/bulk_set_state", json=payload)
        print(resp.json())  # per-run orchestration results

# Example call
asyncio.run(cancel_late_runs_for_deployment("my_flow", "my_deployment", limit=500, finalize=False))

Notes - For “Late” runs that are not running, setting

CANCELLED

directly is typically fine; if any might be running, use

CANCELLING

. - If you have more than 500 matches, call this in a loop until no more runs are updated. References - Bulk set flow run state (REST) - FlowRunFilter schema If you’d like, I can adapt this to page through more than 500 results automatically or switch to

CANCELLED

for your case.

Thanks for sharing the traceback — that 404 explains it. What’s happening - The bulk endpoint POST /flow_runs/bulk_set_state was added after Prefect 3.4.1. It is not available on servers running 3.4.1 or earlier, so your request is hitting a valid base URL but an endpoint your server doesn’t implement yet. - The client version (3.4.1) does not guarantee the server has the same features; the server must be ≥ the version that introduced this endpoint. What you can do Option 1 — Upgrade your Prefect server - If you can upgrade to a version that includes the bulk endpoint, the payload you’re sending (with minimal state {"type": "CANCELLING"} or {"type": "CANCELLED"}) will work. - Reference: Bulk set flow run state (REST) Option 2 — Use an iterative cancel that works on 3.4.1 - This uses only methods that exist in 3.4.1: read_flow_runs + set_flow_run_state. - You can run this as-is; it cancels in batches and loops until no “Late” runs remain.

import asyncio
from typing import Sequence
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter, FlowRunFilterDeploymentId, FlowRunFilterState, FlowRunFilterStateName,
)
from prefect.states import Cancelling, Cancelled

async def cancel_late_runs_for_deployment_iter(
    flow_name: str,
    deployment_name: str,
    page_size: int = 200,
    finalize: bool = False,   # False -> CANCELLING, True -> CANCELLED
    force: bool = False,      # True to bypass orchestration rules if needed
):
    async with get_client() as client:
        dep = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")

        total = 0
        while True:
            filt = FlowRunFilter(
                deployment_id=FlowRunFilterDeploymentId(any_=[dep.id]),
                state=FlowRunFilterState(name=FlowRunFilterStateName(any_=["Late"])),
            )
            runs = await client.read_flow_runs(flow_run_filter=filt, limit=page_size)
            if not runs:
                break

            state_obj = Cancelled() if finalize else Cancelling()

            async def _cancel_one(fr_id):
                try:
                    await client.set_flow_run_state(fr_id, state=state_obj, force=force)
                except Exception as exc:
                    # log and continue
                    print(f"Failed to set state for {fr_id}: {exc}")

            await asyncio.gather(*[_cancel_one(fr.id) for fr in runs])
            total += len(runs)
            # Optional: small sleep to be gentle on the API
            await asyncio.sleep(0.2)

        print(f"Updated {total} Late runs for {flow_name}/{deployment_name} to "
              f"{'CANCELLED' if finalize else 'CANCELLING'}")

# Example
# asyncio.run(cancel_late_runs_for_deployment_iter("my_flow", "my_deployment", page_size=200, finalize=False, force=False))

Notes - For Late runs that have not started, setting CANCELLED directly is typically fine; for safety if any could be running, use CANCELLING. - If you see orchestration rejections (e.g., can’t transition from current state), set force=True. - Docs for the methods used: - Read flow runs (filters) - Set flow run state A couple of checks - What version is your Prefect server? If you’re not sure, checking the server container/env or the UI footer usually shows it. That’ll confirm whether bulk endpoints are available. - Your URL shows /api/… which is fine if your server is configured with a base path (the client will handle that). The 404 here is due to the endpoint not existing on that server version, not the base path. If upgrading is an option, I can give you a tiny loop that pages bulk_set_state in chunks; otherwise the iterative snippet above is safe on 3.4.1.

Thanks — with a Prefect server at 3.4.20, the 404 is expected. The bulk endpoint POST /flow_runs/bulk_set_state was introduced after 3.4.x, so it isn’t implemented on your server. Two paths forward: 1) Keep current server: use the iterative client approach (works on 3.4.20) - This loops over Late runs for the specific deployment and sets them to Cancelling (or Cancelled), batching to avoid huge requests.

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter, FlowRunFilterDeploymentId, FlowRunFilterState, FlowRunFilterStateName,
)
from prefect.states import Cancelling, Cancelled

async def cancel_late_runs_for_deployment_iter(
    flow_name: str,
    deployment_name: str,
    page_size: int = 200,
    finalize: bool = False,   # False -> CANCELLING, True -> CANCELLED
    force: bool = False,      # True to bypass orchestration rules if needed
):
    async with get_client() as client:
        dep = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")

        total = 0
        while True:
            filt = FlowRunFilter(
                deployment_id=FlowRunFilterDeploymentId(any_=[dep.id]),
                state=FlowRunFilterState(name=FlowRunFilterStateName(any_=["Late"])),
            )
            runs = await client.read_flow_runs(flow_run_filter=filt, limit=page_size)
            if not runs:
                break

            state_obj = Cancelled() if finalize else Cancelling()

            async def _cancel_one(fr_id):
                try:
                    await client.set_flow_run_state(fr_id, state=state_obj, force=force)
                except Exception as exc:
                    print(f"Failed to set state for {fr_id}: {exc}")

            await asyncio.gather(*[_cancel_one(fr.id) for fr in runs])
            total += len(runs)
            await asyncio.sleep(0.2)  # gentle on API

        print(f"Updated {total} Late runs for {flow_name}/{deployment_name} to "
              f"{'CANCELLED' if finalize else 'CANCELLING'}")

# Example
# asyncio.run(cancel_late_runs_for_deployment_iter("my_flow", "my_deployment", page_size=200, finalize=False, force=False))

Tips: - If Late runs are definitely not running, set finalize=True to go straight to CANCELLED. - If you hit orchestration rejections, set force=True. 2) If you can upgrade your Prefect server - Upgrading to a version that includes the bulk endpoint will let you use the single-call bulk_set_state approach you tried earlier. Want me to adapt the script to page through a very large backlog automatically or add logging/metrics?

Thanks for sticking with it. That Pydantic “StateCreate is not fully defined” error is coming from the client trying to build a

StateCreate

model internally when calling

set_flow_run_state

on Prefect client 3.4.1. It’s a known forward-ref quirk in that client version. Two solid fixes: A) Quick, version-agnostic workaround: call the per-run REST endpoint directly with minimal JSON - This bypasses the client’s

StateCreate

model entirely. - Works on your 3.4.20 server.

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
    FlowRunFilter, FlowRunFilterDeploymentId, FlowRunFilterState, FlowRunFilterStateName,
)

async def cancel_late_runs_for_deployment_iter_raw(
    flow_name: str,
    deployment_name: str,
    page_size: int = 200,
    finalize: bool = False,   # False -> CANCELLING, True -> CANCELLED
    force: bool = False,      # True to bypass orchestration rules if needed
):
    async with get_client() as client:
        dep = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")

        filt = FlowRunFilter(
            deployment_id=FlowRunFilterDeploymentId(any_=[dep.id]),
            state=FlowRunFilterState(name=FlowRunFilterStateName(any_=["Late"])),
        )

        total = 0
        while True:
            runs = await client.read_flow_runs(flow_run_filter=filt, limit=page_size)
            if not runs:
                break

            state_type = "CANCELLED" if finalize else "CANCELLING"

            async def _post(fr_id):
                try:
                    r = await client.request(
                        "POST",
                        f"/flow_runs/{fr_id}/set_state",
                        json={"state": {"type": state_type}, "force": force},
                    )
                    if r.status_code != 200:
                        try:
                            print(f"Failed for {fr_id}: {r.status_code} {r.json()}")
                        except Exception:
                            print(f"Failed for {fr_id}: {r.status_code}")
                except Exception as exc:
                    print(f"Request failed for {fr_id}: {exc}")

            await asyncio.gather(*[_post(fr.id) for fr in runs])
            total += len(runs)
            await asyncio.sleep(0.2)

        print(f"Updated {total} Late runs for {flow_name}/{deployment_name} to {state_type}")

# Example
# asyncio.run(cancel_late_runs_for_deployment_iter_raw("my_flow", "my_deployment", page_size=200, finalize=False, force=False))

Refs: - Read flow runs (filters) - Set flow run state (single run) B) Longer-term: upgrade the Prefect client - Upgrading your local client to match the server minor (e.g., 3.4.20) or latest 3.x typically resolves this Pydantic forward-ref issue. - After upgrading, the simple approach should work:

await client.set_flow_run_state(fr_id, state=Cancelling(), force=force)

Notes - Filtering by “Late” uses the state name exactly as

"Late"

in the filter. - If you know Late runs are not actually executing,

finalize=True

(CANCELLED) is fine. If any could be running, use CANCELLING. - If orchestration blocks the transition, set

force=True

. If you want, I can add paging over very large backlogs and structured logging, or adapt this to cancel by multiple deployments at once.