Got it — that’s a lot of fields. By default, Prefect enriches Python log records with extra context (flow/task IDs, etc.). You can’t change what Prefect sends to Prefect Cloud/Server (those structured fields are used for correlation), but you can absolutely simplify what’s emitted to your own stdout/file logs by customizing the logging handlers/formatters. A couple quick questions to make sure I point you the right way: - Are you trying to simplify what you see on stdout/your log shipper, or do you want to change what appears in Prefect Cloud/Server? - Are you outputting JSON logs intentionally, or do you just want a simpler text line format? - Which Prefect version are you on? (I’ll assume 3.x unless you say otherwise.) In the meantime, here are two practical ways to trim fields for your local/console logs without breaking logs that go to Prefect Cloud. Option A: Simpler text logs (console) - Keep Prefect’s API/Cloud handler intact, only change the StreamHandler(s) to a simple format.

import logging

def simplify_prefect_console_logs():
    prefect_logger = logging.getLogger("prefect")

    simple_fmt = logging.Formatter("%(asctime)s | %(levelname)s | %(name)s | %(message)s")

    for h in prefect_logger.handlers:
        # Only change console-like handlers; leave API/other handlers alone
        if isinstance(h, logging.StreamHandler):
            h.setFormatter(simple_fmt)

simplify_prefect_console_logs()

Option B: JSON logs with only selected keys - Attach a custom formatter that whitelists the fields you care about. This will prevent the “kitchen sink” dict you’re seeing.

import json
import logging

class WhitelistJsonFormatter(logging.Formatter):
    def __init__(self, fields):
        super().__init__()
        self.fields = set(fields)

    def format(self, record):
        # Start with the basics
        data = {
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "time": getattr(record, "asctime", None),
        }
        # Add any whitelisted extras that exist on the record
        for k in self.fields:
            if hasattr(record, k):
                data[k] = getattr(record, k)
        return json.dumps({k: v for k, v in data.items() if v is not None}, ensure_ascii=False)

def simplify_prefect_json_console_logs():
    prefect_logger = logging.getLogger("prefect")
    formatter = WhitelistJsonFormatter(fields={
        # Add only what you want to keep; examples:
        "flow_run_id",
        "flow_run_name",
        "task_run_id",
        "task_run_name",
    })
    for h in prefect_logger.handlers:
        if isinstance(h, logging.StreamHandler):
            h.setFormatter(formatter)

simplify_prefect_json_console_logs()

Notes and gotchas: - Put this setup code early (before your flows start) so it applies to all emitted logs. - Don’t blindly clear all handlers on the “prefect” logger; one of them forwards logs to Prefect Cloud/Server. The snippets above only change StreamHandler(s), leaving the Cloud handler intact. - If you’re running in a worker, ensure this code runs in the same process that emits logs (e.g., at module import time in your flow code or in a small bootstrap module you import first). If you can share: - Your Prefect version - Whether you want plain text or JSON on stdout - Which fields you actually want to keep …I can tailor a drop-in snippet for your exact setup.

Perfect — Prefect 3.x + JSON to stdout, with only the “readable” fields. There isn’t a built-in env var to drop fields from Prefect’s JSON logs to stdout, but you can override just the console handler’s formatter and whitelist the fields you care about. This won’t affect what gets sent to Prefect Cloud/Server. Drop-in snippet (put this at the top of your flow code so it runs before logs start):

import sys
import json
import logging
from datetime import datetime, timezone

class MinimalJsonFormatter(logging.Formatter):
    def __init__(self, extra_fields=None):
        super().__init__()
        self.extra_fields = list(extra_fields or [])

    def format(self, record):
        payload = {
            "timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        # Whitelist Prefect context you want to keep
        for k in self.extra_fields:
            v = getattr(record, k, None)
            if v is not None:
                payload[k] = v
        return json.dumps(payload, ensure_ascii=False)

def configure_minimal_json_stdout(
    keep_fields=("flow_run_id", "flow_run_name", "task_run_id", "task_run_name")
):
    logger = logging.getLogger("prefect")

    # Replace only console handlers; leave the API handler (Cloud/Server) untouched
    for h in list(logger.handlers):
        if isinstance(h, logging.StreamHandler):
            logger.removeHandler(h)

    sh = logging.StreamHandler(stream=sys.stdout)
    sh.setFormatter(MinimalJsonFormatter(extra_fields=keep_fields))
    logger.addHandler(sh)

    # Avoid duplicate logs via propagation to the root logger
    logger.propagate = False

configure_minimal_json_stdout()

What this does: - Emits JSON to stdout with only: timestamp, level, logger, message, and the extra Prefect fields you explicitly allow. - Keeps Prefect’s API handler intact so Cloud/Server logging still has all the context. - Prevents duplicate lines by disabling propagation. If you prefer not to remove handlers, you can just reformat existing console handlers:

logger = logging.getLogger("prefect")
for h in logger.handlers:
    if isinstance(h, logging.StreamHandler):
        h.setFormatter(MinimalJsonFormatter(extra_fields=("flow_run_id","task_run_id")))

Example output:

{
  "timestamp": "2025-01-05T13:00:12.345678+00:00",
  "level": "INFO",
  "logger": "prefect.flow_runs",
  "message": "Beginning flow run 'encouraging-stoat' for flow 'run_daily_ingest_orchestrator'",
  "flow_run_id": "3a1b7ea0-20e8-47cb-8838-7fd74aa888cf",
  "flow_run_name": "encouraging-stoat"
}

Notes: - Place this code in the same module that defines/starts your flow so it executes in the flow run process (workers will import it when they load your flow). - If your log shipper expects stderr, change

stream=sys.stdout

to

sys.stderr

. If you list the exact fields you consider “useful,” I can tailor the

keep_fields

for you. Common picks:

flow_run_id

,

flow_run_name

,

task_run_id

,

task_run_name

, and sometimes `flow_name`/`task_name`.