RuntimeError: Event loop is closed: asyncio cause and fix

What this error means

Something asked the asyncio event loop to do work after the loop had already shut down. The coroutine, the connection or the callback still existed after the loop that was supposed to run it had closed.

It nearly always appears after your program has finished its real work, which makes it confusing: the output is correct, and then a traceback appears anyway.

When you see it

python

Traceback (most recent call last):
  File "main.py", line 14, in <module>
    asyncio.run(second())
  File ".../asyncio/runners.py", line 190, in run
    return runner.run(main)
RuntimeError: Event loop is closed

Or from a destructor, printed after everything else, with no line of yours in it:

python

Exception ignored in: <function _ProactorBasePipeTransport.__del__>
RuntimeError: Event loop is closed

Why it happens

Calling asyncio.run() more than once and reusing something from the first call. asyncio.run() creates a loop, runs the coroutine, then closes the loop. A client created inside the first call holds a reference to a loop that no longer exists.
A background task that was never awaited. The loop closes while the task is still pending, and an error is reported when the task is cleaned up.
A client that was never closed, such as an aiohttp.ClientSession or an async database pool. Its __del__ tries to close a transport on a loop that is already closed.
On Windows, the Proactor transport is cleaned up late. This produces the Exception ignored in: __del__ form even when your own code is correct.

How to fix it

Call asyncio.run() once, at the top, and do everything inside it:

python

import asyncio

async def main():
    await first()
    await second()          # same loop, one entry point

asyncio.run(main())          # exactly once

+ setup added so this can run · defines first, second

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def first(*_a, **_kw):
    print('-> first() called')
    return _AutoMock('first()')
def second(*_a, **_kw):
    print('-> second() called')
    return _AutoMock('second()')

This rule avoids nearly all of these errors: asyncio.run() is the boundary between sync and async code, and a program should cross it only once.

Close clients with async with, so they are gone before the loop is:

python

import aiohttp

async def main():
    async with aiohttp.ClientSession() as session:      # closed when the block ends
        async with session.get(url) as r:
            return await r.json()

asyncio.run(main())

+ setup added so this can run · defines asyncio, url, session, r

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

asyncio = _AutoMock('asyncio')
url = _AutoMock('url')
session = _AutoMock('session')
r = _AutoMock('r')

Await your background tasks before returning. A task that is created and then forgotten is the most common cause:

python

async def main():
    tasks = [asyncio.create_task(fetch(u)) for u in urls]
    return await asyncio.gather(*tasks)      # nothing outlives the loop

+ setup added so this can run · defines urls, asyncio, fetch

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

urls = ["alpha", "beta", "gamma"]
asyncio = _AutoMock('asyncio')
def fetch(*_a, **_kw):
    print('-> fetch() called')
    return _AutoMock('fetch()')

Keep a reference to every task you create. The loop keeps only a weak reference, so a task with no other reference can be garbage collected before it finishes.

In pytest, let the plugin manage the loop instead of calling asyncio.run() in a test:

python

import pytest

@pytest.mark.asyncio
async def test_fetch():
    assert await fetch("/health") == {"ok": True}

+ setup added so this can run · defines fetch

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def fetch(*_a, **_kw):
    print('-> fetch() called')
    return _AutoMock('fetch()')

When you'd see this in real code

A script that runs two asyncio.run() calls in sequence and shares a client between them.
A test suite where one test closes the loop and the next one inherits a client bound to it.
A scraper whose ClientSession is created at module level, outside any loop.
A clean run on Linux that prints Exception ignored in: __del__ on Windows, from the same unclosed transport.

RuntimeError: This event loop is already running: the opposite problem, starting a loop from inside a running loop, typically in a notebook.
TimeoutError / requests.exceptions.ReadTimeout: in async code, timeouts use asyncio.timeout().

RuntimeError: Event loop is closed

1 · The lesson

What this error means

When you see it

Why it happens

How to fix it

When you'd see this in real code

See Also

RuntimeError: Event loop is closed

What this error means

When you see it

Why it happens

How to fix it

When you'd see this in real code

Related errors

See Also