PythonMastery
reference 3 min read · lesson 41 of 45 in Errors

RuntimeError: This event loop is already running

1 · The lesson

read

What this error means

You called asyncio.run() — or loop.run_until_complete() — from inside code that an event loop was already running. asyncio loops do not nest: a loop cannot start a second loop on the same thread while it is in the middle of running the first.

It is the opposite failure to "Event loop is closed". There the loop had finished; here it never stopped.

When you see it

Almost always in a Jupyter or Colab notebook, because the notebook is itself an async application:

python
Traceback (most recent call last):
  File "<ipython-input-3>", line 4, in <module>
    asyncio.run(fetch_all(urls))
RuntimeError: asyncio.run() cannot be called from a running event loop

Or from inside a request handler that is already async:

python
RuntimeError: This event loop is already running

Why it happens

asyncio.run() is designed to be the entry point — it creates a loop, runs one coroutine to completion, and closes the loop. Calling it from inside a running loop asks for a second loop on a thread that already has one.

A notebook cell looks like a plain script, which is what makes this surprising. It is not: IPython runs an event loop so that await works at the top level of a cell.

How to fix it

In a notebook, just await it. Top-level await works in a cell precisely because a loop is already running:

python
results = await fetch_all(urls)      # no asyncio.run() needed

In an async function, await the coroutine rather than starting a loop for it:

python
async def handler(request):
    data = await fetch_all(urls)     # not asyncio.run(fetch_all(urls))
    return data
+ setup added so this can run · defines fetch_all, urls
# 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_all(*_a, **_kw):
    print('-> fetch_all() called')
    return _AutoMock('fetch_all()')
urls = _AutoMock('urls')

To call async code from a synchronous function that a loop is already running, hand the work to another thread that has no loop of its own:

python
import asyncio

def sync_wrapper(urls):
    return asyncio.run(fetch_all(urls))          # fine on a thread with no loop

async def handler():
    return await asyncio.to_thread(sync_wrapper, urls)
+ setup added so this can run · defines fetch_all
# 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_all(*_a, **_kw):
    print('-> fetch_all() called')
    return _AutoMock('fetch_all()')

The cleaner answer is usually to make the caller async instead of bridging back and forth.

nest_asyncio exists and is worth being careful with. It patches asyncio to permit nesting and will make the error stop:

python
import nest_asyncio
nest_asyncio.apply()

It is a reasonable convenience in a notebook. It is a poor idea in a service: it changes how the loop schedules work, and the bugs it produces are timing-dependent and unpleasant to reproduce.

When you'd actually see this in real code

  • Pasting a script into a notebook and running it unchanged, where the asyncio.run(main()) at the bottom is exactly the line that breaks.
  • A synchronous library function called from an async handler that reaches for asyncio.run() internally.
  • A test calling asyncio.run() inside a suite whose plugin already provides a loop.
  • Any code where a sync layer sits between two async layers — the sandwich that produces this most often.

See Also