Running Async Code in Typer and Click

You have an async library you want to use from a command — httpx.AsyncClient, an async database driver, an SDK that only offers coroutines — or you want to run a few hundred network calls concurrently with asyncio. You write async def sync_repos(...), decorate it with @app.command(), run it, and get a warning that the coroutine was never awaited while the command silently does nothing. Typer and Click call command functions synchronously; neither runs an event loop for you. This guide shows the simple, dependency-free way to bridge the gap, a small decorator that removes the boilerplate without breaking Typer's signature inspection, how to structure concurrent work inside the command with TaskGroup, and how to test async commands. It is part of the concurrency and async topic.

Prerequisites

• Python 3.11+ (for asyncio.TaskGroup and asyncio.timeout).
• Typer 0.12+ or Click 8.1+.
• httpx for the examples, which use its AsyncClient.

The bridge: asyncio.run at the edge

The whole technique fits in one sentence: keep the command function synchronous, and have it call asyncio.run() on an async function that does the work.

import asyncio

import httpx
import typer

app = typer.Typer()


async def fetch_titles(urls: list[str]) -> dict[str, int]:
    async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client:
        responses = await asyncio.gather(*(client.get(u) for u in urls))
    return {str(r.url): len(r.content) for r in responses}


@app.command()
def sizes(urls: list[str]) -> None:
    """Print the size of each page."""
    for url, size in asyncio.run(fetch_titles(urls)).items():
        typer.echo(f"{size:>8}  {url}")


if __name__ == "__main__":
    app()

asyncio.run() creates a fresh event loop, runs the coroutine to completion, cancels anything left over, shuts down async generators and the default executor, and closes the loop. Call it once per command invocation, at the outermost point. Calling it from inside already-running async code raises RuntimeError: asyncio.run() cannot be called from a running event loop — a sign that the async boundary is in the wrong place.

The recipe: a decorator for async commands

Writing asyncio.run(...) in every command is repetitive, and it tempts people to split each command into a sync wrapper and an async twin. A decorator lets you write async def commands directly:

# src/mytool/aio.py
from __future__ import annotations

import asyncio
import functools
from collections.abc import Callable, Coroutine
from typing import Any, ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def run_async(fn: Callable[P, Coroutine[Any, Any, R]]) -> Callable[P, R]:
    """Let Typer/Click call an async function as a normal command."""

    @functools.wraps(fn)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        return asyncio.run(fn(*args, **kwargs))

    return wrapper

functools.wraps copies the name, docstring and annotations and sets __wrapped__. Typer uses inspect.signature(), which follows __wrapped__, so it sees the original parameters and builds the same options and arguments as for a normal function. Order matters: @app.command() must be the outer decorator so that it receives the wrapped, synchronous function.

# src/mytool/cli.py
import asyncio

import httpx
import typer

from mytool.aio import run_async

app = typer.Typer()


@app.callback()
def main() -> None:
    """Repository tools."""


@app.command()
@run_async
async def check(
    repos: list[str],
    jobs: int = typer.Option(8, "--jobs", "-j", min=1),
    timeout: float = typer.Option(60.0, help="Overall time limit in seconds."),
) -> None:
    """Report whether each GitHub repository exists."""
    sem = asyncio.Semaphore(jobs)
    async with httpx.AsyncClient(base_url="https://api.github.com", timeout=10.0) as client:

        async def exists(repo: str) -> tuple[str, bool]:
            async with sem:
                r = await client.get(f"/repos/{repo}")
                return repo, r.status_code == 200

        try:
            async with asyncio.timeout(timeout):
                async with asyncio.TaskGroup() as tg:
                    tasks = [tg.create_task(exists(r)) for r in repos]
        except TimeoutError:
            typer.echo(f"error: gave up after {timeout:g}s", err=True)
            raise typer.Exit(124)

    missing = [repo for repo, ok in (t.result() for t in tasks) if not ok]
    for repo in missing:
        typer.echo(f"✗ {repo}", err=True)
    typer.echo(f"{len(repos) - len(missing)}/{len(repos)} exist", err=True)
    raise typer.Exit(1 if missing else 0)


if __name__ == "__main__":
    app()

typer.Exit raised inside the coroutine propagates out of asyncio.run() like any other exception, so exit codes work exactly as in synchronous commands. Click users can apply the same decorator beneath @click.command(); if you would rather have the framework handle it, the asyncclick fork supports async def commands natively at the cost of an extra dependency.

TaskGroup or gather?

Both run coroutines concurrently; they differ when something fails. asyncio.TaskGroup cancels the remaining tasks as soon as one raises, waits for them to finish cancelling, and raises every error together in an ExceptionGroup. asyncio.gather lets the others keep running, raises only the first exception, and — unless you pass return_exceptions=True — leaves the rest of the results unreachable.

In a CLI, the choice follows the command's semantics. If one failure means the whole operation is meaningless (a deploy of several services that must all succeed), use a TaskGroup and let it fail fast. If each item is independent (checking 300 URLs), you want every result: catch errors inside each task and return them as values, or use gather(..., return_exceptions=True) and sort successes from failures afterwards. The value-returning approach is the one the concurrency topic recommends for batch commands.

Async setup shared across commands

Sometimes a group callback needs async setup — opening a connection pool every subcommand will use. Callbacks run synchronously too, and each asyncio.run() gets its own loop, so an async client created in one asyncio.run() cannot be used in another. Instead, store a factory on the context and create the async resource inside the command's own loop:

from collections.abc import Callable

import httpx
import typer


@app.callback()
def main(ctx: typer.Context, api_url: str = typer.Option("https://api.example.com")) -> None:
    ctx.obj: Callable[[], httpx.AsyncClient] = lambda: httpx.AsyncClient(base_url=api_url)


@app.command()
@run_async
async def ping(ctx: typer.Context) -> None:
    async with ctx.obj() as client:
        r = await client.get("/health")
        typer.echo(r.status_code)

The same rule applies to anything bound to an event loop: locks, queues, semaphores and clients must be created inside the loop that uses them.

UX considerations

• Bound everything. A semaphore sized by --jobs and an overall asyncio.timeout mean an async command never floods a server or runs forever.
• Mixing in blocking calls freezes everything. A synchronous requests.get() or time.sleep() inside a coroutine blocks the whole loop, so every other task stalls and progress bars stop. Use async libraries, or push unavoidable blocking calls to a thread with await asyncio.to_thread(fn, *args).
• Ctrl+C should just work. Since Python 3.11, asyncio.run() handles the first SIGINT by cancelling the main task, so finally blocks and async with exits run. Make sure your code does not swallow CancelledError; cancelling async tasks on Ctrl+C covers the details.
• Keep startup fast. Importing asyncio and async libraries costs time on every invocation, including --help. Import heavy async dependencies inside the command body if startup matters; see lazy-loading subcommands for faster startup.

Testing the behaviour

Because the decorated command is synchronous from the outside, CliRunner tests it exactly like any other command. Swap in a MockTransport so no network is touched — httpx.MockTransport works with AsyncClient too:

# tests/test_check.py
import httpx
import pytest
from typer.testing import CliRunner

from mytool import cli

runner = CliRunner()
REAL_CLIENT = httpx.AsyncClient


@pytest.fixture
def fake_github(monkeypatch):
    existing = {"/repos/psf/requests", "/repos/pallets/click"}

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200 if request.url.path in existing else 404)

    def client(**kwargs):
        return REAL_CLIENT(transport=httpx.MockTransport(handler), **kwargs)

    monkeypatch.setattr(cli.httpx, "AsyncClient", client)


def test_all_exist(fake_github):
    result = runner.invoke(cli.app, ["check", "psf/requests", "pallets/click"])
    assert result.exit_code == 0
    assert "2/2 exist" in result.output


def test_missing_repo_fails(fake_github):
    result = runner.invoke(cli.app, ["check", "psf/requests", "nobody/nothing"])
    assert result.exit_code == 1
    assert "nobody/nothing" in result.output


def test_options_survive_the_decorator():
    result = runner.invoke(cli.app, ["check", "--help"])
    assert "--jobs" in result.output and "--timeout" in result.output

The last test is small but valuable: it fails immediately if someone swaps the decorator order or drops functools.wraps, which would make Typer see (*args, **kwargs) and lose every option. For unit-testing the async helpers directly, pytest-asyncio or anyio's pytest plugin let you write async def test_... functions.

Conclusion

Typer and Click do not run event loops, and they do not need to. Keep commands synchronous from the framework's point of view, cross into async with one asyncio.run() per invocation — wrapped in a signature-preserving decorator if you like — and do everything concurrent inside that boundary with TaskGroup, semaphores and timeouts. Create loop-bound resources inside the loop, keep blocking calls off it, and test the whole thing through CliRunner like any other command.

Frequently asked questions

Related

• Up: Concurrency and async in Python CLIs
• Cancelling async tasks on Ctrl+C
• Rate-limiting concurrent requests in CLIs
• Typer callback functions explained
• Building an API client CLI with httpx