Skip to content

Typing Async Decorators with ParamSpec

Decorators are where async codebases quietly lose their types. A retry, timeout or tracing wrapper written with Callable[..., Any] returns a function that accepts anything and returns Any, and every function it decorates silently stops being checked. Measured with mypy 2.4.0 (--strict) and pyright 1.1.414: after an Any-typed retry decorator, get_user revealed as def (*Any, **Any) -> Any in mypy and (...) -> Any in pyright, and a call with a string ID and a misspelt keyword produced no errors. With a ParamSpec decorator the revealed signature stayed (user_id: int, *, fresh: bool = False), and the same call produced two errors in mypy (wrong type, unknown keyword with a "did you mean" hint) and one in pyright. One more detail decided whether the decorated function could be passed to asyncio.create_task: returning Awaitable[T] broke it in both checkers, returning Coroutine[Any, Any, T] did not. This guide writes async decorators whose types survive.

Prerequisites

1. See what an untyped decorator erases

The usual first version of a retry decorator types everything as Any:

import asyncio
import functools
from collections.abc import Callable
from typing import Any


def retry_loose(fn: Callable[..., Any]) -> Callable[..., Any]:
    @functools.wraps(fn)
    async def wrapper(*args: Any, **kwargs: Any) -> Any:
        for attempt in range(3):
            try:
                return await fn(*args, **kwargs)
            except ConnectionError:
                await asyncio.sleep(0.1 * 2 ** attempt)
        return await fn(*args, **kwargs)
    return wrapper


@retry_loose
async def get_user_loose(user_id: int, *, fresh: bool = False) -> dict[str, int]:
    return {"id": user_id}

await get_user_loose("42", fressh=True)   # no error from either checker

Measured: mypy revealed def (*Any, **Any) -> Any; pyright revealed (...) -> Any. functools.wraps copies __name__, __doc__ and __wrapped__ at runtime but tells the type checker nothing. The call with a string ID and a misspelt keyword passed both checkers; at runtime it raises TypeError — but only on the path that calls it, and since TypeError is not a ConnectionError the decorator does not even retry it.

Verify: reveal_type on a decorated function shows the original parameters, not *Any, **Any.

What the checkers saw after each decorator A grid of 3 rows by 3 columns. What the checkers saw after each decorator decorator revealed signature errors on get_user("42", fressh=True) Callable[..., Any] (*Any, **Any) -> Any 0 in mypy, 0 in pyright ParamSpec, returns Awaitable[T] (user_id: int, *, fresh: bool = False) 2 in mypy, 1 in pyright ParamSpec, returns Coroutine[Any, Any, T] same signature same errors, create_task works mypy 2.4.0 --strict and pyright 1.1.414 on Python 3.14.

2. Capture parameters with ParamSpec

ParamSpec captures the whole parameter list of the decorated function, and P.args / P.kwargs thread it through the wrapper. A TypeVar captures the awaited result:

from collections.abc import Awaitable, Callable
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
T = TypeVar("T")


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


@retry(attempts=5)
async def get_user(user_id: int, *, fresh: bool = False) -> dict[str, int]:
    return {"id": user_id}

Measured: both checkers revealed (user_id: int, *, fresh: bool = False) -> Awaitable[dict[str, int]], and await get_user(1) revealed dict[str, int]. The bad call produced Argument 1 to "get_user" has incompatible type "str"; expected "int" and Unexpected keyword argument "fressh" for "get_user"; did you mean "fresh"? in mypy. Pyright stopped at the first problem in the call and reported No parameter named "fressh"; fixing it would have revealed the second. A decorator factory — one that takes options like attempts — needs the extra layer of nesting; the inner deco carries the ParamSpec.

Verify: a call to a decorated function with a wrong argument type is an error at the call site.

3. Return Coroutine so callers can schedule it

The decorator above returns Callable[P, Awaitable[T]]. That is enough for await get_user(1), but not for scheduling it:

task = asyncio.create_task(get_user(1))
# mypy:    Argument 1 to "create_task" has incompatible type "Awaitable[dict[str, int]]";
#          expected "Coroutine[Any, Any, Never]"
# pyright: "Awaitable[dict[str, int]]" is not assignable to "Coroutine[Any, Any, _T@create_task]"

The wrapper is an async def, so it really does return a coroutine; the annotation threw that information away. Declare it:

from collections.abc import Coroutine
from typing import Any


def retry(fn: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, Coroutine[Any, Any, T]]:
    @functools.wraps(fn)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        return await fn(*args, **kwargs)
    return wrapper

task = asyncio.create_task(get_user(1))   # mypy: Task[dict[str, int]]; pyright: Task[dict[str, int]]

Measured: with Coroutine[Any, Any, T] on both sides, create_task type-checked and the task revealed as Task[dict[str, int]] in both checkers. The rule is the one from typing coroutine functions: a decorator's output type is an API, and callers will schedule what it returns.

Verify: asyncio.create_task(decorated(...)) and tg.create_task(decorated(...)) type-check without casts.

How the types flow through an async decorator A flow of 5 stages. How the types flow through an async decorator async def f(user_id: int) -> R the original P captures (user_id: int) T captures R wrapper(*P.args, **P.kwargs) -> T async def returns Callable[P, Coroutine[.., T]] signature kept await f(1) is R create_task(f(1)) is Task[R] P carries the parameters through; T carries the result; Coroutine keeps it schedulable.

4. Use the Python 3.12 generic syntax

On Python 3.12 and later, type parameters can be declared inline, without module-level ParamSpec and TypeVar objects:

def timed[**P, R](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:
        start = time.perf_counter()
        try:
            return await fn(*args, **kwargs)
        finally:
            log.info("%s took %.1f ms", fn.__qualname__, (time.perf_counter() - start) * 1e3)
    return wrapper

Both checkers revealed the same preserved signature for a function decorated this way as for the ParamSpec("P") version. The new syntax scopes P and R to the one function, which avoids accidentally sharing type variables between unrelated decorators in a module. Libraries that still support 3.10 and 3.11 keep the older spelling; the two are interchangeable for the checker.

Verify: on 3.12+, the decorator module has no module-level ParamSpec or TypeVar left that only one function uses.

5. Type decorators that change the signature

Some decorators add or remove parameters — injecting a database connection, adding a timeout keyword. Concatenate describes the change:

from typing import Concatenate


def with_connection[**P, R](
    fn: Callable[Concatenate[Connection, P], Coroutine[Any, Any, R]],
) -> Callable[P, Coroutine[Any, Any, R]]:
    @functools.wraps(fn)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        async with pool.acquire() as conn:
            return await fn(conn, *args, **kwargs)
    return wrapper


@with_connection
async def load_user(conn: Connection, user_id: int) -> User:
    ...

await load_user(42)          # the checker knows conn is supplied by the decorator

Concatenate[Connection, P] says "a Connection first, then whatever else"; the returned callable takes only the rest. Callers see load_user(user_id: int), and passing a connection explicitly is an error. Adding keyword-only parameters, such as an injected timeout=, cannot be expressed with Concatenate; use a callback Protocol with an explicit __call__ for that case, as shown in defining async protocols for dependency injection.

Verify: calling a Concatenate-decorated function with the injected argument supplied by hand is a type error.

How should this async decorator be typed? A decision on What does the decorator do to the signature with 4 outcomes. How should this async decorator be typed? What does the decorator do to the signature? keeps it Callable[P, Coroutine[Any, Any, R]] retry, timing, tracing takes options factory returning the decorator retry(attempts=5) removes a first parameter Concatenate[X, P] injected connection adds keyword parameters callback Protocol timeout= injection Never fall back to Callable[..., Any]: it switches checking off for every decorated function.

Verification

Async decorators preserve types when:

  • reveal_type on decorated functions shows their real parameters and awaited return type.
  • Wrong arguments to decorated functions are errors in mypy and pyright.
  • Decorators return Coroutine[Any, Any, R], so decorated calls can be scheduled with create_task.
  • Signature-changing decorators use Concatenate or a callback Protocol.

Diagnostic Hook: add reveal_type(some_decorated_function) to a scratch file for every decorator in the codebase and run mypy once. Any result containing *Any, **Any or (...) identifies a decorator that is silently disabling type checking for everything it wraps.

Pitfalls & edge cases

  • Callable[..., Any]. Measured: a bad call through it produced zero errors in both checkers.
  • Returning Awaitable[T]. It blocks create_task on decorated functions.
  • functools.wraps is runtime only. It does not inform the type checker.
  • Pyright reports one call error at a time. Fix and re-run before assuming a call is clean.

Frequently Asked Questions

How do I type an async decorator in Python?

Use ParamSpec for the parameters and a TypeVar for the result: def deco(fn: Callable[P, Coroutine[Any, Any, R]]) -> Callable[P, Coroutine[Any, Any, R]], with an inner async def wrapper(args: P.args, *kwargs: P.kwargs) -> R. On Python 3.12+ the same decorator can declare P and R in the new type-parameter list after its name.

Why does my decorated async function accept any arguments?

The decorator is typed with Callable[..., Any], which erases the signature: mypy revealed def (Any, *Any) -> Any and a call with wrong arguments passed both mypy and pyright.

Why can't I pass a decorated coroutine to asyncio.create_task?

The decorator returns Callable[P, Awaitable[T]], and create_task requires a Coroutine. Return Callable[P, Coroutine[Any, Any, T]] instead; both checkers then inferred Task[T].

How do I type a decorator that injects an argument?

Use Concatenate: accept Callable[Concatenate[Injected, P], Coroutine[Any, Any, R]] and return Callable[P, Coroutine[Any, Any, R]], so callers see the signature without the injected parameter.