Here’s how to cancel a task run in Prefect 3.x using both the Python SDK and the REST API. Assumptions: - You’re on Prefect 3.x. - You already have the task run ID you want to cancel. Key idea: - To initiate cancellation, set the task run to a Cancelling state. The system will try to stop execution and transition the run to Cancelled. If you must immediately mark it cancelled (bypassing orchestration), you can force a Cancelled state. Links: - API route docs: Set Task Run State (REST) - State types overview: States in Prefect Python SDK (async)

from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.states import Cancelling, Cancelled

task_run_id = UUID("00000000-0000-0000-0000-000000000000")  # replace

async def cancel_task_run(task_run_id: UUID):
    async with get_client() as client:
        # Preferred: request cancellation (cooperative)
        result = await client.set_task_run_state(
            task_run_id=task_run_id,
            state=Cancelling(message="User requested cancellation")
        )
        print(result.status)  # e.g., "ACCEPT", "REJECT", etc.

        # If you must hard-mark it as cancelled (bypass orchestration), use force=True:
        # await client.set_task_run_state(task_run_id, Cancelled(message="Force-cancelled"), force=True)

Python SDK (sync)

from uuid import UUID
from prefect.client.orchestration import SyncPrefectClient
from prefect.states import Cancelling, Cancelled

task_run_id = UUID("00000000-0000-0000-0000-000000000000")  # replace

with SyncPrefectClient() as client:
    # Preferred: request cancellation (cooperative)
    result = client.set_task_run_state(
        task_run_id=task_run_id,
        state=Cancelling(message="User requested cancellation")
    )
    print(result.status)

    # Force a terminal Cancelled state (optional, not usually needed):
    # client.set_task_run_state(task_run_id, Cancelled(message="Force-cancelled"), force=True)

REST API - Endpoint: POST

/api/task-runs/{id}/set-state

- Body: a

state

object (StateCreate) specifying the new state. To request cancellation, use type

CANCELLING

. To force terminal cancellation, use

CANCELLED

. - In Prefect Cloud, include your API key. Example: request cancellation (preferred)

curl -X POST "$PREFECT_API_URL/api/task-runs/<TASK_RUN_ID>/set-state" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
        "state": {
          "type": "CANCELLING",
          "message": "User requested cancellation"
        }
      }'

Example: force mark as cancelled (terminal)

curl -X POST "$PREFECT_API_URL/api/task-runs/<TASK_RUN_ID>/set-state?force=true" \
  -H "Authorization: Bearer $PREFECT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
        "state": {
          "type": "CANCELLED",
          "message": "Force-cancelled by operator"
        }
      }'

Notes and tips: - Cancelling is cooperative: the runtime must be able to acknowledge cancellation. Long-running blocking code may not stop immediately unless it checks for cancellation or runs in infrastructure that supports termination. - If your goal is to stop an entire flow, you can set the flow run to

Cancelling

or

Cancelled

similarly; that will cascade to its task runs. - The server returns an orchestration result indicating whether the state was accepted. If rejected, you may need

force=True

(SDK) or add

?force=true

on the REST endpoint if you truly want to override orchestration.

Thanks for the details — a couple key points explain what you’re seeing: - In a live flow run (i.e., not run by a Prefect worker), flipping a task run’s state to Cancelling via the API is accepted by orchestration, but it does not preempt or kill the Python that’s already running. With a DaskTaskRunner, your task code is running on Dask workers in separate processes; unless your task cooperatively exits, it will keep going. - Prefect cancellation is cooperative at the task level. To actually stop work, your task must either: - detect the cancellation and exit, or - be running in infrastructure where a worker can truly terminate the process (e.g., a deployed flow run managed by a worker). What to do next 1) Make tasks cooperative (recommended) - For long-running tasks, periodically check whether the task has been cancelled and exit cleanly by raising

prefect.exceptions.CancelledRun

. This yields a terminal Cancelled state instead of Failed. Async example (Dask works fine with async tasks too):

from prefect import task
from prefect.exceptions import CancelledRun
from prefect.runtime import task_run
from prefect.client.orchestration import get_client
import anyio

@task
async def long_running_task():
    async with get_client() as client:
        while True:
            # Do a chunk of work here...

            # Yield to the event loop so cancellation can be processed
            await anyio.sleep(0.2)

            # Poll orchestration for cancellation
            tr = await client.read_task_run(task_run.id)
            # Depending on your needs, treat either Cancelling or Cancelled as a stop signal
            if tr.state and tr.state.name in {"Cancelling", "Cancelled"}:
                raise CancelledRun("Cancelled by user")

Sync example:

from prefect import task
from prefect.exceptions import CancelledRun
from prefect.runtime import task_run
from prefect.client.orchestration import SyncPrefectClient
import time

@task
def long_running_cpu_task():
    with SyncPrefectClient() as client:
        while True:
            # Do a chunk of work...

            # Periodically poll for cancellation
            tr = client.read_task_run(task_run.id)
            if tr.state and tr.state.name in {"Cancelling", "Cancelled"}:
                raise CancelledRun("Cancelled by user")
            time.sleep(0.1)  # give the loop a checkpoint

Notes: - For async tasks, ensure you “yield” regularly (e.g.,

await anyio.sleep(...)

) so cancellation can be observed. - For CPU-bound loops, break work into chunks and check between chunks. A single, long, blocking call in C-extension code won’t be interruptible. 2) Prefer cancelling the flow run (if appropriate) If you cancel the overall flow run (instead of individual tasks), Prefect’s engine will enter cancellation scopes and attempt to cancel child tasks. In a live run, this is still cooperative. When run as a deployment on a worker, the worker can terminate the process which is more forceful. 3) Consider running as a deployment on a worker (for enforceable cancellation) When a flow is run by a Prefect worker (Process/Docker/Kubernetes), cancelling the flow run via the UI/API instructs the worker to terminate the underlying process/container/pod. This is the closest thing to a “hard” kill that Prefect supports and will stop a stuck flow even if the task doesn’t cooperate. 4) Timeouts as a safety net You can set timeouts to avoid indefinite tasks:

from prefect import task

@task(timeout_seconds=600)  # 10 minutes
def maybe_hanging_task():
    ...

Dask-specific caveats - Prefect does not automatically call

future.cancel()

on Dask futures when you flip a task’s state to Cancelling from the API for a live run. Without a worker managing the run, there’s no external controller to do so. - Even with Dask, cancellation inside a task is best accomplished by cooperative checks (and raising

CancelledRun

) as shown above. Small note about your snippet

- Make sure you pass an actual UUID object to your

cancel_task_run

function. In the example you showed,

asyncio.run(cancel_task_run('my_task_run_id'))

passes a string; it should be

asyncio.run(cancel_task_run(UUID("<actual-task-run-uuid>")))

. Good references - Cancel workflows (Prefect 3) - States in Prefect 3 - prefect-dask integration If you can share whether you’d prefer to: - keep running “live” but make your tasks cancellable, or - move to a deployment/worker so cancellation can be enforced, I can tailor a concrete example for your setup (including Dask cluster configs and task structure).

Got it — thanks for clarifying it’s a deployment run by a Prefect worker. Key behavior with Dask in deployments - Cancelling a task run via orchestration is a cooperative request. With DaskTaskRunner, the user function is running on a Dask worker process. Dask cannot forcibly kill a Python function that is already executing; at best, it can cancel queued work or mark the future as cancelled. So a running Dask task will keep executing unless your task cooperates and exits. - Cancelling the flow run in a deployment instructs the Prefect worker to stop the flow’s process/container/pod, which ends the flow orchestration. However, this does not “kill” an already-running Dask function on a remote Dask worker; the Dask cluster is separate infrastructure. What will actually stop the work 1) Cancel the entire flow run (enforceable at the worker): - This stops the Prefect runtime. It won’t terminate a running Dask function, but it will stop the flow’s orchestration immediately so your run doesn’t hang. - CLI (verified):

prefect flow-run cancel <flow_run_id>

- Or via API:

from prefect.client.orchestration import get_client
from prefect.states import Cancelling

async def cancel_flow_run(flow_run_id):
    async with get_client() as client:
        res = await client.set_flow_run_state(flow_run_id, Cancelling(message="User cancel"))
        print(res.status)

Ref: Cancel workflows (Prefect 3) 2) Make the Dask-executed task cooperative (to stop a single task) - Add periodic checks for cancellation and exit by raising

prefect.exceptions.CancelledRun

. This lets Prefect mark the task Cancelled and your flow won’t wait forever on its future. Async pattern:

from prefect import task
from prefect.exceptions import CancelledRun
from prefect.runtime import task_run
from prefect.client.orchestration import get_client
import anyio

@task
async def long_running_task():
    async with get_client() as client:
        while True:
            # do a chunk of work...

            await anyio.sleep(0.2)  # yield so cancellation can be processed

            tr = await client.read_task_run(task_run.id)
            if tr.state and tr.state.name in {"Cancelling", "Cancelled"}:
                raise CancelledRun("Cancelled by user")

Sync pattern:

from prefect import task
from prefect.exceptions import CancelledRun
from prefect.runtime import task_run
from prefect.client.orchestration import SyncPrefectClient
import time

@task
def long_running_cpu_task():
    with SyncPrefectClient() as client:
        while True:
            # do a chunk of work...

            tr = client.read_task_run(task_run.id)
            if tr.state and tr.state.name in {"Cancelling", "Cancelled"}:
                raise CancelledRun("Cancelled by user")
            time.sleep(0.1)  # checkpoint between chunks

3) Add timeouts as a safety net

from prefect import task

@task(timeout_seconds=600)
def maybe_hanging_task():
    ...

Why your per-task Cancelling request didn’t stop execution - Setting a task run to Cancelling is accepted by orchestration, but Dask cannot interrupt an already-running function. Without cooperative checks in the task, it will continue running on the Dask worker process even though Prefect marks it Cancelling/Cancelled. If your goal is “stop this run now,” prefer cancelling the flow run (CLI above). If your goal is “stop just this long-running task,” add the cooperative checks (and/or a timeout) to the task. Helpful links - Cancel workflows (Prefect 3) - prefect-dask integration - States in Prefect 3 If you can share your Dask setup (local cluster via DaskTaskRunner vs remote cluster, K8s, etc.), I can suggest a concrete pattern to make your specific tasks cancellable and ensure the flow doesn’t hang.

Perfect — with a deployment run by a worker and a temporary PBSCluster behind Dask, here’s how to wire things so cancellation is effective and your CPU task doesn’t run forever. DaskTaskRunner with a temporary PBSCluster - Using

cluster_class="dask_jobqueue.PBSCluster"

creates an ephemeral Dask cluster for the flow run; Prefect will close it when the flow ends (including after cancellation), which will issue

qdel

to shut down PBS jobs. - Tune

cluster_kwargs

for your queue, cores, memory, walltime, etc., and scale/adapt as needed. Example:

from prefect import flow
from prefect_dask.task_runners import DaskTaskRunner

# An ephemeral PBSCluster per flow run
task_runner = DaskTaskRunner(
    cluster_class="dask_jobqueue.PBSCluster",
    cluster_kwargs={
        "queue": "workq",
        "cores": 4,
        "memory": "16GB",
        "walltime": "01:00:00",
        # PBS-specific options:
        # "job_extra": ["-l nodes=1:ppn=4"],
        # "resource_spec": "nodes=1:ppn=4",
        # "shebang": "#!/usr/bin/env bash",
        # "local_directory": "/scratch/$USER/dask",
    },
    adapt_kwargs={"minimum": 0, "maximum": 10},  # or use client.scale(n)
    client_kwargs={"set_as_default": True},
)

@flow(name="pbs-dask-flow", task_runner=task_runner)
def my_flow():
    # call your CPU task here
    ...

CPU-bound task with chunking and cooperative cancel checks - Break work into chunks and check for cancellation between chunks. - Raise

prefect.exceptions.CancelledRun

to mark the task as Cancelled (not Failed). - For performance, don’t check every iteration — check every N chunks or every X seconds. Iteration-based check example:

from prefect import task
from prefect.exceptions import CancelledRun
from prefect.runtime import task_run
from prefect.client.orchestration import SyncPrefectClient
import time

@task
def long_running_cpu_task(total_items: int = 10_000_000, chunk_size: int = 100_000,
                          check_every_n_chunks: int = 20):
    """
    Process 'total_items' in chunks. Checks for cancellation every
    'check_every_n_chunks' chunks.
    """
    processed = 0
    chunks_done = 0

    with SyncPrefectClient() as client:
        while processed < total_items:
            end = min(processed + chunk_size, total_items)

            # Simulate heavy CPU work for this chunk
            # Replace with your real computation, e.g. vectorized numpy ops,
            # hashing, encoding, parsing, etc.
            acc = 0
            for i in range(processed, end):
                acc += (i % 97) * (i % 193)

            processed = end
            chunks_done += 1

            # Check cancellation occasionally
            if chunks_done % check_every_n_chunks == 0:
                tr = client.read_task_run(task_run.id)
                state_name = getattr(tr.state, "name", None)
                if state_name in {"Cancelling", "Cancelled"}:
                    raise CancelledRun(f"Cancelled at progress {processed}/{total_items}")

            # Optional: tiny sleep to yield CPU or IO; keep small for CPU tasks
            # time.sleep(0.001)

    return processed  # or your final result

Time-based check example (useful if chunk sizes vary): ``` from prefect import task from prefect.exceptions import CancelledRun from prefect.runtime import task_run from prefect.client.orchestration import SyncPrefectClient import time @task def long_running_cpu_task_time_checked(total_items: int = 10_000_000, chunk_size: int = 100_000, check_interval_s: float = 1.0): processed = 0 last_check = time.monotonic() with SyncPrefectClient() as client: while processed < total_items: end = min(processed + chunk_size, total_items) # Heavy chunk of work acc = 0 for i in range(processed, end): acc += (i % 97) * (i % 193) processed = end

if time.monotonic() - last_check >= check_interval_s: last_check = time.monotonic() tr = client.read_task_run(task_run.id) state_name = getattr(tr.state, "name", None) if state_name in {"Cancelling", "Cancelled"}: raise CancelledRun(f"Cancelled at progress {processed}/{total_items}") return processed

Why this works with PBSCluster
- Cancelling an individual task run is cooperative: the Dask worker process cannot be force-killed mid-function by Prefect. The chunking + `CancelledRun` pattern ensures the function exits promptly.
- Cancelling the flow run will stop the Prefect orchestration process/container; when the DaskTaskRunner context exits, it calls `cluster.close()`, which should cancel PBS jobs (via `qdel`) and tear down the ephemeral cluster.

Tips
- Combine with a timeout on the task as a safety net:

@task(timeout_seconds=3600) def long_running_cpu_task(...): ... ``` - Ensure the Dask workers launched by PBS have network access to reach your Prefect API; each cancellation check calls the API. - Tune chunk size and check frequency to balance responsiveness vs. overhead for your workload. References - Prefect Dask integration - Cancel workflows (Prefect 3) - Dask Jobqueue PBSCluster docs If you share your PBS queue settings (cores/memory/walltime) and workload shape, I can suggest concrete

cluster_kwargs

and chunk/check parameters.