Short answer: your loop is “hanging” because

async for log in subscriber

only enters the body when a new log arrives. After the flow finishes, no new logs are emitted, so your state check never runs and the websocket stays open. A robust pattern is to: - Use

get_logs_subscriber(...)

(auto-detects OSS vs Cloud) - Poll the flow’s state on a timeout when no new logs arrive - Optionally fetch historical logs first (the subscriber only streams new logs) Here’s a complete example you can drop in. It streams historical logs first, then live logs, and exits promptly once the flow run is final even if no new logs arrive.

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

async def stream_flow_run_logs(flow_run_id, state_check_interval=1.0):
    # Make sure PREFECT_API_URL is set for OSS, e.g. <http://localhost:4200/api>
    log_filter = LogFilter(
        flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]),
        level=LogFilterLevel(ge_=<http://logging.INFO|logging.INFO>),
    )

    async with get_client() as client:
        # 1) Optionally send historical logs that already exist
        history = await client.read_logs(filter=log_filter)
        for log in history:
            yield f"event: message\ndata: {log.message}\n\n"

        # 2) Subscribe to live logs and poll state on timeout
        async with get_logs_subscriber(filter=log_filter) as sub:
            while True:
                try:
                    log = await asyncio.wait_for(sub.__anext__(), timeout=state_check_interval)
                    yield f"event: message\ndata: {log.message}\n\n"
                except asyncio.TimeoutError:
                    flow_run = await client.read_flow_run(flow_run_id)
                    if flow_run.state and flow_run.state.is_final():
                        # No new logs and run is final — exit cleanly
                        break

Common gotchas: - Start the subscriber before or during the run if you want to see “live” logs; it does not replay past logs unless you fetch them via

read_logs

first. - Ensure your filter isn’t too restrictive. For a single flow run, use:

LogFilter(flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]))

and add a level filter only if you want to suppress DEBUG/INFO. - Make sure your OSS API URL is correct and set in the environment where this code runs:

PREFECT_API_URL=http://<your-server>:4200/api

- Verify logs actually exist for the run by calling `client.read_logs(filter=log_filter)`; if this returns nothing, either your filter is wrong or the run didn’t emit logs. References: - Logging clients (PrefectLogsSubscriber): API: logging clients - Filters (LogFilter, LogFilterLevel, LogFilterFlowRunId): API: filters - Server logs API (read logs): REST: read logs If you can share: - Your

log_filter

construction - When you start subscribing relative to the run - Prefect version and your

PREFECT_API_URL

…I can help tailor this further to your setup.

Thanks for the details — a few things stand out. Key points first - You do not need any special env vars on your Docker/K8s workers for streaming. Streaming happens from your subscriber process to the Prefect API. As long as your runs are sending logs to the API (which they are since

read_logs

works), the worker is fine. - If you’re on OSS behind a proxy/ingress, the most common reason for “no logs” with the subscriber is that WebSocket upgrade is blocked by the proxy/ingress. REST

read_logs

will still work in that case, but streaming won’t. - In your snippet,

subscriber._anext_()

looks incorrect — you should call

sub.__anext__()

or just use

async for

. Also, make sure you’re using the documented filter classes (

LogFilter

,

LogFilterFlowRunId

, etc.). A corrected minimal version This streams historical logs first, then listens for new logs and exits promptly when the flow is final. It uses the public

__anext__()

and the documented filter classes.

import asyncio
import logging
from typing import AsyncGenerator

from prefect.client.orchestration import get_client, PrefectClient
from prefect.client.schemas.filters import (
    LogFilter,
    LogFilterFlowRunId,
    LogFilterLevel,
)
from prefect.logging.clients import get_logs_subscriber

async def event_generator(flow_run_id: str) -> AsyncGenerator[str, None]:
    try:
        log_filter = LogFilter(
            flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]),
            # optional: include to suppress DEBUG
            level=LogFilterLevel(ge_=<http://logging.INFO|logging.INFO>),
        )

        async with get_client() as client:
            client: PrefectClient  # type: ignore

            # 1) Send historical logs
            history_logs = await client.read_logs(filter=log_filter)
            for log in history_logs:
                yield f"event: message\ndata: {log.message.replace('USER_OUTPUT: ', '')}\n\n"

            # 2) Stream live logs with timeout + state check
            async with get_logs_subscriber(filter=log_filter) as sub:
                while True:
                    try:
                        log = await asyncio.wait_for(sub.__anext__(), timeout=1.0)
                        yield f"event: message\ndata: {log.message.replace('USER_OUTPUT: ', '')}\n\n"
                    except asyncio.TimeoutError:
                        flow_run = await client.read_flow_run(flow_run_id)
                        if flow_run.state and flow_run.state.is_final():
                            break

        yield "event: message\ndata: Execution completed successfully.\n\n"
    except Exception as e:
        yield f"event: error\ndata: {str(e)}\n\n"
    finally:
        yield "event: close\ndata: Connection closing.\n\n"

Troubleshooting checklist 1) Verify you’re using the correct filter classes - Prefer the documented classes from `prefect.client.schemas.filters`:

LogFilter

,

LogFilterFlowRunId

,

LogFilterLevel

, etc. - If

read_logs

is working for you already, your filter is probably fine — but using the documented classes helps avoid mismatch in the WebSocket payload. 2) Make sure the WebSocket is reachable - For OSS, the subscriber connects to a WebSocket derived from your

PREFECT_API_URL

. Ensure the process running this code has

PREFECT_API_URL

set, e.g.

http://<host>:4200/api

. - Behind a proxy/ingress, you must allow WebSocket upgrades. Typical NGINX requirements: -

proxy_set_header Upgrade $http_upgrade;

-

proxy_set_header Connection "upgrade";

- Adequate

proxy_read_timeout

- Quick way to see what URL the subscriber will use:

from prefect.logging.clients import logs_out_socket_from_api_url
  from prefect.settings import PREFECT_API_URL
  print(logs_out_socket_from_api_url(PREFECT_API_URL.value()))

Then test with a WebSocket client (e.g., wscat) from the same network as the subscriber process. 3) Add debug logging in the subscriber process

- Set

PREFECT_LOGGING_LEVEL=DEBUG

in the environment running this generator. Connection/retry messages can reveal if the WebSocket can’t connect or is being closed by an intermediary. 4) Subscribe early enough if you need “live” logs - The subscriber only streams new logs. You’re already reading history first, which is good. If the flow is already finished and no new logs are emitted after you subscribe, you’ll only see timeouts — your timeout + state check will then exit. 5) Sanity checks - Try loosening the filter temporarily (e.g., remove

level

filter) to ensure it’s not too restrictive. - If you have SSL/HTTPS with a self-signed cert, ensure the client trusts it; otherwise wss:// connection may fail. References - Logging clients (PrefectLogsSubscriber): API: logging clients - Filters (LogFilter, LogFilterLevel, LogFilterFlowRunId): API: filters If you can share: - Your

PREFECT_API_URL

- Whether you’re behind an ingress/proxy and which one - Any DEBUG logs from the subscriber process …I can help pinpoint whether this is a connectivity (WebSocket) issue vs a code/path issue.

Your message was too long, here's your message at the allowed limit: Here is what I am getting on print(logs_out_socket_from_api_url(PREFECT_API_URL value()) flush=True) <ws //prefect 4200/api/logs/out> Also I have tried PREFECT_LOGGING_LEVEL=DEBUG Here is the logs 2025-11-05 16 58 59 11 28 59 920 | DEBUG | Flow run 'golden-tody' - Executing flow 'report-executor-flow' for flow run 'golden-tody' 2025-11-05 16 58 59 11 28 59 945 | DEBUG | prefect client - Connecting to API at <http //prefect 4200/api/> 2025-11-05 16 58 59 11 28 59 959 | DEBUG | Task run 'run_user_code-4f7' - Created task run 'run_user_code-4f7' for task 'run_user_code' 2025-11-05 16 58 59 11 28 59 963 | DEBUG | Task run 'run_user_code-4f7' - Executing task 'run_user_code' for task run 'run_user_code-4f7' 2025-11-05 16 58 59 11 28 59 965 | DEBUG | prefect events clients - Reconnecting websocket connection 2025-11-05 16 58 59 11 28 59 966 | DEBUG | prefect events clients - Opening websocket connection 2025-11-05 16 58 59 11 28 59 977 | DEBUG | prefect events clients - Pinging to ensure websocket connected 2025-11-05 16 59 03 11 29 03 871 | DEBUG | prefect utilities services critical_service_loop - Starting run of 'get_and_submit_flow_runs' 2025-11-05 16 59 03 11 29 03 872 | DEBUG | prefect workers docker dockerworker a6651d37-6303-4adb-b25b-dd18b5ad0ab5 - Querying for flow runs scheduled before 2025-11-05 11 29 13 871938+00 00 2025-11-05 16 59 03 11 29 03 907 | DEBUG | prefect workers docker dockerworker a6651d37-6303-4adb-b25b-dd18b5ad0ab5 - Discovered 0 scheduled_flow_runs 2025-11-05 16 59 10 11 29 10 175 | DEBUG | prefect utilities services critical_service_loop - Starting run of 'sync_with_backend' 2025-11-05 16 59 10 11 29 10 198 | DEBUG | prefect workers docker dockerworker a6651d37-6303-4adb-b25b-dd18b5ad0ab5 - Worker synchronized with the Prefect API server 2025-11-05 16 59 10 11 28 59 979 | DEBUG | prefect events clients - Pong received Websocket connected 2025-11-05 16 59 10 11 28 59 979 | DEBUG | prefect events clients - Resending 0 unconfirmed events 2025-11-05 16 59 10 11 28 59 980 | DEBUG | prefect events clients - Finished resending unconfirmed events 2025-11-05 16 59 10 11 28 59 981 | DEBUG | prefect client - Connecting to API at <http //prefect 4200/api/> 2025-11-05 16 59 10 11 29 00 064 | DEBUG | prefect events clients - EventsClient(id=139645254338688) Emitting event id=adbcb257-a16a-4351-8d20-8f757a610319 2025-11-05 16 59 10 11 29 00 079 | DEBUG | prefect events clients - Added event id=adbcb257-a16a-4351-8d20-8f757a610319 to unconfirmed events list There are now 1 unconfirmed events 2025-11-05 16 59 10 11 29 00 080 | DEBUG | prefect events clients - EventsClient(id=139645254338688) Emit reconnection attempt 0 2025-11-05 16 59 10 11 29 00 081 | DEBUG | prefect events clients - EventsClient(id=139645254338688) Sending event id=adbcb257-a16a-4351-8d20-8f757a610319 2025-11-05 16 59 10 11 29 00 113 | DEBUG | prefect events clients - EventsClient(id=139645254338688) Checkpointing event id=adbcb257-a16a-4351-8d20-8f757a610319 2025-11-05 16 59 10 11 29 00 116 | ERROR | GlobalEventLoopThread | prefect _internal concurrency - Service 'EventsWorker' failed to process item Event(occurred=DateTime(2025 11 5 11 28 59 946361 tzinfo=Timezone('UTC')) event='prefect task-run Pending' resource=Resource(root={'prefect resource id' 'prefect task-run

Thanks — this helps. A few concrete things jump out: 1) Your WebSocket URL looks malformed in the printout - You showed:

<ws //prefect 4200/api/logs/out>

and earlier

<http //prefect 4200/api/>

- That suggests the scheme separator

://

may be missing in your

PREFECT_API_URL

. - Please double-check

PREFECT_API_URL

is exactly like: -

<http://prefect:4200/api>

(for OSS without TLS) -

<https://your-domain/api>

(if behind TLS) - To confirm what the subscriber will actually use, print the repr to avoid formatting:

from prefect.logging.clients import logs_out_socket_from_api_url
  from prefect.settings import PREFECT_API_URL
  print(repr(PREFECT_API_URL.value()))
  print(repr(logs_out_socket_from_api_url(PREFECT_API_URL.value())))

Expected:

'<http://prefect:4200/api>'

and

'<ws://prefect:4200/api/logs/out>'

2) Use the documented filter classes and public iteration API - In Prefect 3.x, use

LogFilter

and

LogFilterFlowRunId

(no “ClientSchema” suffix), and iterate with

__anext__()

or

async for

. - Also, the

read_logs

call takes

filter=...

, not

log_filter=...

. - A corrected version of your generator:

import asyncio
  import logging
  from typing import AsyncGenerator
  from prefect.client.orchestration import get_client, PrefectClient
  from prefect.client.schemas.filters import LogFilter, LogFilterFlowRunId, LogFilterLevel
  from prefect.logging.clients import get_logs_subscriber

  async def event_generator(flow_run_id: str) -> AsyncGenerator[str, None]:
      try:
          log_filter = LogFilter(
              flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]),
              level=LogFilterLevel(ge_=<http://logging.INFO|logging.INFO>),  # optional
          )

          async with get_client() as client:
              client: PrefectClient  # type: ignore

              # 1) Historical logs
              history_logs = await client.read_logs(filter=log_filter)
              for log in history_logs:
                  yield f"event: message\ndata: {log.message.replace('USER_OUTPUT: ', '')}\n\n"

              # 2) Live stream + state check timeout
              async with get_logs_subscriber(filter=log_filter) as subscriber:
                  while True:
                      try:
                          log = await asyncio.wait_for(subscriber.__anext__(), timeout=1.0)
                          yield f"event: message\ndata: {log.message.replace('USER_OUTPUT: ', '')}\n\n"
                      except asyncio.TimeoutError:
                          flow_run = await client.read_flow_run(flow_run_id)
                          if flow_run.state and flow_run.state.is_final():
                              break

          yield "event: message\ndata: Execution completed successfully.\n\n"
      except Exception as e:
          yield f"event: error\ndata: {str(e)}\n\n"
      finally:
          yield "event: close\ndata: Connection closing.\n\n"

3) No special env is required on Docker/K8s workers for streaming - Workers just send logs to the API (which is working since

read_logs

returns data). - Streaming is between your subscriber process and the Prefect API. 4) If you are behind an ingress/proxy, ensure WebSocket upgrades are allowed - REST log reads can work while WebSockets are blocked. - For NGINX, confirm headers and timeouts:

proxy_set_header Upgrade $http_upgrade;
  proxy_set_header Connection "upgrade";
  proxy_read_timeout 600s;
  proxy_send_timeout 600s;

- Then test the handshake from the subscriber’s network:

from prefect.logging.clients import logs_out_socket_from_api_url
  from prefect.settings import PREFECT_API_URL
  print("logs ws:", repr(logs_out_socket_from_api_url(PREFECT_API_URL.value())))
  # Try connecting with a ws client (e.g., wscat) to the printed URL to confirm upgrades work

5) Turn up debug specifically for logging clients - You already set

PREFECT_LOGGING_LEVEL=DEBUG

which is great. To be extra sure you see subscriber logs, you can add: ```

import logging logging.getLogger("prefect.logging.clients").setLevel(logging.DEBUG)

Next steps to pinpoint the issue
- Please share the exact repr output of:

repr(PREFECT_API_URL.value()) repr(logs_out_socket_from_api_url(PREFECT_API_URL.value())) ``` - Are you running the subscriber behind an ingress/proxy? If so, which one (NGINX/ALB/etc.)? - Confirm your Prefect version. References - Logging clients (streams): API: logging clients - Filters (LogFilter, LogFilterFlowRunId, LogFilterLevel): API: filters