Great question. In Prefect 3.x you can trigger an Automation when a log line is written, then send a notification. Here’s how to word/configure it in the UI: - Trigger: When an event occurs - Event name: starts with

prefect.log.write

(type “log” and pick the log write event from the autocomplete) - Text contains:

Using device: cpu

(you can also include

sleap_nn.predict:run_inference

if you want it extra specific) - Scope (to avoid noise): - Resource type:

prefect.flow-run

- Resource name: your flow name (or scope to a specific deployment/flow-run tag if you prefer) - Action: Send a notification - Choose your notification block (Slack/Email/PagerDuty) and add a helpful message, e.g.: - Title: “SLEAP is using CPU” - Body: “Flow run {{ event.resources[0].name }} logged ‘Using device: cpu’.” Notes and tips: - Matching on “Text contains” is usually the most reliable for logs; the full line includes pipes and line qualifiers, so keying on

Using device: cpu

(and optionally

sleap_nn.predict:run_inference

) keeps it robust. - If you don’t see a log write event when you type “log”, it may not be enabled in your workspace. In that case, a reliable alternative is to emit a custom event from your code at the same point you log that message, then trigger on that custom event name. - Docs on filters you’ll see in the UI: Event filters reference If you share: - Your flow name (or deployment name), - Your notification target (Slack vs Email), I can suggest the exact filters and a concise notification template text you can paste into the UI.

Got it. Since you’re self-hosting, here’s how to ensure events/log-based automations work, plus a Python-first option that doesn’t require the UI. 1) Enable events in your self-hosted server - Check which services are enabled

prefect server services list-services

You should see entries like EventLogger, EventPersister, ProactiveTriggers, ReactiveTriggers, Actions. EventPersister/Triggers/Actions are typically enabled by default; EventLogger may be off. - Enable the EventLogger service and start the server - Set this environment variable where you start Prefect Server:

export PREFECT_SERVER_SERVICES_EVENT_LOGGER_ENABLED=true

- Start the server:

prefect server start

- Re-run the services list to confirm:

prefect server services list-services

- Ensure your flows/workers point at your server and send logs - Set

PREFECT_API_URL

in the environment of your workers/flows so logs and events reach your server:

export PREFECT_API_URL=http://<your-server-host>:<port>/api

- If you’ve customized logging, ensure logs are still sent to the API. Docs: Logging customization - Verify events are flowing - In a separate terminal:

prefect events stream --format text

- Run a flow. You should see events streaming in the terminal. 2) UI-based automation on logs (if you want to try it first) - In the Automations page, create a new automation: - Trigger: When an event occurs - Event name: select the log write event (type “log” and pick it; it’s typically shown as a “log write” event) - Text contains: Using device: cpu - Scope to your flow: Resource type = prefect.flow-run, Resource name = your flow name (or scope via deployment/tags) - Action: Send a notification with your Slack/Email block - Event and filter reference: Event filters reference 3) Python-first approach (no UI): emit a custom event when that log appears If you’d prefer to keep everything in code (and not rely on log-write events), attach a logging handler to the

sleap_nn.predict

logger that emits a Prefect event the moment it sees the message. Then create a simple automation on that custom event name. - Code to emit a custom event on the target log line:

import logging
  from prefect import flow, get_run_logger
  from prefect.events import emit_event
  from prefect.events.related import related_resources_from_run_context

  class SleapCpuDetectHandler(logging.Handler):
      def emit(self, record):
          msg = record.getMessage()
          if "Using device: cpu" in msg:
              # Attach current run context as related resources for traceability
              related = related_resources_from_run_context()
              emit_event(
                  event="custom.sleap.device.cpu",           # your custom event name
                  resource={"prefect.resource.id": "sleap-monitor", "prefect.resource.name": "SLEAP Monitor"},
                  related=related,
                  payload={"logger": record.name, "message": msg},
              )

  @flow
  def my_flow():
      # Attach our handler to the sleap logger at runtime
      logger = get_run_logger()
      sleap_logger = logging.getLogger("sleap_nn.predict")
      sleap_logger.setLevel(<http://logging.INFO|logging.INFO>)
      sleap_logger.addHandler(SleapCpuDetectHandler())

      # Your code that triggers sleap inference and logs the line
      <http://logger.info|logger.info>("Starting inference...")
      # ... run your sleap code that logs: "| INFO | sleap_nn.predict:run_inference:335 | Using device: cpu" ...
      # When that line appears, the handler will emit the Prefect event above

  if __name__ == "__main__":
      my_flow()

- Then, create an Automation that triggers on your custom event: - Trigger: When an event occurs

- Event name equals:

custom.sleap.device.cpu

- Action: Send a notification (Slack/Email/PagerDuty block) - Include variables like: - Title: “SLEAP is using CPU” - Body: “Flow {{ event.related[0].name }} logged ‘Using device: cpu’.” 4) Prefer to notify directly from Python (skip Automations)? If you don’t need an Automation at all, you can send the Slack message directly when the log is detected:

from prefect.blocks.notifications import SlackWebhook

  # once: save a SlackWebhook block in the UI or programmatically
  # SlackWebhook(url="<https://hooks.slack.com/services/...>").save("my-slack")

  async def notify_slack(text: str):
      block = await SlackWebhook.load("my-slack")
      await block.notify(text)

You can call

notify_slack("SLEAP is using CPU")

inside the logging handler instead of emitting a custom event. Would you like me to: - Give you exact filter settings for your flow/deployment name? - Or provide a minimal curl/HTTP example to create the Automation programmatically against your self-hosted API?