Skip to content

Installing a Custom Exception Handler on the Event Loop

Some asyncio errors have nowhere to go. A background task raises and nobody ever awaits it; a callback scheduled with call_soon() throws; a protocol's data_received() fails deep inside a transport. There is no caller to propagate to, so the loop hands the error to its exception handler, and the default handler writes a multi-line message to the asyncio logger at ERROR level. In a service that ships structured JSON logs to an error tracker, that message arrives as free text without a request ID, without a fingerprint, and often minutes late — a never-retrieved task exception is only reported when the task object is garbage-collected. This guide replaces the default with a handler that turns every orphaned error into a structured event, keeps the default behaviour as a fallback, distinguishes expected shutdown noise from real failures, and is installed early enough to catch errors during startup.

Prerequisites

Where does an exception go? A decision on Is anyone awaiting the code that raised with 3 outcomes. Where does an exception go? Is anyone awaiting the code that raised? yes, a coroutine awaits it propagates to the caller normal try / except no, orphaned task loop exception handler reported at GC time no, callback or protocol loop exception handler reported immediately The handler sees only errors with nowhere else to go.

1. See which errors reach the handler

The handler is not a global except. Exceptions inside a coroutine that someone awaits propagate to that awaiter and never reach it. The handler receives only errors the loop cannot deliver anywhere else. Record what arrives before designing the handler.

import asyncio
import gc

seen: list[tuple[str, str | None, list[str]]] = []


def recording_handler(loop: asyncio.AbstractEventLoop, context: dict) -> None:
    exc = context.get("exception")
    extra_keys = sorted(k for k in context if k not in ("message", "exception"))
    seen.append((context["message"], type(exc).__name__ if exc else None, extra_keys))


async def boom() -> None:
    raise ValueError("nobody awaits me")


def bad_callback() -> None:
    raise KeyError("callback failed")


async def main() -> None:
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(recording_handler)

    asyncio.create_task(boom())                  # orphaned: reference dropped, never awaited
    await asyncio.sleep(0.01)
    gc.collect()                                 # the report happens when the task is collected

    loop.call_soon(bad_callback)                 # a plain callback that raises
    await asyncio.sleep(0.01)

    awaited = asyncio.create_task(boom())        # awaited: handled by the caller instead
    try:
        await awaited
    except ValueError:
        pass


asyncio.run(main())
for entry in seen:
    print(entry)

Two entries are recorded. The orphaned task produces ('Task exception was never retrieved', 'ValueError', ['future']), and the callback produces a message beginning Exception in callback bad_callback() with a handle key. The awaited task produces nothing, because its exception reached the except block.

Verify: the output has exactly two entries. If the orphaned-task entry is missing, the task object is still referenced somewhere — which is its own bug, since the error will then never be reported at all.

2. Write a handler that emits structured events

The context dictionary carries a message, usually an exception, and optional keys describing where it happened: future or task, handle, protocol, transport, socket, asyncgen. A production handler extracts what is useful, attaches a stable fingerprint, logs a single structured record, and never raises itself.

import asyncio
import hashlib
import logging
import traceback

log = logging.getLogger("asyncio.orphaned")


def fingerprint(exc: BaseException | None, message: str) -> str:
    if exc is None:
        return hashlib.sha1(message.encode()).hexdigest()[:12]
    frames = traceback.extract_tb(exc.__traceback__)
    where = f"{frames[-1].filename}:{frames[-1].name}" if frames else ""
    return hashlib.sha1(f"{type(exc).__name__}|{where}".encode()).hexdigest()[:12]


def structured_handler(loop: asyncio.AbstractEventLoop, context: dict) -> None:
    try:
        exc = context.get("exception")
        message = context.get("message", "unhandled error in event loop")
        source = context.get("task") or context.get("future") or context.get("handle")
        event = {
            "event": "asyncio_orphaned_error",
            "message": message,
            "exception_type": type(exc).__name__ if exc else None,
            "fingerprint": fingerprint(exc, message),
            "source": repr(source)[:200] if source is not None else None,
            "task_name": source.get_name() if isinstance(source, asyncio.Task) else None,
        }
        log.error("orphaned asyncio error", extra={"asyncio": event}, exc_info=exc)
    except Exception:                                     # the handler must never raise
        loop.default_exception_handler(context)

Falling back to loop.default_exception_handler(context) inside the except means a bug in your handler degrades to the stock output rather than losing the error. The fingerprint groups identical failures by exception type and the innermost frame, which is how error trackers keep one bug from becoming ten thousand alerts. Log handlers that add request IDs, as in propagating request IDs with contextvars, enrich this record too when the failing callback ran inside a request; errors reported at garbage collection carry whatever context is current at that moment, so do not rely on them for attribution.

Verify: trigger both failures from step 1 with structured_handler installed; each produces one log record carrying exception_type, fingerprint and, for the task, task_name.

3. Separate shutdown noise from real failures

During shutdown and connection churn, some messages are expected: a client disconnects mid-write and the transport reports ConnectionResetError; a task is cancelled while the loop closes. Alerting on those trains people to ignore the channel. Classify before logging, and downgrade known-benign cases instead of dropping them.

import asyncio
import logging

log = logging.getLogger("asyncio.orphaned")

BENIGN = (ConnectionResetError, BrokenPipeError, asyncio.CancelledError)


def classifying_handler(loop: asyncio.AbstractEventLoop, context: dict) -> None:
    exc = context.get("exception")
    if isinstance(exc, BENIGN):
        log.info("benign asyncio error: %s (%s)", context.get("message"), type(exc).__name__)
        return
    if exc is None and "Unclosed" in context.get("message", ""):
        log.warning("resource not closed: %s", context["message"])   # a leak, not a crash
        return
    structured_handler(loop, context)

Keeping benign cases at INFO preserves the evidence if a "benign" error suddenly spikes — a surge of ConnectionResetError from one upstream is still worth seeing on a dashboard. The unclosed-resource branch catches messages such as unclosed transports reported when objects are finalised, which point at leaks rather than failures.

Verify: a peer that resets its connection mid-response produces an INFO record, while a KeyError in a callback still produces the ERROR event from step 2.

Inside a production exception handler 5 stages from read context to fallback. Inside a production exception handler read context message, exception classify benign, leak, failure structure type, fingerprint emit one log record fallback default handler The last stage only runs when the handler itself fails.

4. Install the handler before anything can fail

A handler set after startup misses errors in startup itself — an orphaned warm-up task, a failing connection callback. Install it on the loop before the main coroutine runs, by creating the loop yourself through asyncio.Runner's loop_factory.

import asyncio


def loop_with_handler() -> asyncio.AbstractEventLoop:
    loop = asyncio.new_event_loop()
    loop.set_exception_handler(classifying_handler)
    return loop


async def serve() -> None:
    asyncio.create_task(warm_cache())             # an orphaned startup failure is now caught
    await asyncio.sleep(0.05)


async def warm_cache() -> None:
    raise RuntimeError("cache backend unreachable")


if __name__ == "__main__":
    with asyncio.Runner(loop_factory=loop_with_handler) as runner:
        runner.run(serve())

The same factory is the right place to select a loop implementation, so the handler survives a switch to uvloop — which calls the handler set with set_exception_handler() just like the stdlib loop. Frameworks that create the loop for you, such as uvicorn, usually offer a startup hook; set the handler there, as the first statement, with asyncio.get_running_loop().set_exception_handler(...).

Installing early versus late 2 columns contrasting inside main(), in the loop factory. Installing early versus late inside main() set after the loop starts misses earlier failures easy to forget in tests per framework hook in the loop factory set before any code runs covers startup errors survives backend changes one place to audit Install it where the loop is created, not where the app happens to start.

Verify: RuntimeError: cache backend unreachable appears as a structured event even though it happened before serve() finished starting.

5. Test the handler like production code

A handler that silently broke is worse than none, because it swallows the errors it was meant to surface. Call it directly with representative contexts in unit tests, and assert that it falls back on failure.

import asyncio
import logging
import unittest


class ExceptionHandlerTests(unittest.IsolatedAsyncioTestCase):
    async def test_task_error_is_structured(self) -> None:
        loop = asyncio.get_running_loop()
        exc = ValueError("boom")
        with self.assertLogs("asyncio.orphaned", level="ERROR") as logs:
            classifying_handler(loop, {"message": "Task exception was never retrieved",
                                       "exception": exc})
        record = logs.records[0]
        self.assertEqual(record.asyncio["exception_type"], "ValueError")

    async def test_benign_errors_are_downgraded(self) -> None:
        loop = asyncio.get_running_loop()
        with self.assertLogs("asyncio.orphaned", level="INFO") as logs:
            classifying_handler(loop, {"message": "write failed",
                                       "exception": ConnectionResetError()})
        self.assertEqual(logs.records[0].levelno, logging.INFO)


if __name__ == "__main__":
    unittest.main()

Testing through assertLogs checks the observable behaviour — what reaches the logs — rather than implementation details. Add a case for a context without an exception key, which some loop messages use.

Verify: both tests pass, and deliberately breaking fingerprint() makes the first test fail while the fallback still emits the default handler's output.

Verification

The custom handler is working when:

  • Orphaned errors are structured: never-retrieved task exceptions and failing callbacks appear as single structured records with a type, fingerprint and source.
  • Nothing is lost on handler bugs: a failure inside the handler falls back to default_exception_handler.
  • Noise is classified, not dropped: benign connection errors log at INFO, leaks at WARNING, and real failures at ERROR.
  • Startup is covered: the handler is installed through the loop factory or the framework's first startup hook.
  • The handler is tested: unit tests exercise error, benign and message-only contexts.

Pitfalls & edge cases

  • Relying on the handler instead of awaiting tasks. A never-retrieved exception is reported only at garbage collection, possibly much later, and never if a reference survives. Use a TaskGroup or keep and await task references; the handler is a safety net.
  • Raising inside the handler. An exception escaping the handler is itself logged by the loop and obscures the original error. Wrap the body and fall back to the default handler.
  • Blocking work in the handler. The handler runs on the loop thread. Sending an HTTP request to an error tracker synchronously stalls the loop; enqueue the event for a background exporter instead.
  • Handler set on the wrong loop. set_exception_handler() affects one loop. Services running multiple loops in separate threads must install it on each.
  • Swallowing everything as benign. A broad except OSError downgrade hides disk-full and permission errors. Keep the benign list narrow and specific.

Frequently Asked Questions

What does loop.set_exception_handler do in asyncio?

It replaces the function the event loop calls for errors it cannot deliver to any awaiting code, such as exceptions in tasks that were never awaited, exceptions raised by callbacks scheduled with call_soon, and errors inside protocols and transports. The handler receives the loop and a context dictionary with a message, usually an exception, and details about where it happened.

Why isn't my exception handler called for errors in awaited tasks?

Because those errors are not orphaned. When a coroutine awaits a task, the task's exception propagates to that coroutine, where normal try and except handling applies. The loop's exception handler only receives errors that have no awaiting caller, such as a background task whose result nobody retrieves.

When is Task exception was never retrieved reported?

When the failed task object is garbage-collected without anyone having called result, exception or awaited it. That can happen immediately after the last reference is dropped or much later, and it never happens if a reference is kept forever. Awaiting tasks or using TaskGroup reports errors deterministically instead.

How do I set an asyncio exception handler before asyncio.run starts my code?

Create the loop yourself with a factory function that calls asyncio.new_event_loop, sets the exception handler on it, and returns it, then pass that function as loop_factory to asyncio.Runner, or to asyncio.run on Python 3.12 and later. The handler is then active before the first line of your main coroutine runs.