How to Fix the asyncio.run() RuntimeError in Jupyter Notebook

In Jupyter Notebook, do not call asyncio.run() directly from a cell. The IPython kernel already runs an event loop, so a second loop cannot be started in the same thread. Replace asyncio.run(main()) with await main().

import asyncio
async def main():
await asyncio.sleep(1)
return "Готово"
result = await main()
print(result)

This is the primary fix for standard Jupyter Notebook, JupyterLab, and IPython environments that support top-level await.

Why the error occurs

Typical problematic code looks like this:

import asyncio
async def main():
await asyncio.sleep(1)
print("Готово")
asyncio.run(main())

When executed in a notebook, it may fail with the following exception:

RuntimeError: asyncio.run() cannot be called from a running event loop

The asyncio.run() function is intended for running a coroutine from ordinary synchronous code. It creates a new event loop, executes the supplied coroutine, finalizes asynchronous generators, and closes the loop it created.

According to the Python documentation, asyncio.run() cannot be called while another event loop is already running in the same thread. In Jupyter, that loop is managed by the IPython kernel. It is required to handle messages between the notebook interface and the running Python process, so user code normally should not stop or replace it.

In a Jupyter cell, use await coroutine(). In a regular Python script, use asyncio.run(coroutine()). These approaches are intended for different execution environments.

Fixing code in a notebook cell

Step 1. Find the asyncio.run() call

For example:

response = asyncio.run(load_data())

Step 2. Leave the asynchronous function unchanged

import asyncio
async def load_data():
await asyncio.sleep(0.5)
return {"status": "ok"}

Step 3. Call the coroutine with await

response = await load_data()
print(response)

IPython supports top-level await, so you can use it directly in a cell without placing the call inside another function.

Step 4. Verify the result

Do not verify only that no exception was raised. Check the expected value or state:

response = await load_data()
assert response["status"] == "ok"
print(response)

This test confirms that the coroutine actually completed and returned the expected data.

When main() is already implemented

You do not need to rewrite it. Only the entry point changes.

In a Python script:

import asyncio
async def main():
await asyncio.sleep(1)
print("Завершено")
if name == "main":
asyncio.run(main())

In Jupyter Notebook:

await main()

Do not move the if __name__ == "__main__" construct into a cell just to run the coroutine. In a notebook, it does not resolve the event-loop conflict: the asyncio.run() call would still be nested inside an already running loop.

Running multiple asynchronous operations

When tasks are independent, you can run them concurrently with asyncio.gather():

import asyncio
async def fetch_item(item_id):
await asyncio.sleep(0.2)
return {"id": item_id}
results = await asyncio.gather(
fetch_item(1),
fetch_item(2),
fetch_item(3),
)
print(results)

asyncio.run() is not needed here either: gather() returns an awaitable object that can be awaited with await.

When operations must run strictly in sequence, use multiple await calls:

first = await fetch_item(1)
second = await fetch_item(2)
print(first, second)

Creating a separate task

To run work in the background within the current event loop, create a task:

task = asyncio.create_task(fetch_item(10))
In this or the next cell:
result = await task
print(result)

asyncio.create_task() requires a running event loop. Jupyter normally already has one, so calling it from a cell’s asynchronous context is valid. Keep a reference to the task when its result will be retrieved later.

You can use the standard methods to inspect its state:

print(task.done())
if task.done() and not task.cancelled():
print(task.result())

The result() method must not be used as a replacement for await before the task has finished. Calling it on an incomplete task raises InvalidStateError.

How to cancel a running task

When cells are executed repeatedly, old tasks may continue running. A saved task can be cancelled:

task.cancel()
try:
await task
except asyncio.CancelledError:
print("Задача отменена")

Cancellation is delivered to the coroutine at the next await point. If the coroutine contains a long-running synchronous section without await, it cannot process the cancellation immediately.

When the error occurs inside a third-party library

Sometimes user code contains no asyncio.run(), but the exception appears after a library function is called:

result = library_function()

This means the library or one of its wrappers may be starting its own coroutine with asyncio.run(). Inspect the full traceback: it should identify the file and line containing the conflicting call.

Use the following preferred order of fixes:

Find the library’s asynchronous API, such as a method with an async suffix, a separate asynchronous client, or a function that returns a coroutine.
Call that API with await.
When you control the library, remove asyncio.run() from the internal function and make the function asynchronous.
When only a synchronous API is available, do not assume it is safe to call from the current event loop. Check the official documentation for that specific library.

Example of an incorrect wrapper:

def load_sync():
    return asyncio.run(load_data())

In a notebook, replace such a wrapper with an asynchronous one:

async def load_async():
    return await load_data()
response = await load_async()

A universal function for scripts and Jupyter

Avoid creating a function that automatically calls asyncio.run() or returns a task depending on whether an event loop exists. Such a function changes its return type: in one case it returns a completed value, while in the other it returns a Task. This complicates error handling and makes the behavior unclear.

It is more reliable to separate the asynchronous logic from the entry points:

import asyncio
async def application():
await asyncio.sleep(0.1)
return 42
def run_script():
return asyncio.run(application())

In a script:

value = run_script()
print(value)

In Jupyter:

value = await application()
print(value)

This keeps the asynchronous function shared while making the startup method explicitly match the execution environment.

Checking the active event loop

For diagnostics inside a cell, you can retrieve the currently running loop:

import asyncio
loop = asyncio.get_running_loop()
print(type(loop).name)
print(loop.is_running())

When called in the context of an active loop, get_running_loop() returns the loop object. In synchronous code without a running loop, the function raises RuntimeError.

Do not use this check as a reason to call loop.run_until_complete() in Jupyter. The run_until_complete() method also attempts to control the loop and is not intended to restart an event loop that is already running.

Why you should not close Jupyter’s event loop manually

The following code is not a valid fix:

loop = asyncio.get_event_loop()
loop.close()

The event loop belongs to the notebook runtime. Closing it forcibly may disrupt the kernel, asynchronous libraries, and tasks that have already been created. User code should run within the provided loop through await, not replace it.

What to do about the unawaited coroutine warning

After a failed asyncio.run(main()) call, the main exception may be accompanied by the following warning:

RuntimeWarning: coroutine 'main' was never awaited

It means that calling main() created a coroutine object, but that object was never executed with await. The fix is the same:

await main()

Do not suppress this warning with warnings filters. It indicates a real coroutine-management error.

Minimal diagnostic example

Run the cells in order.

First cell:

import asyncio
async def check_asyncio():
await asyncio.sleep(0.1)
return "asyncio работает"

Second cell:

message = await check_asyncio()
assert message == "asyncio работает"
print(message)

If this example runs successfully, top-level await is supported. Look for the original error in a specific asyncio.run() call or in the internal code of the library being used.

Limitations of this solution

This solution applies to Jupyter/IPython environments where top-level await is available.
The exact event-loop integration may depend on the kernel and the async backend selected in IPython.
For third-party libraries, follow their official asynchronous APIs and resource-management requirements.
Replacing asyncio.run() with await does not fix blocking synchronous code inside a coroutine.
Network clients, file objects, and other asynchronous resources must still be closed correctly according to the documentation of the relevant library.

Final checklist

Find asyncio.run(...) in the cell or traceback.
Call the coroutine with await.
For multiple independent coroutines, use await asyncio.gather(...).
For a separately managed operation, use asyncio.create_task() and then await task.
Do not call run_until_complete() on an event loop that is already running.
Do not close the event loop managed by Jupyter.
Separate the shared asynchronous function from the entry points used by scripts and notebooks.
Verify the result with a return value, an assert statement, or the expected state.