Shut Down an asyncio Service Gracefully

Your service pulls jobs from a queue and handles each one in its own task. When you deploy a new version, the process manager sends the process a SIGTERM signal, waits a while, then kills it. Docker waits 10 seconds by default, and Kubernetes waits 30. If the service ignores SIGTERM, every job in progress dies halfway through.

A graceful shutdown completes within that wait. It stops taking new jobs, lets the jobs in progress finish, and cancels whatever is still running when time is up.

Turn the signal into an event

On Linux and macOS, loop.add_signal_handler runs a plain function on the event loop when a signal arrives. Have it set an asyncio.Event, and let your code await the event:

import asyncio
import signal


async def main() -> None:
    stopping = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGTERM, stopping.set)
    loop.add_signal_handler(signal.SIGINT, stopping.set)  # Ctrl-C

    loop.call_later(0.1, signal.raise_signal, signal.SIGTERM)  # pretend to deploy
    await stopping.wait()
    print("stop requested")


asyncio.run(main())

Follow these rules for the handler:

Pass a plain function such as stopping.set. An async def function raises TypeError here.
Make a repeated request harmless. Setting an event twice is the same as setting it once, so a second Ctrl-C changes nothing.
From another thread, call loop.call_soon_threadsafe(stopping.set).

Shut down in three steps

Stop intake. Stop pulling new jobs. The intake loop may be stuck waiting for the next job, and a flag it checks between jobs cannot wake it. Cancel the intake task instead.
Drain. Wait for the jobs in progress, up to a grace period.
Cancel the rest. Cancel the jobs still running, and wait again.

asyncio.wait(tasks, timeout=...) does the waiting. It returns two sets, the tasks that finished and the ones still pending. It never cancels anything, even when the timeout runs out:

import asyncio


async def job(seconds: float) -> None:
    await asyncio.sleep(seconds)


async def drain(tasks: set[asyncio.Task[None]], grace: float) -> set[asyncio.Task[None]]:
    if not tasks:
        return tasks
    _, pending = await asyncio.wait(tasks, timeout=grace)
    for task in pending:
        task.cancel()
    if pending:
        _, pending = await asyncio.wait(pending, timeout=grace)
    return pending


async def main() -> None:
    tasks = {asyncio.create_task(job(s)) for s in (0.01, 0.02, 60)}
    left = await drain(tasks, grace=0.1)
    print(len(left), sum(t.cancelled() for t in tasks))  # 0 1


asyncio.run(main())

The two quick jobs finish in the first wait, and the slow one is cancelled. The if checks matter because asyncio.wait raises ValueError when you give it an empty set.

Cancelling is a request

task.cancel() asks the task to stop. The task sees CancelledError at its next await, runs its finally blocks, and usually ends. It can also catch the error and keep going. That is why drain waits again after cancelling, and returns the tasks that are still running.

When tasks survive, say so. Log them, keep your references to them, and report that the shutdown did not finish. If you report a clean stop, nobody learns about them until the process manager kills the process, and a kill runs no cleanup at all.

When your own stop request started the shutdown, the service finishes normally and returns. A cancellation from the code that called your service is different. Clean up, then re-raise the CancelledError so the caller sees it.

Collect every failure

When a handler raises, its task stores the exception. If nothing reads it, asyncio logs "Task exception was never retrieved" when the task is garbage collected, long after the failure. Add a done callback that calls task.exception() on each finished task and logs the result. Skip cancelled tasks, because exception() raises CancelledError for them.

Name the states

A service moves through four states. A Literal type lists them, and assert_never makes the type checker report any state a match has no case for:

from typing import Literal, assert_never

type State = Literal["idle", "running", "draining", "stopped"]


def describe(state: State) -> str:
    match state:
        case "idle":
            return "not started"
        case "running":
            return "accepting work"
        case "draining":
            return "finishing in-flight work"
        case "stopped":
            return "done"
        case _:
            assert_never(state)

If you add "paused" to State and forget its case, the type checker reports an error on the assert_never line.

Check your understanding Your drain calls asyncio.wait(tasks, timeout=5), and two tasks are still running when it returns. What state are those two tasks in? Show answer

They are still running. asyncio.wait only stops waiting, and it never cancels anything. Cancel the two tasks yourself and wait again. If they survive that too, report them instead of claiming a clean stop.

Going further

asyncio.run cancels any tasks still running when your main coroutine returns, and waits for them. That happens all at once, with no grace period, so drain your own work first. add_signal_handler works only on Unix, and only from the main thread.

Read more about asyncio runners and signal handlers on the event loop.