Fix: Python RuntimeError: no running event loop / This event loop is already running

The Error

You run async Python code and get:

RuntimeError: no current event loop

Or variations:

RuntimeError: There is no current event loop in thread 'Thread-1'.

RuntimeError: This event loop is already running

RuntimeError: cannot be called from a running event loop

DeprecationWarning: There is no current event loop

Python’s asyncio event loop is either not running when you need it, already running when you try to start another one, or you are calling async code from the wrong context.

Why This Happens

Python’s asyncio uses an event loop to run asynchronous code. Only one event loop can run per thread at a time. Common situations that cause errors:

• Calling asyncio.run() from inside an already-running loop (Jupyter notebooks, some web frameworks).
• Using asyncio.get_event_loop() in a thread where no loop exists.
• Mixing sync and async code incorrectly.
• Running async code in a background thread without creating a loop for that thread.
• Python 3.10+ deprecation of implicit loop creation.

Fix 1: Use asyncio.run() Correctly

asyncio.run() is the standard entry point for async code:

Broken — calling asyncio.run() inside an existing loop:

import asyncio

async def inner():
    return "hello"

async def outer():
    result = asyncio.run(inner())  # RuntimeError: This event loop is already running!
    return result

asyncio.run(outer())

Fixed — just await the coroutine:

async def outer():
    result = await inner()  # Use await, not asyncio.run()
    return result

asyncio.run(outer())  # Only one asyncio.run() at the top level

Rule: Use asyncio.run() only once at the top level of your program. Inside async functions, use await.

async def main():
    result1 = await fetch_data()
    result2 = await process_data(result1)
    return result2

# Single entry point
if __name__ == "__main__":
    asyncio.run(main())

Pro Tip: Think of asyncio.run() as the bridge between synchronous and asynchronous worlds. You cross this bridge once at program start. After that, everything is async and uses await.

Fix 2: Fix “No Current Event Loop” in Threads

Each thread needs its own event loop:

Broken:

import asyncio
import threading

async def fetch():
    await asyncio.sleep(1)
    return "done"

def thread_func():
    loop = asyncio.get_event_loop()  # RuntimeError in Python 3.10+
    result = loop.run_until_complete(fetch())

Fixed — create a new loop for the thread:

def thread_func():
    result = asyncio.run(fetch())  # Creates a new loop for this thread

thread = threading.Thread(target=thread_func)
thread.start()
thread.join()

Fixed — use asyncio.new_event_loop() explicitly:

def thread_func():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        result = loop.run_until_complete(fetch())
    finally:
        loop.close()

Best practice — use asyncio.run() in each thread:

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def async_task(n):
    await asyncio.sleep(0.1)
    return n * 2

def run_in_thread(n):
    return asyncio.run(async_task(n))

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(run_in_thread, range(10)))

Fix 3: Fix Jupyter Notebook Issues

Jupyter already runs an event loop. asyncio.run() conflicts with it:

Broken in Jupyter:

import asyncio

async def fetch():
    return "data"

asyncio.run(fetch())  # RuntimeError: This event loop is already running

Fixed — use await directly in Jupyter:

# Jupyter cells support top-level await
result = await fetch()
print(result)

Fixed — use nest_asyncio if top-level await does not work:

import nest_asyncio
nest_asyncio.apply()

import asyncio
result = asyncio.run(fetch())  # Works now

Fixed — use the existing loop:

loop = asyncio.get_event_loop()
result = loop.run_until_complete(fetch())

Common Mistake: Using asyncio.run() in Jupyter notebooks, Google Colab, or IPython. These environments already have a running event loop. Use await directly or install nest_asyncio.

Fix 4: Fix Mixing Sync and Async Code

Calling async functions from synchronous code:

Broken — calling coroutine without await:

async def get_data():
    return "data"

# Wrong — returns a coroutine object, not the result
result = get_data()
print(result)  # <coroutine object get_data at 0x...>

Fixed — use asyncio.run():

result = asyncio.run(get_data())
print(result)  # "data"

Calling sync functions from async code (blocking the loop):

import asyncio
import time

def slow_sync_function():
    time.sleep(5)  # Blocks the entire event loop!
    return "done"

async def main():
    # Wrong — blocks the loop
    result = slow_sync_function()

    # Fixed — run in a thread pool
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, slow_sync_function)

Using asyncio.to_thread() (Python 3.9+):

async def main():
    result = await asyncio.to_thread(slow_sync_function)

Fix 5: Fix Web Framework Integration

Different web frameworks handle async differently:

Flask (synchronous by default):

from flask import Flask
import asyncio

app = Flask(__name__)

async def async_fetch():
    await asyncio.sleep(0.1)
    return "data"

@app.route("/data")
def get_data():
    # Flask routes are sync — use asyncio.run()
    result = asyncio.run(async_fetch())
    return result

FastAPI (async native):

from fastapi import FastAPI

app = FastAPI()

@app.get("/data")
async def get_data():
    result = await async_fetch()  # Just use await
    return {"data": result}

Django (async views in Django 4.1+):

# Async view
async def my_view(request):
    data = await async_fetch()
    return JsonResponse({"data": data})

# Calling async from sync Django code
from asgiref.sync import async_to_sync

def sync_view(request):
    data = async_to_sync(async_fetch)()
    return JsonResponse({"data": data})

Fix 6: Fix asyncio.gather() and Task Issues

Running multiple coroutines concurrently:

Broken — awaiting sequentially when you want concurrency:

async def main():
    result1 = await fetch_url("https://api1.example.com")
    result2 = await fetch_url("https://api2.example.com")
    result3 = await fetch_url("https://api3.example.com")
    # Total time: sum of all three (sequential!)

Fixed — use asyncio.gather():

async def main():
    result1, result2, result3 = await asyncio.gather(
        fetch_url("https://api1.example.com"),
        fetch_url("https://api2.example.com"),
        fetch_url("https://api3.example.com"),
    )
    # Total time: max of the three (concurrent!)

Handle errors in gather:

results = await asyncio.gather(
    fetch_url("https://api1.example.com"),
    fetch_url("https://api2.example.com"),
    return_exceptions=True,  # Returns exceptions instead of raising
)

for result in results:
    if isinstance(result, Exception):
        print(f"Error: {result}")
    else:
        process(result)

Using TaskGroups (Python 3.11+):

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_url("https://api1.example.com"))
        task2 = tg.create_task(fetch_url("https://api2.example.com"))

    result1 = task1.result()
    result2 = task2.result()

Fix 7: Fix Timeout and Cancellation

Set timeouts on async operations:

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
    except asyncio.TimeoutError:
        print("Operation timed out")

Python 3.11+ timeout context:

async def main():
    async with asyncio.timeout(5.0):
        result = await slow_operation()

Handle task cancellation:

async def cancellable_task():
    try:
        while True:
            await asyncio.sleep(1)
            do_work()
    except asyncio.CancelledError:
        print("Task was cancelled, cleaning up...")
        cleanup()
        raise  # Re-raise to properly cancel

Fix 8: Fix Python Version-Specific Issues

Python 3.10+ removed implicit loop creation:

# Before Python 3.10 — worked but deprecated
loop = asyncio.get_event_loop()  # Created a loop if none existed

# Python 3.10+ — raises DeprecationWarning or RuntimeError
# Fixed:
asyncio.run(main())  # Preferred way
# Or:
loop = asyncio.new_event_loop()

Python 3.12+ changes:

# asyncio.get_event_loop() in a non-async context raises RuntimeError
# Always use asyncio.run() instead

Still Not Working?

Check for library compatibility. Some libraries (like requests) are synchronous. Use async alternatives:

Debug event loop state:

loop = asyncio.get_event_loop()
print(f"Running: {loop.is_running()}")
print(f"Closed: {loop.is_closed()}")

Check for unclosed resources:

import warnings
warnings.filterwarnings("error", category=ResourceWarning)

For Python connection errors, see Fix: Python requests ConnectionError: Max retries exceeded. For import errors, see Fix: Python ModuleNotFoundError: No module named.