Yes. After

run_deployment

you can grab the created flow run’s ID and either: - stream logs live (programmatically), or - fetch logs in batches, or - use the CLI to tail them. Key note:

run_deployment

returns a

FlowRun

object (including

id

), so you can filter logs by that ID. CLI (quickest) - Verified command:

prefect flow-run logs <FLOW_RUN_ID> --tail -n 200

Use

--head/--tail

and

-n

to control how many logs are shown. Programmatic — live streaming (async) - Uses Prefect’s logs subscriber (websocket) for real-time logs.

import asyncio
import logging
from prefect.deployments import arun_deployment
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import LogFilter
from prefect.logging.clients import get_logs_subscriber

async def stream_flow_run_logs(flow_run_id: str):
    lf = LogFilter(flow_run_id={"any_": [flow_run_id]})
    async with get_logs_subscriber(filter=lf) as sub:
        async for log in sub:
            level = logging.getLevelName(log.level)
            print(f"{log.timestamp} | {level:7} | {log.message}")

async def wait_until_done(flow_run_id: str, poll_seconds: float = 3):
    async with get_client() as client:
        while True:
            fr = await client.aread_flow_run(flow_run_id)
            if fr.state and fr.state.is_terminal():
                return fr.state
            await asyncio.sleep(poll_seconds)

async def main():
    # timeout=0 returns immediately so we can start streaming right away
    fr = await arun_deployment("my-flow/my-deployment", timeout=0)
    flow_run_id = str(fr.id)

    # stream logs and wait for completion concurrently
    log_task = asyncio.create_task(stream_flow_run_logs(flow_run_id))
    state = await wait_until_done(flow_run_id)
    # give a moment for any last logs to arrive, then cancel stream
    await asyncio.sleep(1)
    log_task.cancel()
    print(f"Flow run finished in state: {state.type}")

asyncio.run(main())

Programmatic — batch fetch (pull logs at once)

import logging
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import LogFilter
from prefect.client.schemas.sorting import LogSort

# Launch and return immediately
fr = run_deployment("my-flow/my-deployment", timeout=0)
flow_run_id = fr.id

with get_client() as client:
    logs = client.read_logs(
        logs=LogFilter(flow_run_id={"any_": [flow_run_id]}),
        limit=None,
        sort=LogSort.TIMESTAMP_ASC
    )
    for log in logs:
        level = logging.getLevelName(log.level)
        print(f"{log.timestamp} | {level:7} | {log.message}")

Notes - There isn’t a single “tail_flow_run_logs()” helper; use the logs subscriber (streaming) or

read_logs

(batch), or the CLI. - In CI (e.g., AWS CodeBuild), the async streaming example is a good fit to mirror CLI tailing and print to stdout. If websockets are restricted, fall back to the batch fetch or the CLI command. If you want, I can adapt one of these into a copy-pasteable script that: - launches a specific deployment, - streams logs to the build console, - exits with a non-zero code if the flow run fails.