Python asyncio Tutorial: async, await, and Tasks Explained

A script calls an API 100 times. Each call takes 200 milliseconds, almost all of it spent waiting for the server. Run one after another, the calls take 20 seconds. Run concurrently with asyncio, they finish in roughly the time of the slowest single call. The CPU did the same tiny amount of work in both cases. The second version just stopped waiting in a queue.

This Python asyncio tutorial covers the standard library’s asyncio package: a framework for writing concurrent code with the async and await keywords. Concurrency here means several operations are in progress during the same period, with one thread switching between them. It does not mean they execute at the same instant.

The examples use Python 3.11, because asyncio.TaskGroup and asyncio.timeout() are new in that version. Run python --version first. Most examples use asyncio.sleep() to stand in for a network call, so you can run them with no packages and no network.

My position: asyncio is a tool for waiting efficiently, not a speed setting. Adopt it when your program spends its time waiting on many things at once. In every other case, it adds complexity and gives nothing back.

How asyncio works: one thread and an event loop

Four terms carry the whole model.

• A coroutine function is defined with async def. Calling it does not run it. It returns a coroutine object.
• await runs an awaitable and pauses the current coroutine until the result is ready. While paused, the coroutine gives control back.
• The event loop is the scheduler. It runs one coroutine until that coroutine awaits something slow, then it runs another that is ready.
• A task wraps a coroutine and schedules it on the loop, so it runs concurrently with other tasks.

One thread, three tasks, each waiting 1 second on the network

time  0.0s                                   1.0s
      |                                       |
A     [run]....... waiting on socket .........[run] done
B       [run]..... waiting on socket ...........[run] done
C         [run]... waiting on socket .............[run] done

[run] = Python code executing (microseconds)
....  = task is paused at an await, and the loop is free to run others

Total: about 1 second, not 3

The loop never interrupts a coroutine. A coroutine keeps the thread until it reaches an await that actually has to wait. This is called cooperative scheduling, and it explains both the strength and the main hazard of asyncio. Between two await points, no other task can run, so you need far fewer locks. For the same reason, a coroutine that never awaits blocks every other task.

If you know generators, the pausing will feel familiar. A coroutine suspends at await much as a generator suspends at yield, keeping its local variables until it resumes.

Your first coroutines: sequential versus concurrent

Save this complete script as first.py. It runs the same three operations twice: first one by one, then concurrently.

import asyncio
import time


async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)          # stands in for a network call
    return f'{name} done'


async def sequential() -> list[str]:
    return [
        await fetch('a', 1.0),
        await fetch('b', 1.0),
        await fetch('c', 1.0),
    ]


async def concurrent() -> list[str]:
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(fetch(name, 1.0)) for name in 'abc']
    return [task.result() for task in tasks]


async def main() -> None:
    for runner in (sequential, concurrent):
        start = time.perf_counter()
        results = await runner()
        elapsed = time.perf_counter() - start
        print(f'{runner.__name__}: {len(results)} results in {elapsed:.1f}s')


if __name__ == '__main__':
    asyncio.run(main())

python first.py

sequential: 3 results in 3.0s
concurrent: 3 results in 1.0s

asyncio.run() creates an event loop, runs main() to completion, and closes the loop. Call it once, at the top of your program.

The key lesson sits in sequential(). It is fully async, and it is no faster than ordinary blocking code. await on a coroutine means “run this and wait for it”. It does not start anything in the background. Concurrency appears only when you create tasks.

Coroutine versus task

These two are the most confused pair in asyncio.

Forgetting await entirely is the classic first bug. The call creates a coroutine object, nothing runs, and Python prints a warning when the object is discarded:

RuntimeWarning: coroutine 'fetch' was never awaited

Running tasks: TaskGroup, gather, and create_task

Python offers three ways to run coroutines concurrently. They differ mostly in what happens when something fails.

TaskGroup is an async context manager. The async with block does not exit until every task in the group has finished. No task can outlive the block, which is the idea known as structured concurrency. You can read about the release that introduced it in this summary of what changed in Python 3.11.

Handle group failures with except*, which receives every matching error:

import asyncio


async def fetch(name: str, fail: bool) -> str:
    await asyncio.sleep(0)
    if fail:
        raise ConnectionError(f'{name} unreachable')
    return f'{name} ok'


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(fetch('billing', fail=True))
            group.create_task(fetch('search', fail=False))
            group.create_task(fetch('auth', fail=True))
    except* ConnectionError as errors:
        for error in errors.exceptions:
            print(f'failed: {error}')


if __name__ == '__main__':
    asyncio.run(main())

failed: billing unreachable
failed: auth unreachable

One warning applies to bare asyncio.create_task(). The event loop keeps only a weak reference to a task. If you do not store the task somewhere, the garbage collector may destroy it before it finishes. Keep a reference, or use a TaskGroup, which holds its tasks for you.

Timeouts and cancellation

Every network operation needs a time limit. In Python 3.11, wrap the code in asyncio.timeout():

import asyncio


async def main() -> None:
    try:
        async with asyncio.timeout(1):
            await asyncio.sleep(10)
    except TimeoutError:
        print('gave up after 1 second')


if __name__ == '__main__':
    asyncio.run(main())

gave up after 1 second

A timeout works by cancelling the task. Cancellation raises asyncio.CancelledError inside the coroutine, at the await where it is currently paused. That gives the coroutine a chance to clean up.

import asyncio


async def worker() -> None:
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print('cleaning up')
        raise


async def main() -> None:
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('worker cancelled')


if __name__ == '__main__':
    asyncio.run(main())

cleaning up
worker cancelled

The raise inside the handler is essential. If a coroutine catches CancelledError and does not re-raise it, the cancellation is lost, and timeouts and task groups stop working correctly. CancelledError inherits from BaseException, so a normal except Exception does not swallow it. A bare except: does. For plain cleanup, prefer try and finally, which cannot swallow anything.

The one rule: never block the event loop

Because all tasks share one thread, a single blocking call stops all of them. Blocking calls include time.sleep(), the requests library, most database drivers that are not async, and any long calculation.

import asyncio
import time


async def handler() -> None:
    time.sleep(1)                        # WRONG: blocks the whole loop


async def handler_fixed() -> None:
    await asyncio.to_thread(time.sleep, 1)   # RIGHT: runs in a worker thread


async def timed(func) -> None:
    start = time.perf_counter()
    async with asyncio.TaskGroup() as group:
        for _ in range(3):
            group.create_task(func())
    print(f'{func.__name__}: {time.perf_counter() - start:.1f}s')


async def main() -> None:
    await timed(handler)
    await timed(handler_fixed)


if __name__ == '__main__':
    asyncio.run(main())

handler: 3.0s
handler_fixed: 1.0s

The wrong version is “async” in name only. Three tasks take three seconds, because each one holds the thread while it sleeps. asyncio.to_thread() moves a blocking function into a thread pool and gives you something to await. It is the right bridge for blocking libraries you cannot replace.

Find blocking calls with debug mode

Here is the technique I wish every asyncio guide led with. The loop can tell you when something blocked it. Enable debug mode:

asyncio.run(main(), debug=True)

You can also set the environment variable PYTHONASYNCIODEBUG=1. In debug mode, asyncio logs a warning whenever a single step of a task holds the loop for more than 100 milliseconds:

Executing <Task finished name='Task-2' coro=<handler() done, defined at block.py:5> result=None created at ...> took 1.001 seconds

The message names the coroutine and the file. Run your test suite once with debug mode on, and every blocking call on a tested path reveals itself. The threshold is adjustable through loop.slow_callback_duration. I treat any step above 50 milliseconds in a web service as a bug, because during that time no other request makes progress.

Limit concurrency with a semaphore

Starting 10,000 requests at once will exhaust connection limits, or get you rate-limited. An asyncio.Semaphore caps how many tasks run a section at the same time.

import asyncio
import time


async def download(item: int, limit: asyncio.Semaphore) -> int:
    async with limit:
        await asyncio.sleep(1)           # stands in for a network call
        return item


async def main() -> None:
    limit = asyncio.Semaphore(2)
    start = time.perf_counter()
    async with asyncio.TaskGroup() as group:
        for item in range(6):
            group.create_task(download(item, limit))
    print(f'6 downloads, 2 at a time: {time.perf_counter() - start:.1f}s')


if __name__ == '__main__':
    asyncio.run(main())

6 downloads, 2 at a time: 3.0s

Six one-second downloads, two at a time, take three seconds. Choose the limit from what the remote service tolerates, not from what your machine can start.

A real example: concurrent HTTP requests with httpx

Real network calls need an async HTTP client. The popular requests library is blocking, so use httpx or aiohttp. Install httpx in a virtual environment:

python -m venv .venv
source .venv/bin/activate          # macOS and Linux
.venv\Scripts\Activate.ps1         # Windows PowerShell
python -m pip install httpx

import asyncio

import httpx

URLS = ['https://example.com', 'https://example.org', 'https://example.net']


async def status(client: httpx.AsyncClient, url: str) -> tuple[str, int]:
    response = await client.get(url)
    return url, response.status_code


async def main() -> None:
    async with httpx.AsyncClient(timeout=10.0) as client:
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(status(client, url)) for url in URLS]
    for task in tasks:
        print(*task.result())


if __name__ == '__main__':
    asyncio.run(main())

https://example.com 200
https://example.org 200
https://example.net 200

Create one client and share it between tasks. A client holds a connection pool, so a new client per request throws away the reuse that makes concurrent requests cheap.

Decorators need an async wrapper

A decorator written for normal functions breaks quietly on coroutines. A synchronous wrapper calls the coroutine function, gets a coroutine object back at once, and returns it. A timing decorator then measures microseconds, and a retry decorator never sees the exception. The wrapper must itself be async def and must await the call:

import functools
import time


def timed(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return await func(*args, **kwargs)
        finally:
            print(f'{func.__name__} took {time.perf_counter() - start:.2f}s')
    return wrapper

The myth: asyncio makes everything faster

The most common misunderstanding is that adding async speeds up code. It does not speed up any single operation. It removes idle waiting by overlapping operations that wait. If nothing waits, nothing overlaps.

This illustrative script runs a CPU-bound function three times, sequentially and then as tasks:

import asyncio
import time


async def crunch() -> int:
    return sum(n * n for n in range(5_000_000))


async def main() -> None:
    start = time.perf_counter()
    for _ in range(3):
        await crunch()
    print(f'sequential: {time.perf_counter() - start:.2f}s')

    start = time.perf_counter()
    async with asyncio.TaskGroup() as group:
        for _ in range(3):
            group.create_task(crunch())
    print(f'as tasks:   {time.perf_counter() - start:.2f}s')


if __name__ == '__main__':
    asyncio.run(main())

On my machine, both lines print nearly the same time. The function contains no await, so each task runs from start to finish before the next begins. Worse, in a server those calculations would freeze every other request while they ran. CPU-bound work belongs in separate processes, not in tasks.

How real systems use asyncio

• Async web frameworks. Servers built on ASGI frameworks handle each request as a task. One process can hold thousands of open connections, because idle connections cost almost nothing.
• Fan-out to other services. A request handler calls several backends concurrently inside a TaskGroup, under a single timeout, and combines the results.
• Long-lived connections. WebSocket servers, chat backends, and streaming clients keep many mostly idle connections open at once.
• Bounded crawlers and importers. Batch jobs fetch thousands of URLs with a semaphore that limits concurrency and a shared client.
• Blocking work pushed aside. Calls to blocking libraries go through asyncio.to_thread(), and heavy computation goes to a process pool or a separate worker service.

We once hit a bug when an async API service showed latency spikes on every endpoint at once, including the health check. One rarely used handler called a blocking PDF library that took about two seconds. While it ran, the event loop served nobody. The slow endpoint looked fine in its own metrics, and the damage showed up everywhere else. Asyncio’s debug mode named the coroutine in the first test run, and moving that call to asyncio.to_thread() ended the spikes.

Deciding whether to use asyncio: a decision framework

1. Does the program mostly wait on networks or sockets? If not, stop here and write synchronous code.
2. Does it wait on many things at the same time? Hundreds or thousands of simultaneous waits are where asyncio wins. For a handful, a small thread pool is simpler.
3. Do async libraries exist for everything you call? You need an async HTTP client and an async database driver. One blocking dependency in the hot path undermines the design.
4. Is the team ready for the split? Async functions can only be awaited from async functions, so the style spreads through the codebase.
5. Is there CPU-heavy work? Plan a process pool or separate workers for it from the start.

When NOT to use asyncio

• CPU-bound programs. Data crunching, image processing, and numeric simulations gain nothing, because they never wait. They need multiple processes or compiled libraries.
• Simple scripts and command-line tools. A script that makes five requests in a row finishes in a second either way. Async adds ceremony and harder tracebacks.
• Codebases tied to blocking libraries. If your database driver or SDK is synchronous and has no async alternative, wrapping every call in a thread gives you threads with extra steps.

Common mistakes

• Forgetting await. The coroutine never runs, and you get a RuntimeWarning that is easy to miss in logs. Data silently fails to save.
• Awaiting in a loop and expecting concurrency. for url in urls: await fetch(url) is sequential. Total time is the sum of all calls.
• Calling blocking functions. time.sleep(), requests.get(), or a slow calculation freezes every task. Latency rises on unrelated endpoints.
• Swallowing CancelledError. A handler that catches it without re-raising makes timeouts and shutdown hang.
• Dropping the reference to a task. A task created with asyncio.create_task() and not stored can be garbage-collected mid-flight, and its exception is never seen.
• Unbounded fan-out. Creating a task per item for a million items exhausts memory and sockets. Use a semaphore or process the input in batches.

Key takeaways

• asyncio runs many waiting operations on one thread by switching at await points.
• Awaiting a coroutine runs it to completion. Only tasks run concurrently.
• On Python 3.11, use asyncio.TaskGroup to run tasks and except* to handle their errors.
• Put a timeout on every network operation with asyncio.timeout().
• Always re-raise CancelledError after cleanup.
• Never block the loop. Use asyncio.to_thread() for blocking calls, and debug mode to find them.
• Limit concurrency with a semaphore, and share one HTTP client.

FAQ

What is asyncio in Python?

asyncio is a standard library package for writing concurrent code with async and await. It runs an event loop that switches between coroutines whenever one is waiting on I/O.

What is the difference between a coroutine and a task?

A coroutine is the object returned by calling an async def function, and it runs only when awaited. A task wraps a coroutine and schedules it on the event loop, so it runs concurrently with other tasks.

What does await do in Python?

await pauses the current coroutine until the awaited operation finishes, and lets the event loop run other tasks in the meantime. It can be used only inside an async def function.

Should I use asyncio.gather or TaskGroup?

On Python 3.11, prefer TaskGroup. It cancels the remaining tasks when one fails and reports every error. Use gather() when you support older versions or need partial results with return_exceptions=True.

Is asyncio faster than threads?

For thousands of simultaneous network waits, asyncio usually uses less memory and scales further. For a few dozen blocking calls, a thread pool performs similarly and is simpler. Neither speeds up CPU-bound work.

Await what waits, and move the rest off the loop

asyncio rewards one habit above all: knowing which calls wait and which calls work. Waiting belongs behind await, grouped into tasks with a timeout. Work that blocks belongs in a thread or a process. Keep those two apart, and the event loop stays responsive.

Rule of thumb: async code is only as concurrent as its slowest stretch between two awaits.