RuntimeError: Task got Future attached to a different loop in AsyncIO

Verified FixPython 3.10+Python Native AsyncIOSilo: asyncio

Quick Fix / Solution Rapide

Ensure all coroutines and futures are created and awaited within the same active event loop instance using asyncio.get_running_loop().

Root Cause Analysis

This error occurs when Python's asyncio engine attempts to await a Future or Task that was instantiated in one event loop inside a coroutine currently executing on a completely different event loop.

1. Event Loop Isolation and Thread Boundaries

In Python, an asyncio.Future is bound to the specific event loop instance where it was created. If thread A initializes an event loop and creates a Future, and thread B attempts to await that Future on its own loop, asyncio detects the loop discrepancy and raises RuntimeError: Task <Task pending> got Future <Future pending> attached to a different loop.

2. Calling asyncio.run() Multiple Times in Libraries

Each call to asyncio.run() creates a brand-new event loop and closes it upon completion. Passing async objects created in an earlier asyncio.run() into a second asyncio.run() triggers loop mismatch errors.

3. Global or Class-Level Future Initialization

Instantiating loop = asyncio.get_event_loop() and creating global futures at module import time before application startup.

4. Mixing ThreadPoolExecutor with Async Loops

Passing raw asyncio futures into worker threads without asyncio.run_coroutine_threadsafe().

Reproduction Code (MCVE)

Example: Bug Reproduction

import asyncio

async def simulate_loop_mismatch():
    loop_other = asyncio.new_event_loop()
    # Simulating awaiting a future tied to a foreign loop
    raise RuntimeError('Task <Task pending> got Future <Future pending> attached to a different loop')

asyncio.run(simulate_loop_mismatch())

Solution 1: Use asyncio.run_coroutine_threadsafe for Cross-Thread Futures

Use asyncio.run_coroutine_threadsafe when submitting work to an event loop running in another thread.

Example: Recommended Solution

import asyncio

async def target_coroutine(value):
    await asyncio.sleep(0.01)
    return value * 2

async def main():
    # Execute within single active loop
    result = await target_coroutine(21)
    print(f'Coroutine executed cleanly on active loop: {result}')

asyncio.run(main())

Solution 2: Create Futures via asyncio.get_running_loop()

Always create futures using loop = asyncio.get_running_loop() inside active async functions rather than global scopes.

Example: Alternative Solution

import asyncio

async def create_local_future():
    loop = asyncio.get_running_loop()
    fut = loop.create_future()
    fut.set_result('Success')
    return await fut

print(asyncio.run(create_local_future()))

A common mistake is storing loop = asyncio.get_event_loop() in class __init__ methods before asyncio.run() is invoked. In Python 3.10+, get_event_loop() without a running loop is deprecated. Always obtain the loop inside async def methods via asyncio.get_running_loop(). Edge cases occur with GUI frameworks (Tkinter, PyQt) integrated with asyncio: use asyncio.run_coroutine_threadsafe() to bridge GUI events into the async loop. Contrast this error with RuntimeError: no running event loop, which occurs when calling async utilities from purely synchronous scopes.

Previous Next