Short answer: don’t rely on

TaskRunContext.get()

inside helper coroutines. In Prefect 3.4.14 the context is a contextvar and can be lost when you spawn new asyncio tasks, switch threads, or use background tasks. Instead, grab a run logger at the start of your task and pass it into helpers (or pass the captured context and use

get_run_logger(context=...)

). Why you’re seeing None -

TaskRunContext

is stored in a contextvar. Python only guarantees propagation within the same execution context. Spawning work with

asyncio.create_task

, running things in a thread/executor, or using background tasks (

.delay()

) can run in a different context, so

TaskRunContext.get()

returns

None

. - Even with

asyncio.gather

, if you’ve created new tasks (directly or indirectly) the context may not be present unless it was captured at task creation. Recommended patterns 1) Pass a Prefect logger down

from prefect import task, get_run_logger
from prefect.utilities.asyncio import gather  # Prefer Prefect's helpers

async def helper(logger, i):
    <http://logger.info|logger.info>(f"helper {i} running")

@task
async def my_task():
    logger = get_run_logger()
    await gather(*(helper(logger, i) for i in range(3)))

2) If you need the context, capture once and pass it

from prefect import task, get_run_logger
from prefect.context import TaskRunContext

async def needs_ctx(ctx):
    logger = get_run_logger(context=ctx)
    <http://logger.info|logger.info>("logging with explicit context")

@task
async def my_task():
    ctx = TaskRunContext.get()  # valid here
    await needs_ctx(ctx)

3) Prefer Prefect’s asyncio utilities - Use

prefect.utilities.asyncio.gather

and

prefect.utilities.asyncio.create_task

which preserve Prefect context better than raw

asyncio

in some cases:

from prefect.utilities.asyncio import create_task, gather

4) If you must use raw asyncio.create_task, copy the context explicitly

import asyncio, contextvars

async def helper():
    # will see TaskRunContext if copied explicitly
    ...

ctx = contextvars.copy_context()
t = asyncio.create_task(ctx.run(helper))
await t

Background tasks (.delay) and logging - A call to

some_task.delay(...)

launches a separate task run (often on a task worker). Its logs belong to that new task run, not the parent task. Use

get_run_logger()

inside the background task function itself; do not try to reuse the parent’s context/logger. - If you want to correlate logs between parent and children, include the parent flow run id or another correlation id in the child’s logs or emit an artifact/event from each child. Gotchas that drop context -

asyncio.create_task

without context copying -

asyncio.to_thread

/

loop.run_in_executor

(threads don’t inherit contextvars) - Prefer

prefect.utilities.asyncio.run_sync_in_worker_thread

, or pass the logger/context explicitly. - Using

.delay()

(separate task run) and expecting logs to land on the parent Docs and references - Background tasks: Background tasks in Prefect 3 - Runtime/context concepts: Run context in Prefect 3 -

get_run_logger

API: Add logging in flows and tasks If you can share a minimal snippet (how you’re calling the helper, whether you use

asyncio.create_task

,

asyncio.gather

, threads, or

.delay()

), plus your Python version, I can pinpoint exactly where the context is being lost and suggest the smallest fix.