Yes. In Prefect 3.x you can watch a deployment run’s logs from Python in two ways: - Real-time streaming over WebSocket (best for “tail -f” style watching) - Polling the REST API (simple and stateless) Below are minimal, ready-to-run examples for both. Set

PREFECT_API_URL

and (for Cloud)

PREFECT_API_KEY

in your environment. Option A — Stream logs in real time (recommended)

import asyncio
import logging
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import LogFilter, LogFilterFlowRunId, LogFilterLevel
from prefect.logging.clients import get_logs_subscriber

async def stream_run_logs(flow_run_id: UUID):
    # Only INFO and above; remove level filter for all logs
    log_filter = LogFilter(
        flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]),
        level=LogFilterLevel(ge_=<http://logging.INFO|logging.INFO>),
    )
    async with get_logs_subscriber(filter=log_filter) as sub:
        async for log in sub:
            level = logging.getLevelName(log.level)
            print(f"{log.timestamp} | {level:<7} | {log.message}")

# If you need to first kick off a deployment run and then stream:
async def start_and_stream(deployment_id: UUID):
    async with get_client() as client:
        fr = await client.create_flow_run_from_deployment(deployment_id)
        print(f"Started flow run {fr.id}")
    await stream_run_logs(fr.id)

# Usage:
# asyncio.run(stream_run_logs(UUID("YOUR-FLOW-RUN-ID")))
# or
# asyncio.run(start_and_stream(UUID("YOUR-DEPLOYMENT-ID")))

Option B — Poll logs via REST (simple, works anywhere)

import asyncio
import logging
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import LogFilter, LogFilterFlowRunId
from prefect.client.schemas.sorting import LogSort

async def poll_run_logs(flow_run_id: UUID, interval=2.0, page_size=200):
    log_filter = LogFilter(flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]))
    offset = 0
    async with get_client() as client:
        while True:
            logs = await client.read_logs(
                log_filter=log_filter,
                limit=page_size,
                offset=offset,
                sort=LogSort.TIMESTAMP_ASC,
            )
            if logs:
                for log in logs:
                    level = logging.getLevelName(log.level)
                    print(f"{log.timestamp} | {level:<7} | {log.message}")
                offset += len(logs)
            else:
                await asyncio.sleep(interval)

# Usage:
# asyncio.run(poll_run_logs(UUID("YOUR-FLOW-RUN-ID")))

Tip: To stop when the flow run finishes, poll

client.read_flow_run(flow_run_id)

in a small parallel loop and break when the state is terminal (Completed, Failed, or Cancelled). References (source): - Prefect logs subscriber (WebSocket): clients.py - Log filters and models: filters.py, objects.py - Orchestration client (read_logs, create_flow_run_from_deployment): orchestration.py - Server log APIs: logs.py If you share whether you’re on Prefect Cloud or OSS and how you kick off runs (Python vs CLI), I can tailor this into a drop-in helper that blocks until completion.

You’re right — the simple polling loop will run forever unless you add a termination condition. The usual pattern is: - Poll logs - Poll the flow run state - Exit when the flow run is terminal and there haven’t been any new logs for a short grace period Here’s a drop-in version of Option B that does exactly that:

import asyncio
import logging
from uuid import UUID
from datetime import datetime, timezone, timedelta

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import LogFilter, LogFilterFlowRunId
from prefect.client.schemas.sorting import LogSort

TERMINAL_STATES = {"COMPLETED", "FAILED", "CANCELLED", "CRASHED"}

async def poll_run_logs_until_done(
    flow_run_id: UUID,
    interval=2.0,
    idle_grace_iters=2,        # how many empty polls after terminal state before exiting
    time_grace_seconds=3,      # or exit if last log is older than this after terminal state
    page_size=200,
):
    log_filter = LogFilter(flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]))
    offset = 0
    idle = 0
    last_ts = None

    async with get_client() as client:
        while True:
            # 1) Pull any new logs
            logs = await client.read_logs(
                log_filter=log_filter,
                limit=page_size,
                offset=offset,
                sort=LogSort.TIMESTAMP_ASC,
            )

            if logs:
                for log in logs:
                    level = logging.getLevelName(log.level)
                    print(f"{log.timestamp} | {level:<7} | {log.message}")
                offset += len(logs)
                last_ts = logs[-1].timestamp
                idle = 0
            else:
                idle += 1

            # 2) Check run state
            fr = await client.read_flow_run(flow_run_id)
            state_type = getattr(fr.state.type, "value", str(fr.state.type))
            is_terminal = state_type in TERMINAL_STATES

            # 3) Decide if we're done
            done = False
            if is_terminal:
                # a) no new logs for N polls
                if idle >= idle_grace_iters:
                    done = True
                # b) or last log is sufficiently old
                if last_ts is not None:
                    age = (datetime.now(timezone.utc) - last_ts).total_seconds()
                    if age >= time_grace_seconds:
                        done = True

            if done:
                print(f"Flow run finished with state {state_type}. Exiting log tail.")
                break

            await asyncio.sleep(interval)

Notes: - Using offset keeps pagination simple. The small grace handles delayed/late-arriving logs. - If you prefer not to manage

offset

, you can advance a timestamp filter instead; just use

LogFilterTimestamp(after_=last_ts)

and update

last_ts

as you go. Source references: - Client methods and models: orchestration.py - Log filters/sorting: filters.py, sorting.py If you want, I can adapt this to start a deployment, tail logs until completion, and return the final state/summary.