Fixing Silently Ignored Exceptions in Python asyncio.gather Calls

You fire off five concurrent tasks with asyncio.gather, and three of them fail. Your program prints no traceback, returns partial results, and continues as if nothing happened. This is not a bug in asyncio — it is a deliberate design choice that bites almost every developer who uses it for the first time.

Understanding exactly when and why exceptions disappear, and knowing the right patterns to surface them, will save you hours of confused debugging.

What you'll learn

• Why asyncio.gather can silently drop exceptions under certain conditions
• How the return_exceptions parameter changes gather's behavior
• Patterns for inspecting per-task results and exceptions together
• How to use asyncio.TaskGroup (Python 3.11+) as a stricter alternative
• Common mistakes that make exception handling in gather worse, not better

Prerequisites

You should be comfortable writing basic async functions with async def and await. The examples use Python 3.10 for gather patterns, and Python 3.11 for TaskGroup. Nothing else is required beyond the standard library.

How asyncio.gather Handles Exceptions by Default

By default, asyncio.gather runs all awaitables concurrently and returns a list of results in the same order as the input. When one of the coroutines raises an exception, gather cancels nothing — the other tasks keep running — but the exception is re-raised to the caller the moment you await the gather call.

Here is a minimal example that shows the default behavior:

import asyncio

async def good_task(n):
    await asyncio.sleep(0.1)
    return n * 2

async def bad_task():
    await asyncio.sleep(0.05)
    raise ValueError("something went wrong")

async def main():
    results = await asyncio.gather(
        good_task(1),
        bad_task(),
        good_task(3),
    )
    print(results)  # never reached

asyncio.run(main())

When bad_task raises, the await asyncio.gather(...) line re-raises ValueError. The two good_task coroutines still run to completion in the background, but their return values are discarded. You only see one exception, even if several tasks failed.

This is where the silent loss begins: you lose results from tasks that succeeded, and if more than one task failed, you only find out about the first exception.

The return_exceptions Parameter

Passing return_exceptions=True changes everything. Instead of re-raising the first exception, gather collects both results and exceptions into the return list. Exceptions are returned as objects, not raised.

import asyncio

async def good_task(n):
    await asyncio.sleep(0.1)
    return n * 2

async def bad_task():
    raise ValueError("something went wrong")

async def main():
    results = await asyncio.gather(
        good_task(1),
        bad_task(),
        good_task(3),
        return_exceptions=True,
    )
    for i, result in enumerate(results):
        if isinstance(result, BaseException):
            print(f"Task {i} failed: {result}")
        else:
            print(f"Task {i} succeeded: {result}")

asyncio.run(main())

Now you get every outcome. The list might look like [2, ValueError('something went wrong'), 6]. You can loop through it, separate successes from failures, log them, and decide whether the overall operation should proceed or abort.

One important detail: return_exceptions=True catches instances of Exception and BaseException, which includes asyncio.CancelledError. Be careful — you probably do not want to silently swallow cancellation.

Why Exceptions Still Feel Silent

Even with return_exceptions=False (the default), exceptions can feel invisible in certain patterns. The most common cause is creating tasks with asyncio.create_task and then passing them to gather without ever awaiting the gather result or wrapping it in a try/except.

import asyncio

async def flaky():
    raise RuntimeError("I failed")

async def main():
    task = asyncio.create_task(flaky())
    # We never await the task or gather it.
    # Python will print a warning, but only when the task is garbage-collected.
    await asyncio.sleep(1)

asyncio.run(main())

Python does print a warning like "Task exception was never retrieved" when the task is garbage-collected, but by then you may have already served a broken response or written bad data. The warning is easy to miss in noisy logs.

The fix is simple: always await every task you create, either directly or through gather.

Handling Exceptions Per-Task with return_exceptions

A robust pattern is to separate the inspection logic into its own function so you are not repeating isinstance checks everywhere.

import asyncio
from typing import Any

def extract_results(outcomes: list[Any]) -> tuple[list[Any], list[BaseException]]:
    successes = []
    failures = []
    for item in outcomes:
        if isinstance(item, BaseException):
            failures.append(item)
        else:
            successes.append(item)
    return successes, failures

async def fetch_data(url: str) -> dict:
    # Simulated fetch
    if "bad" in url:
        raise ConnectionError(f"Could not connect to {url}")
    return {"url": url, "data": "ok"}

async def main():
    urls = [
        "https://good.example.com",
        "https://bad.example.com",
        "https://good2.example.com",
    ]
    outcomes = await asyncio.gather(
        *[fetch_data(u) for u in urls],
        return_exceptions=True,
    )
    successes, failures = extract_results(outcomes)
    print(f"Succeeded: {len(successes)}, Failed: {len(failures)}")
    for exc in failures:
        print(f"Error: {exc}")

asyncio.run(main())

This keeps your main flow readable and gives you a clean place to add metrics, retries, or alerting.

Preserving Task Identity

One limitation of the list-based return is that you lose track of which input task produced which result if you only look at the output list. If you need to correlate results with inputs, zip them together explicitly.

import asyncio

async def process(item: str) -> str:
    if item == "bad":
        raise ValueError(f"Cannot process: {item}")
    return item.upper()

async def main():
    items = ["apple", "bad", "cherry"]
    outcomes = await asyncio.gather(
        *[process(i) for i in items],
        return_exceptions=True,
    )
    for item, outcome in zip(items, outcomes):
        if isinstance(outcome, BaseException):
            print(f"{item!r} -> ERROR: {outcome}")
        else:
            print(f"{item!r} -> {outcome}")

asyncio.run(main())

Because gather preserves input order in its output list, the zip is always accurate. This is one of the guarantees you can rely on.

Using asyncio.TaskGroup in Python 3.11+

asyncio.TaskGroup was added in Python 3.11 as a structured alternative to gather. It raises an ExceptionGroup (also new in 3.11) when any task fails, which means you get all exceptions, not just the first one.

import asyncio

async def risky(n: int) -> int:
    if n % 2 == 0:
        raise ValueError(f"Even number rejected: {n}")
    return n

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(risky(i)) for i in range(5)]
    except* ValueError as eg:
        print(f"Caught {len(eg.exceptions)} ValueError(s):")
        for exc in eg.exceptions:
            print(f"  {exc}")
    else:
        results = [t.result() for t in tasks]
        print(results)

asyncio.run(main())

The except* syntax (Python 3.11+) lets you handle specific exception types within an ExceptionGroup without writing manual loops. TaskGroup also cancels all sibling tasks the moment any task fails, which prevents partial work from continuing unnoticed — a stricter but safer default than gather.

If you are on Python 3.10 or earlier, stick with gather(return_exceptions=True) and the inspection patterns above.

Common Pitfalls

Catching BaseException instead of Exception

When you call isinstance(result, BaseException), you catch asyncio.CancelledError too. In most applications, cancellation is a control-flow signal that should propagate, not be swallowed alongside value errors and network failures. Check for CancelledError explicitly and re-raise it if your task is cancelled:

for outcome in outcomes:
    if isinstance(outcome, asyncio.CancelledError):
        raise  # let cancellation propagate
    elif isinstance(outcome, Exception):
        print(f"Handled error: {outcome}")

Mixing create_task and gather carelessly

If you create tasks with asyncio.create_task before passing them to gather, their exceptions are attached to the task object. Gather will still surface them correctly, but if the event loop moves forward before gather runs (because you await something else in between), a task could already be finished and its exception sitting unread. Create tasks and gather them in the same logical block.

Forgetting that gather returns results in input order

It is tempting to assume gather returns results as tasks complete. It does not. Results always match the order of the input awaitables, regardless of completion time. Relying on timing order will produce subtle, hard-to-reproduce bugs.

Using return_exceptions=True without checking the results

Passing return_exceptions=True and then ignoring the exception objects in the returned list is the most common way to silently swallow errors. The parameter does not handle exceptions for you — it only prevents gather from raising. You still have to inspect every item in the list.

Wrapping Up

Silent exceptions in asyncio.gather are almost always a handling problem, not an asyncio problem. Once you know the rules, the fixes are straightforward.

Here are concrete next steps:

• Audit every asyncio.gather call in your codebase. If any use return_exceptions=True, confirm that the caller actually iterates the results and handles BaseException instances.
• Add a wrapper function like extract_results so exception-checking logic is not duplicated across multiple call sites.
• If you are on Python 3.11+, migrate high-stakes concurrent workloads to asyncio.TaskGroup for automatic sibling cancellation and ExceptionGroup support.
• Search your logs for the warning "Task exception was never retrieved" — each one is a task whose failure went undetected at runtime.
• Write a unit test that deliberately raises in one of several gather coroutines and asserts that your handler logs or records the failure correctly.