Skip to content

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

FixDevs · (Updated: )

Part of:  Python Errors

Quick Answer

How to fix Python asyncio RuntimeError no running event loop and event loop already running caused by mixing sync and async code, Jupyter, and wrong loop management.

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 schedule coroutines. The loop is a per-thread object that owns a queue of pending tasks, a selector for I/O readiness, and an executor for blocking work. Only one event loop can run per thread at a time — calling loop.run_until_complete() or asyncio.run() while another is already running raises RuntimeError.

The error space splits into two flavors that look similar but have opposite causes. “No running event loop” means you called a function that needs a loop (like asyncio.get_event_loop() from a thread or asyncio.create_task() outside an async context) but no loop exists in the current thread. “This event loop is already running” means you tried to bootstrap a second loop on top of one that is already spinning — typical inside Jupyter, inside an async def function, or inside an ASGI server worker.

Python 3.10 tightened these rules. Previously, asyncio.get_event_loop() would create a new loop on the fly if none existed. That convenience masked bugs (threads silently getting their own loops, finalizers running after the main loop closed) and was deprecated in 3.10 and removed for the no-loop case in 3.12. Newer code that runs on older Python “just works” but breaks loudly on modern interpreters.

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.

Platform and Environment Differences

asyncio runs on every platform Python runs on, but the default loop implementation, the performance characteristics, and the environments that already own a loop vary widely. The same code can succeed on macOS, deprecation-warn on Linux, and crash on Windows.

Windows: SelectorEventLoop vs ProactorEventLoop. Python 3.7 used SelectorEventLoop on Windows by default. Python 3.8 switched to ProactorEventLoop because the selector loop cannot use subprocesses or pipes on Windows. The proactor loop uses Windows I/O Completion Ports. This change broke older libraries that called Unix-only socket APIs. If you need subprocess support on Windows 3.7, call asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) before asyncio.run. If a library breaks on Windows 3.8+ because it expects the selector loop, force the older policy with asyncio.WindowsSelectorEventLoopPolicy().

Linux and macOS: uvloop. uvloop is a drop-in replacement built on libuv (the same loop Node.js uses). It is 2–4× faster than the stdlib loop for I/O-heavy workloads but does not run on Windows. If your team has mixed laptops (Mac/Linux developers and a Windows colleague), guard the import:

import sys
if sys.platform != "win32":
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

Python 3.10 changes. asyncio.get_event_loop() emits DeprecationWarning when called with no running loop and no current loop. The function still returns a loop (by creating one), but the warning signals you should switch to asyncio.new_event_loop() or asyncio.run().

Python 3.12 changes. asyncio.get_event_loop() now raises DeprecationWarning and will eventually raise RuntimeError instead of creating a loop. The loop parameter was removed from asyncio.sleep, asyncio.wait, and many other functions — they implicitly use the running loop. Code targeting 3.7-era APIs may need updates.

Python 3.14+. The default child-watcher policy on Linux changed to PidfdChildWatcher (kernel 5.4+). Old kernels fall back to ThreadedChildWatcher. If you spawn subprocesses on an old CentOS image, you may see NotImplementedError instead of expected behavior.

Jupyter, IPython, Google Colab. These shells already own an event loop because IPython integrates with Tornado. Calling asyncio.run() inside a Jupyter cell raises “This event loop is already running.” Use top-level await directly (Jupyter 7+ supports it) or install nest_asyncio to monkey-patch the loop to be re-entrant.

ASGI servers (Uvicorn, Hypercorn, Daphne). Each worker owns one event loop. Your handler runs inside that loop. Calling asyncio.run() from inside an async def route is the most common mistake — use await directly. Multi-worker setups give each worker its own loop in its own process; loops do not share state across workers without an external store.

Tornado. Tornado has its own loop abstraction (IOLoop). Modern Tornado (6+) wraps asyncio under the hood, so the same warnings apply. But if you mix Tornado handlers that call IOLoop.current() with asyncio-only code, you can end up with two loops referencing the same thread.

FastAPI / Starlette. Both default to the running loop. Background tasks created via BackgroundTasks run on the same loop after the response is sent. If a background task calls asyncio.run(), it raises “already running” immediately.

Django (async views in 4.1+). Async views run on a loop managed by the ASGI server. Sync views call async code via asgiref.sync.async_to_sync, which spawns a thread with its own loop. Calling asyncio.run from inside async_to_sync’s thread fails because that thread already has a loop assigned.

AWS Lambda / Vercel / Cloudflare Workers. Serverless runtimes that target Python (Lambda, Vercel Python) create a fresh interpreter per cold start. The first invocation has no loop, so asyncio.run(handler()) works. Warm invocations reuse the interpreter — if your handler leaked a closed loop, the next call sees RuntimeError: Event loop is closed. Always create the loop inside the handler, never at module scope.

Containers and threading. A container with --cpus=0.5 produces unpredictable thread scheduling. Coroutines that race against the GIL can deadlock or report “no running event loop” if the loop thread is starved. Increase CPU allocation or use asyncio.get_event_loop_policy().new_event_loop() to control which thread owns the loop.

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.

nest_asyncio monkey-patches the loop to be re-entrant. It does not work with uvloop because uvloop’s C-level loop does not expose the hooks needed. If you use uvloop, switch to top-level await.

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

When you need a reference to the current loop inside a coroutine, use asyncio.get_running_loop() instead of get_event_loop(). The former raises RuntimeError cleanly when no loop is running; the latter has the deprecated implicit-creation behavior.

Still Not Working?

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

Sync LibraryAsync Alternative
requestsaiohttp, httpx
psycopg2asyncpg, psycopg[binary] (v3)
pymongomotor
redis-pyaioredis / redis.asyncio

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)

Check for a closed loop being reused. Lambda warm starts, long-running scripts that call asyncio.run multiple times, and test fixtures that close the loop between tests can leave the current loop slot pointing at a closed loop. The next get_event_loop() finds the closed loop and either returns it (causing “Event loop is closed” on the next await) or refuses to create a new one. Force a fresh loop with asyncio.set_event_loop(asyncio.new_event_loop()) at the start of each invocation.

Check pytest-asyncio config. pytest-asyncio 0.21+ changed the default asyncio_mode to strict, which requires every async test to be marked. Without the mark, the test runs in a thread without a loop and prints “no running event loop” mid-test. Add asyncio_mode = "auto" to pytest.ini or decorate tests with @pytest.mark.asyncio.

Check for signal handlers on Windows. loop.add_signal_handler() raises NotImplementedError on Windows. Wrap calls in try/except NotImplementedError or detect the platform and skip signal-based shutdown.

Check for nested async with mistakes. Forgetting to await an async context manager (async with ... as x: vs with ... as x:) raises RuntimeError with a misleading message about loops. Verify every async resource uses async with.

For Python connection errors, see Fix: Python requests ConnectionError: Max retries exceeded. For import errors, see Fix: Python ModuleNotFoundError: No module named. For mixing async libraries with sync ones, see Fix: Python async/sync mix. For framework-specific async problems, see Fix: FastAPI 422 Unprocessable Entity.

F

FixDevs

Solo developer based in Japan. Every solution is cross-referenced with official documentation and tested before publishing.

Was this article helpful?

Related Articles