Skip to content

Typing Async Code in Python

Async code has a class of bugs that synchronous code does not: values that look like results but are coroutine objects, decorators that quietly erase signatures, generators that cannot be closed, and calls that never happen because nothing awaited them. Most of them pass tests that do not exercise the exact path, and most of them are visible to a type checker before the code runs — if the annotations say the right thing. This section is about saying the right thing. Every behaviour described was checked with mypy 2.4.0 in --strict mode and pyright 1.1.414 on Python 3.14, with ruff 0.16.10 for the syntactic rules. In a file of eight common async mistakes, default mypy and pyright each caught 2; opt-in checks and ruff together caught 6; and the last two needed pytest configured to fail on never-awaited coroutines. An Any-typed retry decorator turned a call with a wrong argument type and a misspelt keyword into zero errors; a ParamSpec version made it two. A gather over seven different calls lost its types in mypy (list[object]); over six it kept them.

The parent section, Asyncio Fundamentals & Event Loop Architecture, covers what coroutines, tasks and futures are at runtime. This topic covers how to describe them so the checker can hold the code to it.

Architectural principles

  • Annotate what await produces, not what the call returns. async def f() -> User is right; the checker adds the Coroutine wrapper itself.
  • Accept wide, return narrow. Parameters take Awaitable[T] and AsyncIterable[T]; functions return Coroutine[Any, Any, T] and AsyncGenerator[T, None] when callers need to schedule or close them.
  • Never let a decorator erase a signature. Callable[..., Any] disables checking for everything it wraps; ParamSpec keeps it.
  • Describe dependencies as protocols with async def methods, and type-check the tests that supply fakes.
  • Treat the type checker as one layer of three. Static types, syntactic lint rules and test-time warnings each catch a different slice of missing-await bugs.
Three layers that catch async mistakes 3 stacked layers. Three layers that catch async mistakes type checker discarded coroutines, wrong awaitable types, signature drift lint rules blocking calls in async def, unreferenced tasks, unused values tests with warnings as errors coroutines created and never awaited In the eight-bug sample, the static layers caught 6 and the test layer the remaining 2.

Execution model: what the checker sees

The event loop runs coroutine objects; the type checker reasons about them. Calling an async def function produces a Coroutine[Any, Any, T] — mypy reveals typing.Coroutine, pyright the concrete types.CoroutineType — and only await turns it into T. Everything else follows from that one rule. A coroutine object assigned to a variable declared str is a type error, which is how annotations find missing awaits. A coroutine passed to asyncio.create_task is scheduled on the loop and becomes a Task[T], which is a Future[T], which is Awaitable[T]. A coroutine used in an if is an object, always truthy, and that is how an authorization check that forgot its await returned "ok" for user "guest" in testing.

The scheduler's own requirements show up in the types too. create_task accepts only coroutines: given a future it raised TypeError: a coroutine was expected, got <Future finished result=1>, and both checkers rejected an argument typed Awaitable[str] before it ever ran. Async iteration is driven by __aiter__ (a plain method) and awaited __anext__ calls; the checkers know that a synchronous for over an async iterator, or an await on one, cannot work, and both rejected each. The rest of this section is about keeping that information intact as code passes coroutines, iterators and callables through layers of its own.

How one async call is typed from definition to result A flow of 5 stages. How one async call is typed from definition to result async def fetch() -> User declared fetch() Coroutine[Any, Any, User] create_task(fetch()) Task[User] await task User missing await Coroutine where User expected The checker follows the coroutine through every hop; a missing await breaks the chain visibly.

Pattern catalogue

Awaitable for parameters, Coroutine for scheduling

A callback that is only awaited should accept any awaitable, so callers may pass coroutine functions, functions returning tasks, or custom awaitables. A callback whose result is scheduled must be a coroutine:

from collections.abc import Awaitable, Callable, Coroutine
from typing import Any


async def run_awaitable(make: Callable[[int], Awaitable[str]]) -> str:
    return await make(1)                         # any awaitable is fine


async def run_task(make: Callable[[int], Coroutine[Any, Any, str]]) -> str:
    task = asyncio.create_task(make(1))          # Task[str]
    return await task

Typing run_task's parameter as Awaitable produced Argument 1 to "create_task" has incompatible type "Awaitable[str]" in mypy, plus two knock-on errors, and the equivalent in pyright. Passing a synchronous function to run_awaitable was rejected by both. Details and the registry pattern are in typing coroutine functions with Awaitable and Coroutine.

Signature-preserving decorators

Retry, timeout, tracing and caching decorators wrap coroutine functions. Typed with ParamSpec and returning Coroutine, they are transparent to the checker:

def retry[**P, R](attempts: int = 3) -> Callable[
    [Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]]
]:
    def deco(fn: Callable[P, Coroutine[Any, Any, R]]) -> Callable[P, Coroutine[Any, Any, R]]:
        @functools.wraps(fn)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            for attempt in range(attempts - 1):
                try:
                    return await fn(*args, **kwargs)
                except ConnectionError:
                    await asyncio.sleep(0.05 * 2 ** attempt)
            return await fn(*args, **kwargs)
        return wrapper
    return deco

The Callable[..., Any] version revealed decorated functions as (*Any, **Any) -> Any and let a call with "42" for an int and a misspelt keyword through both checkers. Returning Awaitable[R] instead of Coroutine kept the signature but broke create_task on decorated calls. Concatenate covers decorators that inject a leading argument. See typing async decorators with ParamSpec.

Async iterators and generators

Generator functions return AsyncIterator[T] or, when callers must be able to close them, AsyncGenerator[T, None]; consumers accept AsyncIterable[T]:

async def rows(cursor: Cursor) -> AsyncGenerator[Row, None]:
    try:
        while batch := await cursor.fetch(500):
            for row in batch:
                yield row
    finally:
        await cursor.close()


async def count(source: AsyncIterable[Row]) -> int:
    return sum([1 async for _ in source])

async with aclosing(rows(cur)) as it:
    n = await count(it)

With -> AsyncIterator[Row], both checkers rejected aclosing(rows(cur)) and .aclose() — although the runtime object has both. @asynccontextmanager functions take the same iterator annotation; annotating them with the yielded type produced two mypy errors and four pyright errors. See typing async iterators and generators.

Protocols for async dependencies

Services depend on protocols, and fakes satisfy them structurally:

class OrderStore(Protocol):
    async def page(self, after: int, limit: int) -> list[Order]: ...
    async def mark_billed(self, order_id: int) -> None: ...


class Billing(Protocol):                         # a callable dependency with a keyword
    async def __call__(self, order: Order, *, idempotency_key: str) -> str: ...

Both checkers rejected a fake with synchronous methods and listed the expected Coroutine return per method; isinstance() against a @runtime_checkable version of the same protocol returned True for that fake, which then raised TypeError: 'dict' object can't be awaited. A bare AsyncMock() type-checks as Any and checks nothing. See defining async protocols for dependency injection.

Typed fan-out

gather keeps per-position types for up to six awaitables; TaskGroup keeps them for any number, one variable per task:

user, orders = await asyncio.gather(fetch_user(uid), fetch_orders(uid))   # User, list[Order]

async with asyncio.TaskGroup() as tg:
    user_t = tg.create_task(fetch_user(uid))
    orders_t = tg.create_task(fetch_orders(uid))
    prefs_t = tg.create_task(fetch_prefs(uid))
render(user_t.result(), orders_t.result(), prefs_t.result())

With return_exceptions=True every position becomes T | BaseException, and narrowing must use BaseException: a cancelled child came back as CancelledError, which is not an Exception. See typing TaskGroup and gather results.

The pattern catalogue, with the mistake each one prevents A grid of 5 rows by 3 columns. The pattern catalogue, with the mistake each one prevents pattern prevents measured Coroutine for scheduled callables create_task on an Awaitable rejected by both ParamSpec decorators erased signatures 0 vs 2 errors on a bad call AsyncGenerator returns unclosable generators aclose() rejected otherwise async-method protocols sync fakes isinstance said True, await crashed TaskGroup task variables untyped wide gathers list[object] at 7 args Each pattern corresponds to one guide in this section.

Where checking stops

Types only protect code the checker can see into. Four boundaries routinely leak Any into async code:

  • Untyped libraries. A client library without type hints or stubs returns Any from every coroutine, so a missing await on its calls is invisible. Install stub packages where they exist, or wrap the library behind a small typed protocol of your own.
  • Mocks. AsyncMock is typed as Any; tests using bare mocks are not checked against the interface. create_autospec restores argument checking at runtime; a hand-written fake restores it statically.
  • return_exceptions and asyncio.wait. Results widen to unions or object; narrow them immediately.
  • Dynamic dispatch. Handler tables built with getattr or string keys erase types unless the table itself has a declared callable type.

The cost of checking is not a reason to skip it: on the integrated example below, a cold mypy run took 0.67 s and a warm one, with its cache, 0.09 s; pyright took 0.67–0.78 s; ruff finished in under 10 ms. All three are fast enough for a pre-commit hook on changed files and a full run in CI.

Adopting strict checks in an existing codebase

Turning on --strict across a large async codebase at once produces thousands of errors and gets reverted. A staged rollout keeps the build green while the coverage grows:

  1. Enable the cheap, high-value checks everywhere first. mypy's default unused-coroutine, pyright's default reportUnusedCoroutine, and ruff's RUF006 and ASYNC rules have almost no false positives and find real bugs on day one.
  2. Make the test suite fail on never-awaited coroutines. The two pytest filters cost nothing to add and catch what the static layers cannot.
  3. Type the shared plumbing next. Decorators, base clients, repositories and task helpers are imported everywhere; fixing one Callable[..., Any] decorator restores checking for every function it wraps.
  4. Turn on strict mode per package, using [[tool.mypy.overrides]] or pyright's strict list, starting with the packages that already pass.
  5. Add the opt-in codes last — truthy-bool and unused-awaitable — once the remaining noise is low enough that new findings are noticed.

Each step is independently useful, and the order puts the checks that catch silent async failures ahead of those that mainly improve annotations.

Integrated production example

A billing job that pages through unbilled orders, charges each with retries and bounded concurrency, and reports failures — every boundary typed with the patterns above. It passed mypy --strict with the truthy-bool and unused-awaitable codes and pyright in strict mode with no errors, and billed 245 of 250 orders against an in-memory store whose fake billing declined every fiftieth:

import asyncio
import functools
import logging
from collections.abc import AsyncGenerator, AsyncIterable, Callable, Coroutine
from contextlib import aclosing
from dataclasses import dataclass
from typing import Any, Protocol

log = logging.getLogger("orders")


@dataclass(frozen=True)
class Order:
    id: int
    customer_id: int
    total_cents: int


class OrderStore(Protocol):
    async def page(self, after: int, limit: int) -> list[Order]: ...
    async def mark_billed(self, order_id: int) -> None: ...


class Billing(Protocol):
    async def __call__(self, order: Order, *, idempotency_key: str) -> str: ...


def retry[**P, R](attempts: int = 3) -> Callable[
    [Callable[P, Coroutine[Any, Any, R]]], Callable[P, Coroutine[Any, Any, R]]
]:
    def deco(fn: Callable[P, Coroutine[Any, Any, R]]) -> Callable[P, Coroutine[Any, Any, R]]:
        @functools.wraps(fn)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            for attempt in range(attempts - 1):
                try:
                    return await fn(*args, **kwargs)
                except ConnectionError:
                    await asyncio.sleep(0.05 * 2 ** attempt)
            return await fn(*args, **kwargs)
        return wrapper
    return deco


async def unbilled(store: OrderStore, limit: int = 100) -> AsyncGenerator[Order, None]:
    after = 0
    while batch := await store.page(after, limit):
        for order in batch:
            yield order
        after = batch[-1].id


async def bill_one(order: Order, charge: Billing, store: OrderStore) -> str:
    receipt = await retry(attempts=3)(charge)(order, idempotency_key=f"order-{order.id}")
    await store.mark_billed(order.id)
    return receipt


async def bill_all(orders: AsyncIterable[Order], charge: Billing, store: OrderStore,
                   concurrency: int = 10) -> dict[int, str | BaseException]:
    sem = asyncio.Semaphore(concurrency)
    results: dict[int, str | BaseException] = {}

    async def guarded(order: Order) -> None:
        async with sem:
            try:
                results[order.id] = await bill_one(order, charge, store)
            except Exception as exc:          # one bad order must not stop the batch
                results[order.id] = exc

    async with asyncio.TaskGroup() as tg:
        async for order in orders:
            tg.create_task(guarded(order))
    return results


async def run(store: OrderStore, charge: Billing) -> int:
    async with aclosing(unbilled(store)) as orders:
        results = await bill_all(orders, charge, store)
    failed = {k: v for k, v in results.items() if isinstance(v, BaseException)}
    for order_id, exc in failed.items():
        log.warning("order %d not billed: %r", order_id, exc)
    return len(results) - len(failed)

Every layer is visible to the checker: the store and billing dependencies are protocols, so test fakes are verified; the decorator keeps charge's keyword-only idempotency_key; the generator returns AsyncGenerator so aclosing type-checks; and the result map's str | BaseException forces the narrowing in run. To test that claim, each of the ten await expressions in the full file (including the in-memory store and fake billing used to run it) was removed one at a time: all ten mutants failed in both checkers, with between one and nine errors each.

Diagnostic hook callout

Three signals show whether a codebase's async typing is doing its job:

  • Any leakage. Run mypy --strict --warn-return-any and count no-any-return errors in async modules; a decorator or untyped library usually explains a cluster of them.
  • Erased signatures. Add reveal_type() for each decorator's output in a scratch module; anything showing *Any, **Any or (...) is disabling checks downstream.
  • Runtime residue. Configure pytest with filterwarnings = error::RuntimeWarning and error::pytest.PytestUnraisableExceptionWarning; with either alone, a test calling an unawaited coroutine still passed in testing. Any failure from this is a missing await the static layers missed — the guide on catching missing awaits lists which tool catches which shape.

Alert threshold, in CI terms: zero type errors in async modules, zero RUF006 and ASYNC findings, and zero never-awaited warnings in the test run. These are binary, so treat any regression as a failed build rather than a trend.

Failure modes

Failure mode Root cause Detection Fix
Wrong argument accepted by a decorated function Decorator typed Callable[..., Any] reveal_type shows (*Any, **Any) ParamSpec decorator returning Coroutine
create_task rejects a callback's result Parameter typed Awaitable[T] arg-type error on create_task Narrow to Coroutine[Any, Any, T]
aclose() or aclosing() rejected Generator annotated AsyncIterator[T] attr-defined / type-var errors Return AsyncGenerator[T, None]
Fake crashes when awaited Synchronous fake; untyped tests Checker run over tests/ Async methods; protocol-typed parameters
Authorization always passes if coro(): without await mypy truthy-bool, pyright strict Enable the checks; add the await
Fan-out results typed object gather over 7+ mixed awaitables list[object] in reveal_type Split, or one TaskGroup variable per task
Cancellation swallowed return_exceptions results filtered by BaseException and dropped Review; cancelled jobs report success Re-raise CancelledError explicitly

Frequently Asked Questions

Is mypy or pyright better for asyncio code?

In testing they found the same type errors with different wording. Pyright in strict mode caught a truthy coroutine and an unused coroutine variable by default; mypy needed its opt-in truthy-bool and unused-awaitable codes to catch a truthy coroutine and a dropped task. Either works; enable the opt-in checks.

How do I type an async function in Python?

Annotate the value that await produces: async def fetch() -> User. The checker types the call as Coroutine[Any, Any, User]. Use Awaitable[T] for parameters you only await and Coroutine[Any, Any, T] for ones you pass to create_task.

Why does my async decorator lose the function's signature?

It is typed with Callable[..., Any]. Use ParamSpec and return Callable[P, Coroutine[Any, Any, R]]; in testing that turned a bad call from zero errors into two in mypy.

Can a type checker catch every missing await?

No. In an eight-bug sample, mypy, pyright and ruff together caught six. A coroutine passed to print and a list of coroutines never gathered were only caught by pytest configured to treat both RuntimeWarning and PytestUnraisableExceptionWarning as errors.

How should I type gather with return_exceptions=True?

Each element is T | BaseException; narrow with isinstance(r, BaseException) and re-raise CancelledError, because cancellation results are BaseException instances, not Exception.