Short answer: set the console handler to use the JSON formatter. Minimum env var to enable JSON output on stdout/stderr:

PREFECT_LOGGING_HANDLERS_CONSOLE_FORMATTER=json

Notes and useful options: - Set this in the environment where the flow run process executes (e.g., your CodeBuild container or job env). It will JSON-format both Prefect and your logger output on the console. - If you want print() statements to be captured as JSON logs too:

PREFECT_LOGGING_LOG_PRINTS=true

- If you ever see ANSI color codes mixed in (rare with the JSON formatter), disable them:

PREFECT_LOGGING_COLORS=false

- Keep the default compact JSON (single line). “Pretty” formatting is possible via a custom logging config, but that’s usually undesirable in CI. Example (CodeBuild buildspec env section or shell):

export PREFECT_LOGGING_HANDLERS_CONSOLE_FORMATTER=json
export PREFECT_LOGGING_LOG_PRINTS=true

Would you also like worker/infra logs (not just flow/task logs) in JSON in your environment? If so, use the same env vars in the worker/agent process too.

Short answer: there isn’t a built-in setting to “globally” add extras to every Prefect log record.

get_run_logger(**extras)

creates a new adapter each time and only applies the extras you pass on that call. Good ways to avoid passing extras everywhere: Option A — Small wrapper with a ContextVar (simple, no logging plumbing) - Define a tiny utility you import instead of Prefect’s

get_run_logger

. Set your metadata once at flow start; every later call picks it up automatically.

# logctx.py
from contextvars import ContextVar
from prefect.logging import get_run_logger as _prefect_get_run_logger

_LOG_EXTRAS = ContextVar("PREFECT_LOG_EXTRAS", default={})

def set_log_extras(**extras):
    cur = _LOG_EXTRAS.get().copy()
    cur.update(extras)
    _LOG_EXTRAS.set(cur)

def get_run_logger(context=None, **kwargs):
    merged = {*_LOG_EXTRAS.get(), *kwargs}
    return _prefect_get_run_logger(context=context, **merged)

Usage:

from prefect import flow, task
from logctx import set_log_extras, get_run_logger

@flow
def my_flow(request_id: str):
    set_log_extras(request_id=request_id, environment="prod")
    logger = get_run_logger()
    <http://logger.info|logger.info>("flow started")  # includes request_id + environment

    @task
    def do_work():
        tlog = get_run_logger()
        <http://tlog.info|tlog.info>("task running")  # includes the same metadata

    do_work()

- Works well with the default thread-based task runner. - If you switch to a process/distributed runner, ContextVars won’t cross process boundaries; you’ll need to call

set_log_extras(...)

in each process (e.g., at the top of each task) or pass the values to tasks and call

set_log_extras

inside them. Option B — Add a logging.Filter to inject metadata into every LogRecord (global) - Install a filter on Prefect’s loggers/handlers. Any extra attributes you add will be included in JSON logs automatically; for text formatters, you must reference them in the format string.

import logging
from prefect.logging.configuration import ensure_logging_setup

class GlobalMetadataFilter(logging.Filter):
    def __init__(self, **meta):
        super().__init__()
        self.meta = meta
    def filter(self, record: logging.LogRecord) -> bool:
        for k, v in self.meta.items():
            setattr(record, k, v)
        return True

def install_global_log_metadata(**meta):
    ensure_logging_setup()
    flt = GlobalMetadataFilter(**meta)
    for name in ("prefect.flow_runs", "prefect.task_runs"):
        lg = logging.getLogger(name)
        lg.addFilter(flt)
        for h in lg.handlers:
            h.addFilter(flt)

# In your flow:
# install_global_log_metadata(request_id="...", environment="prod")

Caveats: - This applies within the current Python process. For process/distributed task runners, install the filter in each worker/process (e.g., at task start). - With JSON output (

PREFECT_LOGGING_HANDLERS_CONSOLE_FORMATTER=json

), your custom fields appear automatically. With text formatters, update the format string to include your fields if you want them rendered. If your goal is to see this metadata in Prefect Cloud/API logs as well as the console, both approaches work: the extra fields are included in the structured payload Prefect sends to the API. If you prefer the lightest touch for a single-process CI job (like AWS CodeBuild), Option A is usually the cleanest: call

set_log_extras(...)

once in the flow, then use your wrapper

get_run_logger()

everywhere without repeating extras.