How to Fix RuntimeError: Event loop is closed in Python asyncio

The RuntimeError: Event loop is closed error means that the program is trying to schedule a coroutine, callback, or I/O operation on an event loop for which loop.close() has already been called. The fix is not to create new loops in arbitrary places, but to identify the owner of the event loop, move loop closure to a single shutdown point, and stop all background tasks before closing the loop.

The primary solution for Python 3.7 and later is to run the top-level coroutine with asyncio.run() and avoid closing the loop manually:

import asyncio
async def main() -> None:
await asyncio.sleep(0.1)
print("Работа завершена")
if name == "main":
asyncio.run(main())

asyncio.run() creates an event loop, runs the supplied coroutine, finalizes asynchronous generators, and closes the loop after main() exits. A single call should cover the entire lifecycle of the asynchronous part of the program.

Why the error occurs

A closed event loop cannot be reused. After loop.close() is called, methods that schedule tasks and callbacks may raise RuntimeError. Closing the loop is irreversible.

A typical incorrect sequence looks like this:

import asyncio
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(asyncio.sleep(0.1))
loop.close()
Error: the loop is already closed.
loop.run_until_complete(asyncio.sleep(0.1))

A similar problem occurs when a function or class closes a loop even though the loop was created by the calling code:

import asyncio
def run_job(loop: asyncio.AbstractEventLoop) -> None:
loop.run_until_complete(asyncio.sleep(0.1))
loop.close()  # Incorrect: the function does not own the loop.
loop = asyncio.new_event_loop()
run_job(loop)
The calling code expects the loop to remain available.
loop.run_until_complete(asyncio.sleep(0.1))

The ownership rule is simple: the component that created the event loop is responsible for closing it permanently. A helper function may create and await tasks, but it must not close a loop it does not own.

Do not fix this error by repeatedly calling asyncio.new_event_loop() before every operation. This hides a lifecycle violation, leaves tasks unfinished, and may cause leaks of network connections, file descriptors, and executor threads.

Step 1. Find where the loop is closed

Inspect the traceback and find the first call from your code before the line inside asyncio. Then search the project for the following operations:

loop.close();
asyncio.run(...) followed by reuse of a reference to an object created inside it;
loop.stop() followed by loop closure;
close(), shutdown(), or disconnect() methods in your own classes;
background threads that call loop.call_soon_threadsafe() after the application has shut down.

If the loop is passed between objects, temporarily add a check immediately before the failing operation:

if loop.is_closed():
    raise RuntimeError("Попытка использовать уже закрытый event loop")

This check helps locate the error, but it is not a final fix. The loop may still be closed by another thread between the check and the next method call. The lifecycle must eliminate this race condition by design.

Step 2. Keep a single top-level entry point

A common cause of the error is making several sequential calls to asyncio.run() while reusing objects bound to the previous loop:

import asyncio
class Client:
def init(self) -> None:
self.loop = None
async def connect(self) -> None:
    self.loop = asyncio.get_running_loop()

async def send(self) -> None:
    self.loop.call_soon(lambda: None)
client = Client()
asyncio.run(client.connect())
asyncio.run(client.send())  # client stores a reference to the closed loop.

The object was created once, but its methods run in two different loops. After the first asyncio.run() call, the first loop is closed and the stored reference becomes invalid.

Move all operations into a single coroutine:

import asyncio
class Client:
async def connect(self) -> None:
self.loop = asyncio.get_running_loop()
async def send(self) -> None:
    self.loop.call_soon(lambda: None)
async def main() -> None:
client = Client()
await client.connect()
await client.send()
asyncio.run(main())

It is even safer not to store the loop in an object attribute unless necessary. Inside a coroutine, obtain the current loop with asyncio.get_running_loop() immediately before using it.

Step 3. Do not close the loop inside library code

A function that accepts an existing loop must leave it open:

import asyncio
async def perform_job() -> str:
await asyncio.sleep(0.1)
return "done"
def run_with_existing_loop(
loop: asyncio.AbstractEventLoop,
) -> str:
return loop.run_until_complete(perform_job())

The loop should be closed only where it was created:

loop = asyncio.new_event_loop()
try:
asyncio.set_event_loop(loop)
result = run_with_existing_loop(loop)
print(result)
finally:
loop.close()

Manual loop management is unnecessary for a typical application: asyncio.run() is preferred. A manually managed loop is appropriate when the application integrates with a framework, GUI, server, or another component that defines its own lifecycle.

Step 4. Finish background tasks before closing the loop

The error may occur not in the main code, but in a callback that runs after shutdown has already started. Before closing the loop, stop every producer of new tasks: timers, queues, network clients, file watchers, and worker threads.

In an application that uses asyncio.run(), keep references to created tasks and cancel them inside main():

import asyncio
async def worker() -> None:
try:
while True:
await asyncio.sleep(1)
except asyncio.CancelledError:
# Release resources owned by this worker here.
raise
async def main() -> None:
task = asyncio.create_task(worker())
try:
    await asyncio.sleep(0.1)
finally:
    task.cancel()
    await asyncio.gather(task, return_exceptions=True)
asyncio.run(main())

task.cancel() only requests cancellation. You must return control to the event loop and wait for the task to finish. In this example, asyncio.gather() does that. Without awaiting the task, cleanup code in the background coroutine's finally block may not run before the loop is closed.

If a task uses a network client, thread, subprocess, or file, close that resource before shutting down the infrastructure it depends on. For objects that support an asynchronous context manager, use async with:

async def main() -> None:
    async with create_client() as client:
        await client.send_request()

The exact method used to create and close the client depends on the library. Do not replace the documented resource cleanup method with an arbitrary sleep().

Step 5. Follow the correct shutdown order with manual loop management

The following pattern is intended for Python 3.9 and later because it uses loop.shutdown_default_executor():

import asyncio
async def main() -> None:
await asyncio.sleep(0.1)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(main())
finally:
pending = asyncio.all_tasks(loop)
for task in pending:
    task.cancel()

if pending:
    loop.run_until_complete(
        asyncio.gather(*pending, return_exceptions=True)
    )

loop.run_until_complete(loop.shutdown_asyncgens())
loop.run_until_complete(loop.shutdown_default_executor())
loop.close()

The order of operations is essential:

the main coroutine stops creating new work;
remaining tasks receive a cancellation request;
the loop runs until the cancelled tasks finish;
asynchronous generators are finalized;
the default executor is shut down;
only then is loop.close() called.

For Python 3.7–3.8, shutdown_default_executor() is not applicable in this pattern. Use the documentation for the exact installed Python branch and do not copy a call to an API that is unavailable in that version.

Step 6. Stop external threads and callback sources

A worker thread may retain a reference to the loop and submit a callback with call_soon_threadsafe(). If the main thread has already closed the loop, the next callback submission will fail.

The correct shutdown sequence is:

set the worker thread's stop flag;
stop receiving new events from the external source;
wait for the thread to exit;
cancel the remaining asyncio tasks;
close the event loop.

Checking loop.is_closed() inside the thread may reduce the number of invalid calls, but it does not eliminate the race condition. The reliable solution is to guarantee that the thread has exited before loop.close() is called.

Step 7. Enable asyncio debug mode

For diagnostics, enable the documented debug mode:

import asyncio
async def main() -> None:
await asyncio.sleep(0.1)
asyncio.run(main(), debug=True)

It can help detect certain coroutines that were not awaited correctly, slow callbacks, and unsafe API calls from another thread. Diagnostic messages do not replace traceback analysis: the error must still be fixed by changing the startup and shutdown order.

Special case: an event loop is already running

In an interactive environment, server framework, or GUI, the event loop may already be owned by the platform. In that context, do not call asyncio.run() unconditionally or close the loop provided by the platform.

If the code is already running inside an async def function, invoke the coroutine with await:

async def handler() -> None:
    result = await perform_job()
    print(result)

The asyncio.run() cannot be called from a running event loop message is different from Event loop is closed, although both errors often appear after attempts to manage the loop manually. Do not apply the fix for one error to the other without checking the actual traceback.

How to verify the fix

After changing the code, test normal shutdown, exception handling, and forced operation cancellation. The fix can be considered correct when all of the following conditions are met:

the application has one clearly defined owner of the event loop;
loop.close() is called only by the owner and only at the end;
objects do not use a loop reference from a previous asyncio.run() call;
background tasks are cancelled and awaited;
external threads stop submitting callbacks before the loop is closed;
network clients, generators, and executors are shut down before loop.close();
restarting the application does not depend on a global closed loop.

The minimal practical fix for most console applications is to move all asynchronous operations into one main() function, call it once with asyncio.run(main()), remove manual loop.close() calls from nested functions, and explicitly finish background tasks inside main().

Sources

The links point to the general Python 3 documentation branch. Web verification was unavailable when this material was prepared, so for version-dependent behavior, also open the documentation for the specific installed Python version by selecting it in the version switcher on the website.