Python asyncio.gather: Run Multiple Coroutines Concurrently

When you need to run several coroutines concurrently in Python, asyncio.gather is the standard tool. It accepts one or more awaitables, schedules them on the current event loop, and returns a future that completes with all results. The core pattern looks like this:

import asyncio

async def fetch_data(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} finished"

async def main():
    results = await asyncio.gather(
        fetch_data("first", 0.5),
        fetch_data("second", 1.0),
        fetch_data("third", 0.2),
    )
    print(results)

asyncio.run(main())

The coroutines passed to gather start running when the event loop next gets control, not sequentially one after another. The await suspends main until every coroutine completes, and the printed list contains all three return values.

The basic gather pattern

asyncio.gather accepts any number of awaitables as positional arguments. Each argument can be a coroutine object, an existing task, or another awaitable such as a future. The function returns a future that resolves to a list of results, one entry per input, in the same order as the inputs were passed.

The await is where the calling coroutine waits for the whole batch. If you never await the returned future, your program has no way to observe the completed results or exceptions from that batch, so always await it when you care about the outcome.

How gather schedules the coroutines

When you call asyncio.gather, each coroutine argument is wrapped in a task internally. Existing tasks and futures are used as they are. The tasks are scheduled on the event loop and can interleave at each await point. The order in which they finish is not guaranteed, but the order of the returned list is.

Because the scheduling happens through the event loop, the concurrency is cooperative. Each coroutine runs until it hits an await, then control returns to the loop, which can advance another coroutine. This is why asyncio.sleep in the example produces overlapping execution: the three calls do not run one after the other; they progress together.

Result ordering

The list returned by gather preserves the order of the inputs, not the order of completion. This is useful when you need to correlate results with the original calls.

async def main():
    results = await asyncio.gather(
        fetch_data("slow", 1.0),
        fetch_data("fast", 0.1),
    )
    # results[0] corresponds to "slow", results[1] to "fast"

The slow coroutine finishes last, yet its result still appears at index 0. If you need results in completion order, asyncio.as_completed is the better API, because it emits each awaitable as soon as it finishes and you can process the result immediately.

Handling exceptions with return_exceptions

By default, if any coroutine raises an exception, gather propagates the first exception from the await as soon as that task fails. The remaining coroutines are not cancelled; they continue running, but their results are no longer available in the normal result list. To collect exceptions as results instead, set return_exceptions=True.

async def fail() -> str:
    raise RuntimeError("boom")

async def main():
    results = await asyncio.gather(
        fetch_data("ok", 0.2),
        fail(),
        return_exceptions=True,
    )
    for item in results:
        if isinstance(item, Exception):
            print("failed:", item)
        else:
            print("ok:", item)

With return_exceptions=True, the exception object is placed in the result list at the position of the failed coroutine, and gather does not raise. This is useful when you want to run a batch of independent operations and handle failures per item rather than aborting the whole batch.

Cancellation behavior

If the task awaiting gather is cancelled, gather cancels all the child tasks it created. The cancellation propagates up through the await: the caller receives CancelledError, and return_exceptions=True does not turn this cancellation into a normal result. Child coroutines get a chance to run their finally blocks and clean up resources, but they cannot prevent the outer cancellation from completing.

If you need to react to a child task being cancelled by outside code, check for asyncio.CancelledError explicitly; in modern Python it is a BaseException, not an Exception, so an except Exception handler will not catch it.

gather vs create_task, wait, and as_completed

The choice depends on what you need from the batch. gather is the most direct when you want a single await point and a complete result list. create_task is appropriate when you need to store tasks and await them later, or attach callbacks. wait gives explicit control over timeouts and lets you inspect which tasks are still pending. as_completed is the right tool when the first finished result should be handled immediately.

When gather is the right choice

Use gather when you have a fixed set of coroutines, want all results collected in input order, and a single await point is enough. That covers common batch operations: fetching several resources, running independent validation steps, or waiting for multiple background jobs before continuing.

Do not use gather when you need incremental results, because the list is available only after everything completes. Do not use it when you need a timeout on the whole operation, because gather has no timeout parameter; asyncio.wait with a timeout argument is more direct for that case.

Concurrency and performance considerations

The concurrency gain comes from awaiting inside the coroutines, not from gather itself. gather does not make blocking code non-blocking. If a coroutine calls a blocking function such as time.sleep or a synchronous database driver, it blocks the entire event loop, and the other coroutines cannot make progress until it returns.

For blocking work, offload it to a thread with asyncio.to_thread (Python 3.9+) or loop.run_in_executor.

async def main():
    results = await asyncio.gather(
        asyncio.to_thread(blocking_io_call, "first"),
        asyncio.to_thread(blocking_io_call, "second"),
    )

Another constraint: pass each coroutine object to exactly one gather call. If the same object is passed to two gather calls, the second call fails because the coroutine is already scheduled. If you need to share work across multiple consumers, create the task once with asyncio.create_task and pass the task object to each gather call.