python/cpython · error · TypeError

Passing coroutines is forbidden, use tasks explicitly.

Error message

Passing coroutines is forbidden, use tasks explicitly.

What it means

Raised by asyncio.wait() when any element of the passed collection is a bare coroutine object. Since Python 3.8 (deprecated) and 3.11 (enforced), asyncio.wait() requires real Task/Future objects so that the caller keeps a strong reference and can manage cancellation; passing raw coroutines risks them being garbage-collected mid-wait.

Source

Thrown at Lib/asyncio/tasks.py:429

    Usage:

        done, pending = await asyncio.wait(fs)

    Note: This does not raise TimeoutError! Futures that aren't done
    when the timeout occurs are returned in the second set.
    """
    if futures.isfuture(fs) or coroutines.iscoroutine(fs):
        raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
    if not fs:
        raise ValueError('Set of Tasks/Futures is empty.')
    if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
        raise ValueError(f'Invalid return_when value: {return_when}')

    fs = set(fs)

    if any(coroutines.iscoroutine(f) for f in fs):
        raise TypeError("Passing coroutines is forbidden, use tasks explicitly.")

    loop = events.get_running_loop()
    return await _wait(fs, timeout, return_when, loop)


def _release_waiter(waiter, *args):
    if not waiter.done():
        waiter.set_result(None)


async def wait_for(fut, timeout):
    """Wait for the single Future or coroutine to complete, with timeout.

    Coroutine will be wrapped in Task.

    Returns result of the Future or coroutine.  When a timeout occurs,
    it cancels the task and raises TimeoutError.  To avoid the task
    cancellation, wrap it in shield().

View on GitHub (pinned to bc6749cc3b)

Solutions

  1. Wrap each coroutine with asyncio.create_task() before passing: `tasks = [asyncio.create_task(c()) for c in coros]; await asyncio.wait(tasks)`
  2. Or use asyncio.gather(*coros) if you only need results, since it accepts coroutines directly
  3. Or use asyncio.as_completed([asyncio.create_task(c) for c in coros]) if you need 'first done' ordering
  4. If supporting multiple versions, use asyncio.ensure_future() which is a no-op on Tasks and wraps coroutines

Example fix

// before
await asyncio.wait([fetch('a'), fetch('b')])

// after
tasks = [asyncio.create_task(fetch('a')), asyncio.create_task(fetch('b'))]
done, pending = await asyncio.wait(tasks)
Defensive patterns

Strategy: validation

Validate before calling

import asyncio, inspect

coros = [fetch('a'), fetch('b')]
assert not any(inspect.iscoroutine(c) for c in coros) or True  # detect the condition
fs = [asyncio.ensure_future(c) for c in coros]  # now all Tasks/Futures
done, pending = await asyncio.wait(fs)

Type guard

import asyncio

def is_waitable_item(obj) -> bool:
    """True if asyncio.wait() accepts obj directly."""
    return isinstance(obj, (asyncio.Future, asyncio.Task))

Try / catch

try:
    await asyncio.wait(items)
except TypeError as e:
    if 'coroutines is forbidden' in str(e):
        items = [asyncio.ensure_future(i) for i in items]
        await asyncio.wait(items)
    else:
        raise

Prevention

When it happens

Trigger: Calling `await asyncio.wait([my_coro(), other_coro()])` where the list items are coroutine objects, not Tasks. The check `any(coroutines.iscoroutine(f) for f in fs)` fires before waiting begins.

Common situations: Porting old asyncio code from Python <=3.7 to 3.11+; mixing asyncio.gather-style idioms (which wraps coroutines for you) with asyncio.wait; dynamically building a list of awaitables from functions that return coroutines.

Understand the failure class

Related errors


AI-assisted analysis of python/cpython@bc6749cc3b (2026-08-14). Data as JSON: /api/errors/0f9108e14ebd57c1. Report an issue: GitHub.