Short answer: you can’t. In Prefect 3.x, logs are immutable—there’s no UI or API to edit or delete individual log records after a run finishes. The only write operations supported are creating new logs and reading/filtering existing logs. What you can do instead: - Post a corrective log to the same run (recommended)

from prefect import flow
  from prefect.logging import get_run_logger

  @flow
  def my_flow():
      logger = get_run_logger()
      <http://logger.info|logger.info>("Initial message that needs correction")
      # Later, add a corrective note
      logger.warning("CORRECTION: The prior message was inaccurate. See details above/below.")

- Add an event/annotation to the run for clear context

from prefect import flow
  from prefect.events import emit_event
  from prefect.runtime import flow_run

  @flow
  def my_flow():
      # ...
      emit_event(
          event="flow.correction",
          resource={"prefect.resource.id": str(flow_run.id)},
          payload={"note": "Adjusted totals; previous log message was misleading"}
      )

- Delete the entire flow run (cascades and removes all its logs)

import asyncio
  from prefect.client.orchestration import get_client

  async def delete_run(flow_run_id):
      async with get_client() as client:
          await client.delete_flow_run(flow_run_id)

  # asyncio.run(delete_run("<your-flow-run-id>"))

- If your goal is cleanup, you can also filter at read-time and only display the levels/messages you want. Docs and references: - Add logging to workflows - REST API: Create logs - REST API: Read logs - Custom events - Database maintenance and cascading deletes If the reason is that sensitive data landed in logs, I recommend: 1) delete the run, 2) rotate any exposed secrets, and 3) update your logging to avoid future leakage (e.g., avoid logging raw payloads, add redaction). If you can share whether you’re on Cloud or self-hosted and your specific goal (fix a message, remove sensitive data, reduce noise), I can suggest a targeted approach.

Yes—with a caveat. - To watch a run you’re starting now: use the watch flag when you launch the run. This streams logs (and related events) in real time until the run finishes.

prefect deployment run <FLOW_NAME>/<DEPLOYMENT_NAME> --watch

- For an already-running flow run: there isn’t a built-in “attach and follow” command. The CLI command below shows a snapshot (you can tail the latest N logs), but it does not stream:

prefect flow-run logs <FLOW_RUN_ID> --tail --num-logs 100

- FYI:

prefect events stream

shows events only, not logs. If you need true real-time streaming for an existing run from the terminal, a small Python helper using Prefect’s log subscriber can do it; I’m happy to share a ready-to-run snippet if that’d help.

Absolutely—here’s a minimal, ready-to-run example that streams logs for a specific flow run in real time using Prefect’s log subscriber. Notes: - Works with your current Prefect profile (PREFECT_API_URL, PREFECT_API_KEY). Make sure you’re logged in/pointed at the right workspace. - Filters logs by

flow_run_id

and minimum log level.

# tail_logs.py
import asyncio
import sys
from uuid import UUID
from datetime import timezone

from prefect.logging.clients import PrefectLogsSubscriber
from prefect.client.schemas.filters import (
    LogFilter,
    LogFilterFlowRunId,
    LogFilterLevel,
)

LEVELS = {10: "DEBUG", 20: "INFO", 30: "WARNING", 40: "ERROR", 50: "CRITICAL"}

async def tail_flow_run_logs(flow_run_id: str, min_level: int = 10):
    # Validate UUID
    fid = UUID(flow_run_id)

    # Build a filter: this limits the stream to a single flow run and min level
    log_filter = LogFilter(
        flow_run_id=LogFilterFlowRunId(any_=[fid]),
        level=LogFilterLevel(ge_=min_level),
    )

    # Connect and stream logs as they arrive
    async with PrefectLogsSubscriber(filter=log_filter) as sub:
        print(f"Connected. Streaming logs for flow_run_id={fid} (>= {LEVELS.get(min_level, min_level)})")
        async for log in sub:
            level = LEVELS.get(log.level, str(log.level))
            ts = log.timestamp.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] + "Z"
            name = log.name
            msg = log.message
            print(f"{ts} [{level}] {name}: {msg}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python tail_logs.py <FLOW_RUN_ID> [MIN_LEVEL]")
        sys.exit(1)

    flow_run_id = sys.argv[1]
    min_level = int(sys.argv[2]) if len(sys.argv) > 2 else 10
    asyncio.run(tail_flow_run_logs(flow_run_id, min_level))

Run it:

python tail_logs.py 00000000-0000-0000-0000-000000000000 20

Optional (Prefect Cloud explicit): If you prefer to pass your API key explicitly, swap the subscriber with `PrefectCloudLogsSubscriber`:

from prefect.logging.clients import PrefectCloudLogsSubscriber
import os

# ...
async with PrefectCloudLogsSubscriber(
    filter=log_filter,
    api_key=os.environ.get("PREFECT_API_KEY"),
) as sub:
    async for log in sub:
        ...

Other useful filters: - Tail a specific task run instead of the whole flow run - Filter by text search (e.g., only messages containing certain keywords)

from prefect.client.schemas.filters import LogFilterTaskRunId, LogFilterTextSearch

log_filter = LogFilter(
    task_run_id=LogFilterTaskRunId(any_=[UUID("<TASK_RUN_ID>")]),
    # Text search: require all words (all_) or any word (any_)
    text_search=LogFilterTextSearch(all_=["error", "timeout"]),
)

Docs: - Add logging to workflows If you want an interleaved stream of both events and logs (state changes + messages), I can share an example using the FlowRunSubscriber as well.

FlowRunSubscriber is an async helper that streams a single flow run’s activity in real time by interleaving two live feeds: - events (state changes, pauses, cancellations, etc.) - logs (your logger output) It gives you one async iterator that yields both items as they arrive so you can “watch” a run end-to-end in your terminal or code. It’s what powers the CLI’s --watch behavior for deployments. Key points: - Where it lives:

prefect.events.subscribers.FlowRunSubscriber

- What it yields: either an Event (from

prefect.events

) or a Log (from

prefect.client.schemas.objects

) - How it works: opens two WebSocket subscriptions (events and logs), merges them into a single queue, and yields items as they come in - Stop conditions: - When a terminal state event arrives (COMPLETED/FAILED/CANCELLED/CRASHED), it keeps listening a bit longer for “straggler” logs, then stops - Or when both streams end - Tunables: -

straggler_timeout

(seconds to wait for late-arriving logs after terminal event; default 3) -

reconnection_attempts

(WebSocket retries; default 10) Minimal example:

import asyncio
from uuid import UUID
from datetime import timezone

from prefect.events.subscribers import FlowRunSubscriber
from prefect.events import Event
from prefect.client.schemas.objects import Log

async def watch_flow_run(flow_run_id: str):
    async with FlowRunSubscriber(
        flow_run_id=UUID(flow_run_id),
        straggler_timeout=5,        # wait up to 5s for late logs after terminal event
        reconnection_attempts=15,   # retry WS connections if they drop
    ) as sub:
        async for item in sub:
            if isinstance(item, Event):
                ts = item.occurred.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
                print(f"{ts} [EVENT] {item.event}")
            else:  # it's a Log
                ts = item.timestamp.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
                print(f"{ts} [LOG] {item.level} {item.name}: {item.message}")

if __name__ == "__main__":
    # Replace with your flow run id
    asyncio.run(watch_flow_run("00000000-0000-0000-0000-000000000000"))

When to use it: - You want the same experience as

prefect deployment run ... --watch

, but targeted to an existing flow run from Python - You need both logs and lifecycle events (not just logs) If you only need logs, prefer the logs-only subscriber from earlier (

PrefectLogsSubscriber

) to reduce noise and overhead.