How to Fix RuntimeError: Event Loop Is Closed in asyncio

The RuntimeError: Event loop is closed error means that the code is trying to schedule a task, execute a callback, close a network transport, or access an asyncio object after the event loop has been closed. The fix comes down to two rules: the loop must be closed only after all asynchronous operations have finished, and objects bound to the loop must not be reused after it has been closed.

The examples below are intended for CPython with the asyncio module. The main recommended approach for regular programs is a single call to asyncio.run() at the entry point. For libraries, Jupyter, testing frameworks, and applications with their own event loop, the startup method must be coordinated with the environment that manages the loop.

1. Find Where the Event Loop Is Closed Too Early

Start by examining the full traceback, not just the last line. The important part is the first call from your code before execution enters internal asyncio functions. The problem usually falls into one of the following categories:

an asynchronous client, session, queue, lock, or task is reused after asyncio.run() returns;
the loop is closed manually with loop.close() while unfinished tasks still exist;
a background task is created with asyncio.create_task(), but the program exits without waiting for it;
an asynchronous resource is closed only after the event loop has already been closed;
a synchronous function repeatedly calls asyncio.run() while retaining objects from the previous loop between calls;
the code runs inside an environment where an event loop is already running.

Temporarily enable debug mode during diagnosis. It helps detect forgotten coroutine objects, slow callbacks, and incorrect task shutdown:

import asyncio
async def main():
...
if name == "main":
asyncio.run(main(), debug=True)

You can also run the program with the PYTHONASYNCIODEBUG=1 environment variable. This is the standard asyncio debugging mechanism, but the messages depend on the specific scenario and Python version.

2. Use One asyncio.run() Call at the Entry Point

asyncio.run() creates a new event loop, runs the supplied coroutine, finalizes asynchronous generators, shuts down the executor, and then closes the loop. After asyncio.run() returns, that loop must not be used again.

A correct program structure:

import asyncio
async def load_data():
await asyncio.sleep(0.1)
return {"status": "ok"}
async def main():
result = await load_data()
print(result)
if name == "main":
asyncio.run(main())

A problematic structure uses multiple separate runs while retaining an object associated with the first loop:

import asyncio
queue = None
async def create_queue():
global queue
queue = asyncio.Queue()
async def use_queue():
await queue.put("value")
asyncio.run(create_queue())
asyncio.run(use_queue())

The first call finishes and closes its event loop. The second call creates a different loop. Even if a particular object does not immediately raise an error in a certain Python version, this architecture is unreliable: asynchronous tasks and resources should be created and used within a single run.

Corrected version:

import asyncio
async def main():
queue = asyncio.Queue()
await queue.put("value")
print(await queue.get())
if name == "main":
asyncio.run(main())

Do not store tasks, Future objects, network clients, connections, queues, locks, or other asynchronous objects in global variables if the program may start multiple independent event loops. Create them inside a shared coroutine and close them before that coroutine returns.

3. Do Not Close the Event Loop Manually Unless Necessary

In application code, manual management with asyncio.new_event_loop(), run_until_complete(), and loop.close() is usually unnecessary. The error often appears because the loop is closed unconditionally in a finally block while background operations are still running.

Risky version:

import asyncio
loop = asyncio.new_event_loop()
try:
loop.run_until_complete(main())
finally:
loop.close()

This pattern is valid only when you have full control over the lifecycle. If main() creates background tasks without waiting for them to finish, they will still be pending when loop.close() is called.

For a regular program, replace manual management with:

if __name__ == "__main__":
    asyncio.run(main())

If a manually managed loop is genuinely required, cancel the remaining tasks and wait for cancellation handling before closing it:

import asyncio
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()

This code applies when the application creates the event loop itself and fully controls it. Do not use it inside a framework or interactive environment that owns the loop. The shutdown_default_executor() method is part of managing the loop's executor; when using asyncio.run(), a separate call is not required.

4. Wait for Background Tasks

asyncio.create_task() runs a coroutine concurrently, but it does not automatically make that coroutine part of the awaited result. If you create a task without awaiting it, the main code may finish first.

Problematic example:

import asyncio
async def worker():
await asyncio.sleep(1)
print("done")
async def main():
asyncio.create_task(worker())
asyncio.run(main())

main() finishes immediately after creating the task. When asyncio.run() shuts down, remaining tasks are cancelled. If the task or the library it uses handles shutdown incorrectly, subsequent operations may attempt to access an already closed event loop.

The fix is to wait for the task explicitly:

async def main():
    task = asyncio.create_task(worker())
    await task

For several independent operations, use asyncio.gather():

async def main():
    await asyncio.gather(
        worker(),
        worker(),
        worker(),
    )

In Python versions where asyncio.TaskGroup is available, structured concurrency can be written as follows:

async def main():
    async with asyncio.TaskGroup() as group:
        group.create_task(worker())
        group.create_task(worker())

TaskGroup waits for the tasks created within it when the context manager exits and coordinates error handling for the group. Check whether this API is available in the documentation for the installed Python version.

5. Close Asynchronous Resources Before main() Finishes

Network clients, database connections, and other resources may perform asynchronous work while closing. If their destructor runs after asyncio.run() has finished, it can no longer access the closed event loop.

Problematic pattern:

client = None
async def main():
global client
client = AsyncClient()
await client.request()
asyncio.run(main())
Closing happens too late
asyncio.run(client.close())

The second asyncio.run() creates a different loop, while the client may still be bound to the first one. The correct approach is to create, use, and close the resource inside the same coroutine:

async def main():
    client = AsyncClient()
try:
    await client.request()
finally:
    await client.close()
asyncio.run(main())

If the library supports an asynchronous context manager, prefer using it:

async def main():
    async with AsyncClient() as client:
        await client.request()

The method names close() and aclose(), as well as support for async with, depend on the specific library. Check its official documentation, and do not replace synchronous cleanup with asynchronous cleanup or vice versa.

6. Do Not Call asyncio.run() Inside an Already Running Event Loop

In Jupyter, a GUI framework, a web server, or an asynchronous test environment, an event loop may already be running. Calling asyncio.run() from a running loop is not a valid way to start a nested coroutine.

Inside an asynchronous function, use a regular await:

async def handler():
    result = await load_data()
    return result

In an interactive environment that supports top-level await, run:

result = await load_data()

Do not close an event loop obtained from a framework, server, or interactive shell. The environment manages its lifecycle. Calling loop.close() may break subsequent requests, tests, or notebook cells.

7. Check Whether an Object Is Reused After the First Run

A common scenario is a class that creates an asynchronous client once, while its method is called through separate asyncio.run() invocations:

service = Service()
asyncio.run(service.fetch())
asyncio.run(service.fetch())

If Service stores a connection, task, Future, or pool internally, the second call may access an object associated with the first loop, which has already been closed.

Fix this in one of two ways. The first is to perform all operations in a single run:

async def main():
    service = Service()
try:
    await service.fetch()
    await service.fetch()
finally:
    await service.close()
asyncio.run(main())

The second option is to create a completely new resource instance for each independent run:

async def run_once():
    service = Service()
try:
    await service.fetch()
finally:
    await service.close()
asyncio.run(run_once())
asyncio.run(run_once())

The second approach is valid only when the instances truly do not share asynchronous state.

8. Add a Controlled Application Shutdown

Servers, bots, and long-running processes should shut down in a defined order:

stop accepting new work;
signal background tasks to stop;
wait for them to finish or cancel them;
close client sessions, connections, and pools;
only then return from the main coroutine.

A minimal template using a stop event:

import asyncio
async def worker(stop_event):
while not stop_event.is_set():
await asyncio.sleep(0.5)
async def main():
stop_event = asyncio.Event()
task = asyncio.create_task(worker(stop_event))
try:
    await asyncio.sleep(2)
finally:
    stop_event.set()
    await task
if name == "main":
asyncio.run(main())

If a task may hang, design the timeout and cancellation strategy separately. Do not close the event loop as a way to forcibly stop a running coroutine.

How to Verify the Fix

Run the program with asyncio.run(main(), debug=True).
Perform the scenario that previously triggered the error.
Confirm that the full traceback no longer contains an attempt to access a closed event loop.
Check that no Task was destroyed but it is pending or coroutine was never awaited warnings appear.
Repeat the run or operation several times if the error occurred on repeated calls.
Test abnormal shutdown paths: an exception, task cancellation, and application termination.

Final Checklist

The program has one main coroutine and one call to asyncio.run().
All background tasks are retained and awaited with await, gather(), or a task group.
Asynchronous resources are closed before the main coroutine finishes.
Objects from one event loop are not used in another.
loop.close() is not called for an event loop managed by an external environment.
When managing the loop manually, unfinished tasks are cancelled and processed before the loop is closed.
The fix has been verified with asyncio debug mode enabled.