What causes "TaskGroup {self!r} has already been entered" in python/cpython?

Construct a fresh TaskGroup per use: 'async with asyncio.TaskGroup() as tg:' inside the function that runs each batch

python/cpython · error · RuntimeError

TaskGroup {self!r} has already been entered

Error message

TaskGroup {self!r} has already been entered

What it means

RuntimeError raised by TaskGroup.__aenter__ when the same TaskGroup instance is entered a second time (_entered already true). A TaskGroup is a one-shot async context manager tied to a single parent task and lifecycle; it cannot be reused for a second batch of work.

Source

Thrown at Lib/asyncio/taskgroups.py:58

        self._cancel_on_enter = False

    def __repr__(self):
        info = ['']
        if self._tasks:
            info.append(f'tasks={len(self._tasks)}')
        if self._errors:
            info.append(f'errors={len(self._errors)}')
        if self._aborting:
            info.append('cancelling')
        elif self._entered:
            info.append('entered')

        info_str = ' '.join(info)
        return f'<TaskGroup{info_str}>'

    async def __aenter__(self):
        if self._entered:
            raise RuntimeError(
                f"TaskGroup {self!r} has already been entered")
        if self._loop is None:
            self._loop = events.get_running_loop()
        self._parent_task = tasks.current_task(self._loop)
        if self._parent_task is None:
            raise RuntimeError(
                f'TaskGroup {self!r} cannot determine the parent task')
        self._entered = True
        if self._cancel_on_enter:
            self.cancel()

        return self

    async def __aexit__(self, et, exc, tb):
        tb = None
        try:
            return await self._aexit(et, exc)
        finally:

View on GitHub (pinned to bc6749cc3b)

Solutions

Construct a fresh TaskGroup per use: 'async with asyncio.TaskGroup() as tg:' inside the function that runs each batch
If you need a long-lived spawn point, use a dedicated worker task plus an asyncio.Queue instead of a persistent TaskGroup
Never share a TaskGroup across concurrent 'async with' blocks

Example fix

// before
class Runner:
    def __init__(self):
        self.tg = asyncio.TaskGroup()
    async def run(self, fn):
        async with self.tg:            # second call -> RuntimeError
            self.tg.create_task(fn())

// after
class Runner:
    async def run(self, fn):
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fn())

Defensive patterns

Strategy: validation

Validate before calling

null

Type guard

null

Try / catch

null

Prevention

Always write 'async with asyncio.TaskGroup() as tg:' at the scope that owns one batch — never store and re-enter
If a class needs repeated fan-out, construct the group inside the method
For long-lived spawn points use a worker task plus a queue, not a persistent group

When it happens

Trigger: Storing a TaskGroup as an instance attribute and using 'async with self.tg:' in a method called twice; entering the same group in two places (nested or sequential); re-entering after an exception aborted the first run.

Common situations: Framework/service classes that create one TaskGroup in __init__ and re-enter per request; retry wrappers that re-run an 'async with tg' block; refactoring a per-call group into a shared field for convenience.

Related errors

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