How to Use Python Async Await With Examples You Can Run

In Python, async def declares a coroutine function, and await pauses that coroutine until the thing it's waiting on is done, so the event loop can run other coroutines in the meantime. Calling a coroutine function doesn't run it. It returns a coroutine object that only executes when you pass it to asyncio.run() or await it from inside another coroutine.

That's the whole model. The rest of asyncio is about scheduling several coroutines so their waits overlap (create_task, gather, TaskGroup), capping how long you'll wait (asyncio.timeout, wait_for), and moving blocking calls out of the loop's way (to_thread). Every example below uses the standard library only.

The smallest python async await example#

Start with one coroutine, run twice in a row, then run both at once. asyncio.sleep stands in for a network call or any other wait.

import asyncio
import time

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

async def main():
    start = time.perf_counter()
    print(await fetch('a', 0.2))
    print(await fetch('b', 0.3))
    print(f'sequential: {time.perf_counter() - start:.1f}s')

    start = time.perf_counter()
    results = await asyncio.gather(fetch('a', 0.2), fetch('b', 0.3))
    print(results)
    print(f'concurrent: {time.perf_counter() - start:.1f}s')

asyncio.run(main())

The first block awaits each call before starting the next, so the total is the sum of the delays. The second block hands both coroutines to asyncio.gather, which schedules them as tasks and runs them concurrently, so the total is close to the slowest single call. That gap is the entire reason async exists: while one coroutine waits, another gets to make progress.

How python async await works#

Three pieces are involved.

The coroutine function. Anything declared with async def. Calling it builds a coroutine object and nothing else. The body hasn't started.

The await expression. await x is only legal inside an async def body. It says: suspend me here, give control back to the event loop, and resume me when x has a result. The thing after await has to be an awaitable. The docs name three kinds: coroutines, Tasks, and Futures. You'll write coroutines and create Tasks. Futures are the low-level piece that lets callback-based code plug into await, and application code almost never needs to create one.

The event loop. asyncio.run(main()) starts a loop, runs main() to completion, and shuts the loop down. It's the normal entry point for a script. Inside main, you never see the loop directly. You just await things.

The single most common beginner bug follows from the first point:

import asyncio

async def nested():
    return 42

async def main():
    nested()               # builds a coroutine object and drops it, never runs
    print(await nested())  # runs it, prints 42

asyncio.run(main())

The bare nested() call produces a RuntimeWarning and the body never executes. If a function 'does nothing' when you call it, the missing await is the first thing to check.

Running multiple tasks with create_task, gather and TaskGroup#

await on its own is sequential. To overlap work you need Tasks, and there are three ways to create them.

asyncio.create_task wraps a coroutine in a Task and schedules it to run soon. You await the Task later to collect its result:

async def main():
    task1 = asyncio.create_task(say_after(1, 'hello'))
    task2 = asyncio.create_task(say_after(2, 'world'))
    await task1
    await task2

Both tasks start as soon as they're created, so this finishes in about two seconds instead of three.

asyncio.TaskGroup (Python 3.11 and later) does the same with a context manager, and the await is implicit when the block exits:

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(say_after(1, 'hello'))
        task2 = tg.create_task(say_after(2, 'world'))
    # both tasks are done here

The practical difference is error handling. If one task in a TaskGroup raises, the group cancels the remaining tasks. gather doesn't.

asyncio.gather takes any number of awaitables, runs them concurrently, and returns a list of results in the same order you passed them in. It's the shortest way to fan out a list of calls. Three rules matter:

• With the default return_exceptions=False, the first exception propagates to the caller immediately, and the other awaitables keep running in the background.
• With return_exceptions=True, exceptions are placed in the result list alongside normal values, so you get one entry per input.
• If the gather call itself is cancelled, every awaitable still in progress is cancelled too.

Here's the return_exceptions=True form, which is what I'd use when a partial failure shouldn't abort the batch:

import asyncio

async def job(n):
    await asyncio.sleep(0.05 * n)
    if n == 2:
        raise ValueError(f'job {n} failed')
    return n * 10

async def main():
    results = await asyncio.gather(
        *(job(n) for n in range(4)),
        return_exceptions=True,
    )
    for n, r in enumerate(results):
        status = 'error' if isinstance(r, BaseException) else 'ok'
        print(n, status, r)

asyncio.run(main())

Each slot in the list lines up with its input, which makes it easy to report which job failed. If you'd rather stop everything on the first failure, use a TaskGroup instead.

Adding a timeout to an await#

Two tools cover most cases. asyncio.timeout(delay) (3.11 and later) is an async context manager that cancels whatever is running inside it once delay seconds pass. asyncio.wait_for(aw, timeout) wraps a single awaitable.

import asyncio

async def slow():
    await asyncio.sleep(5)
    return 'finished'

async def main():
    try:
        async with asyncio.timeout(0.2):
            await slow()
    except TimeoutError:
        print('asyncio.timeout: gave up after 0.2s, moving on')

    try:
        result = await asyncio.wait_for(slow(), timeout=0.2)
        print(result)
    except TimeoutError:
        print('wait_for: also timed out, slow() was cancelled')

asyncio.run(main())

Details that bite people:

• The TimeoutError from asyncio.timeout can only be caught outside the async with block. The context manager is what converts the internal CancelledError into TimeoutError, so a try inside the block won't see it.
• wait_for cancels the awaitable when the timeout fires, then waits for that cancellation to finish, so the total wait can be slightly longer than timeout. Wrap the awaitable in asyncio.shield() if it must not be cancelled.
• Pass None as the delay when you don't know it yet. The context manager returned by asyncio.timeout has a reschedule(when) method that accepts an absolute loop time, and expired() tells you afterward whether the deadline hit. asyncio.timeout_at(when) takes an absolute deadline directly.
• Timeout context managers nest safely, so an outer deadline around a block of inner ones works as you'd expect.

Taking results as they finish#

gather gives you everything at the end. When you want to react to the first result, or process results in completion order, use asyncio.wait or asyncio.as_completed.

asyncio.wait takes an iterable of Tasks or Futures (coroutine objects aren't accepted since 3.11, so wrap them with create_task first) and returns two sets:

tasks = [asyncio.create_task(fetch(u)) for u in urls]
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

return_when can be FIRST_COMPLETED, FIRST_EXCEPTION, or the default ALL_COMPLETED. The iterable must not be empty. With a timeout, wait doesn't raise. Unfinished tasks simply come back in pending, still running. Cancel them yourself if you don't need them.

asyncio.as_completed yields results in the order they finish. On 3.13 and later you can iterate it with async for, and it yields the original task objects, so you can tell which one finished:

tasks = [asyncio.create_task(fetch(u)) for u in urls]
async for task in asyncio.as_completed(tasks):
    result = await task

On older versions, use a plain for loop. Each iteration gives a new coroutine to await, and you lose the link to the original task. Either way, as_completed doesn't cancel the remaining tasks if a timeout fires or the iteration is cancelled.

Async await vs threading: when async actually helps#

Async pays off when your program spends most of its time waiting on I/O, like network requests, database queries, or subprocess output, and when the libraries you call offer awaitable APIs. One event loop can keep many such waits in flight because a suspended coroutine costs very little.

It does not help CPU-bound code. The loop runs one coroutine at a time, and a coroutine that crunches numbers for two seconds without awaiting blocks every other coroutine for those two seconds. The same goes for blocking calls such as time.sleep or a synchronous HTTP client: they stall the whole loop.

For a blocking function you can't rewrite, asyncio.to_thread runs it in a separate thread and gives you an awaitable for the result. The docs describe it as intended for I/O-bound functions that would otherwise block the loop:

import asyncio
import time

def blocking_io():
    time.sleep(1)  # any blocking call: file I/O, a sync client, etc.
    return 'done'

async def main():
    result, _ = await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1),
    )
    print(result)

asyncio.run(main())

Context variables from the loop thread are propagated into the worker thread, which matters if you rely on contextvars for request-scoped state.

My rule of thumb: if the work is 'wait for many things', use async. If the work is 'wait for one thing' in an otherwise synchronous program, plain threads or a synchronous library are usually simpler. If the work is 'compute', async won't speed it up at all.

Common mistakes with async and await#

• Forgetting await. fetch() alone creates a coroutine object and discards it. You'll see a RuntimeWarning and nothing will run.
• Using await outside an async def. That's a SyntaxError. At module level, the fix is to put the code in async def main() and call asyncio.run(main()) once at the bottom of the file.
• Awaiting in a loop and expecting concurrency. for c in coros: await c runs them one after another. Use gather, create_task, or a TaskGroup to overlap them.
• Calling time.sleep inside a coroutine. It blocks the loop. Use await asyncio.sleep(...), or push the blocking call into asyncio.to_thread.
• Catching TimeoutError inside the asyncio.timeout block. It's raised on exit from the block, so the try has to wrap the async with.
• Assuming gather stops the others on failure. It doesn't. The remaining awaitables keep running after the first exception propagates. Use a TaskGroup for all-or-nothing behavior.
• Passing coroutine objects to asyncio.wait. Forbidden since 3.11. Wrap them with create_task first.
• Expecting gather results in completion order. They come back in input order. Use as_completed when finish order matters.

Quick reference#

Checklist before you ship an async script:

1. Every coroutine call is awaited, scheduled with create_task, or passed to gather or a TaskGroup.
2. No time.sleep or other blocking call inside a coroutine without to_thread.
3. Every external wait sits under a timeout.
4. You've decided what happens to the other tasks when one fails: gather keeps going, TaskGroup cancels.
5. One asyncio.run() at the entry point, nothing more.

The full API, including the details skipped here, is in the asyncio coroutines and tasks reference. The asyncio overview is the right starting page for the rest of the library, and the event loop reference covers the loop clock that timeout_at and reschedule use.

Quick answers