Thanks for the detailed context — a few things stand out for Prefect 2.x: What’s likely happening - The

prefect.blocks.notifications.SlackWebhook

block uses Apprise under the hood. By default, Apprise returns False on HTTP failures and does not raise. Prefect’s

notify()

mirrors that behavior unless you explicitly opt-in to raising. - Additionally, Prefect 2.x task/flow hooks catch and log exceptions (so even if you do raise, the task run won’t fail due to the hook). - This combo can make failures “look successful” in logs unless you capture the Apprise debug output. Quick fix to surface the error Wrap the call in

raise_on_failure()

and optionally set a standard notify_type to avoid edge cases with Prefect’s custom type in 2.x:

from prefect.blocks.notifications import SlackWebhook
from prefect.blocks.abstract import NotificationError

def data_loader_failure_notify_slack(
    task: Task, task_run: TaskRun, state: State, env: str, data_source_name: str
):
    print(f"Sending Slack message for data loader '{data_source_name}' failure")

    slack_webhook_block: SlackWebhook = (
        SlackWebhook.load("slack-prod") if env == "prod" else SlackWebhook.load("slack-staging")
    )

    try:
        # Optional: use a standard Apprise notify type in Prefect 2.x
        slack_webhook_block.notify_type = "info"

        with slack_webhook_block.raise_on_failure():
            slack_webhook_block.notify(
                body=(
                    f"Data loader `{data_source_name}` failed at "
                    f"{datetime.now().astimezone(tz=ZoneInfo('America/New_York'))}\n"
                    f"Flow run URL: {flow_run.get_flow_run_ui_url()}"
                ),
                subject="Data loader failure",
            )
        print(f"Sent Slack message for data loader '{data_source_name}' failure")

    except NotificationError as e:
        # This prints Apprise’s debug logs including HTTP status/connection errors
        print("Slack notification failed. Apprise logs:")
        print(e.log)

Why Automations work but this hook doesn’t - Automations run in Prefect Cloud/Server infrastructure and send the webhook from there. - Your hook runs inside your worker’s runtime (container/VM). If that environment lacks internet egress/DNS/CA certs or is behind a proxy, Slack calls can fail there while Automations still succeed. Debug checklist - Use the

raise_on_failure()

wrapper above and share the

NotificationError.log

output — it’ll include the HTTP status or connection error. - Verify egress from the worker runtime: - Can it resolve and reach

<https://hooks.slack.com>

? - Corporate firewall/proxy? If so, set `HTTP_PROXY`/`HTTPS_PROXY` or allowlist Slack. - Ensure CA certificates are installed in the image (e.g.,

ca-certificates

on Debian/Alpine). - Double-check you’re loading the intended block per env (

slack-prod

vs

slack-staging

) in the same workspace your worker is connected to. - Optional: explicitly set

notify_type = "info"

as above. Minimal in-environment test Run this as a simple flow in the same image/pod as your worker:

from prefect import flow
from prefect.blocks.notifications import SlackWebhook

@flow
def test_slack(env: str = "staging"):
    block = SlackWebhook.load("slack-prod" if env == "prod" else "slack-staging")
    block.notify_type = "info"
    with block.raise_on_failure():
        block.notify("Test message from worker runtime", subject="Prefect Slack test")

if __name__ == "__main__":
    test_slack()

Alternative (sometimes clearer errors) If you prefer using the Slack SDK (bot token) or an incoming webhook helper via the collection: - Incoming webhook helper: - prefect-slack: messages - Bot-token chat messages: - prefect-slack integration Docs links

- Notifications blocks API (Prefect 2.x): Notifications blocks API (Prefect 2.x) - prefect-slack integration: prefect-slack integration A couple quick questions to help pinpoint this: - Where do your flows run (Docker/Kubernetes/VM) and is that environment behind a proxy or egress-restricted? - After adding

raise_on_failure()

, what do you see in

NotificationError.log

?

Thanks for digging in and for the extra detail — that helps. What your result suggests - If both except prints ran but the exception had no message and

NotificationError.log

was empty, that typically means Apprise returned False without emitting any logs. This can happen if Apprise didn’t add any service from the URL (e.g., URL not recognized/empty) or short-circuited before logging. - Since you’re on the latest 2.x (with

notify_type='info'

), the older notify_type issue shouldn’t be the cause. Targeted checks to pinpoint the issue Please try these from the same Docker container/image where your worker runs: 1) Verify the block actually has a usable URL at runtime (redact it!)

from prefect.blocks.notifications import SlackWebhook

b = SlackWebhook.load("slack-integration")  # or slack-staging in your test
u = b.url.get_secret_value()
print("Webhook URL present:", bool(u))
print("Ends with:", u[-12:])  # redact most of it
print("Contains <http://hooks.slack.com|hooks.slack.com>:", "<http://hooks.slack.com|hooks.slack.com>" in u)

2) Bypass Apprise and hit Slack directly (isolates Apprise vs. network/webhook)

import json, requests
from prefect.blocks.notifications import SlackWebhook

b = SlackWebhook.load("slack-integration")  # or your staging name
u = b.url.get_secret_value()

r = <http://requests.post|requests.post>(
    u,
    data=json.dumps({"text": "Prefect test from worker runtime"}),
    headers={"Content-Type": "application/json"},
    timeout=10,
)
print("Status:", r.status_code, "Body:", r.text)

- If this returns 200, network + webhook are fine and the issue is specifically with Apprise’s send path. - If this is non-200, Slack is rejecting the payload or the webhook is invalid/disabled. 3) Try Apprise directly (to see its return and logs outside Prefect’s wrapper)

import logging
from apprise import Apprise

logging.getLogger("apprise").setLevel(logging.DEBUG)

ap = Apprise()
ap.add(u)  # from above
ok = ap.notify(body="Test via Apprise", title="Prefect")
print("Apprise notify returned:", ok)

4) Add a bit more logging inside your hook to confirm the URL and Apprise version (redacted)

from apprise import __version__ as apprise_ver
print("Apprise version:", apprise_ver)
print("Webhook tail:", slack_webhook_block.url.get_secret_value()[-12:])

A minimal drop-in workaround (keeps using your existing block) If step (2) works (200 OK), you can switch your hook to a direct POST so you’re not blocked by Apprise behavior:

import json, requests
from prefect.blocks.notifications import SlackWebhook

def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):
    print(f"Sending Slack message for data loader '{data_source_name}' failure")
    b = SlackWebhook.load("slack-integration" if env == "prod" else "slack-staging")
    u = b.url.get_secret_value()
    payload = {
        "text": (
            f"Data loader `{data_source_name}` failed at "
            f"{datetime.now(ZoneInfo('America/New_York'))}\n"
            f"Flow run URL: {flow_run.get_flow_run_ui_url()}"
        )
    }
    try:
        r = <http://requests.post|requests.post>(u, data=json.dumps(payload), headers={"Content-Type": "application/json"}, timeout=10)
        r.raise_for_status()
        print(f"Sent Slack message for data loader '{data_source_name}' failure")
    except Exception as e:
        print(f"Error sending Slack notification: {e}")

Optional alternative: prefect-slack collection If you’d rather not hand-roll the POST and want clearer error surfaces: - Incoming webhook helper: prefect-slack: messages - You’d create a prefect-slack

SlackWebhook

block (different type than core) and call

send_incoming_webhook_message(...)

in the hook. A couple of clarifications/questions to help me zero in: - What exact versions are you on? Please share

prefect.__version__

and

apprise.__version__

. - What are the results of steps (1)–(3) above?

- Is the webhook URL’s tail the same one Automations are using? (Just compare last 10–12 chars.) Reference - Notifications blocks API (Prefect 2.x): Notifications blocks API (Prefect 2.x) - prefect-slack integration: prefect-slack integration Once we see the direct POST or Apprise test result, we’ll know if it’s the webhook or Apprise behavior and can either stick with the direct POST or address an Apprise-specific quirk.

Super helpful — thanks. Given: - Apprise 1.9.5 - The saved block URL is correct - Direct requests.post works - Direct Apprise notify() works - Your hook’s try/except prints ran, but the exception had no message The most likely culprit is cancellation: in Prefect 2.x, when a task fails, the engine enters a cancellation scope. If your hook awaits async work, it can be cancelled mid-flight. In that case you’ll typically see an exception with an empty str() (often a CancelledError), which matches what you observed. Because the call is cancelled, Apprise never completes the send and LogEavesdropper has no logs to show. Two solid ways to fix this: Option A: Make the hook async and shield the send from cancellation

import anyio
from datetime import datetime
from zoneinfo import ZoneInfo
from prefect.blocks.notifications import SlackWebhook
from prefect.blocks.abstract import NotificationError
from prefect.runtime import flow_run

async def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):
    print(f"Sending Slack message for data loader '{data_source_name}' failure")

    slack_webhook_block: SlackWebhook = SlackWebhook.load(
        "slack-integration" if env == "prod" else "slack-staging"
    )

    try:
        # Shield the async send from engine cancellation
        with anyio.CancelScope(shield=True):
            with slack_webhook_block.raise_on_failure():
                await slack_webhook_block.notify(
                    body=(
                        f"Data loader `{data_source_name}` failed at "
                        f"{datetime.now(ZoneInfo('America/New_York'))}\n"
                        f"Flow run URL: {flow_run.get_flow_run_ui_url()}"
                    ),
                    subject="Data loader failure",
                )
        print(f"Sent Slack message for data loader '{data_source_name}' failure")
    except Exception as e:
        import traceback
        print(f"Error sending Slack notification: {e!r} ({type(e)})")
        traceback.print_exc()
        if isinstance(e, NotificationError):
            print(f"Apprise NotificationError: {e.log}")

- Prefect 2.x supports async hooks; the engine will await them. - The shield prevents cancellation from interrupting the notify call. Option B: Use a synchronous send inside the hook (not cancellable by anyio) Since both your direct POST and Apprise’s sync

notify()

work, you can keep the block but post synchronously to bypass any async cancellation: A1) Using requests.post directly (you already validated this works):

import json, requests
from datetime import datetime
from zoneinfo import ZoneInfo
from prefect.blocks.notifications import SlackWebhook
from prefect.runtime import flow_run

def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):
    print(f"Sending Slack message for data loader '{data_source_name}' failure")
    b = SlackWebhook.load("slack-integration" if env == "prod" else "slack-staging")
    u = b.url.get_secret_value()

    payload = {
        "text": (
            f"Data loader `{data_source_name}` failed at "
            f"{datetime.now(ZoneInfo('America/New_York'))}\n"
            f"Flow run URL: {flow_run.get_flow_run_ui_url()}"
        )
    }
    try:
        r = <http://requests.post|requests.post>(u, data=json.dumps(payload), headers={"Content-Type": "application/json"}, timeout=10)
        r.raise_for_status()
        print(f"Sent Slack message for data loader '{data_source_name}' failure")
    except Exception as e:
        print(f"Error sending Slack notification: {e!r}")

A2) Or use Apprise’s sync client directly: ``` from apprise import Apprise from datetime import datetime from zoneinfo import ZoneInfo from prefect.blocks.notifications import SlackWebhook from prefect.runtime import flow_run def data_loader_failure_notify_slack(task, task_run, state, env: str, data_source_name: str):

print(f"Sending Slack message for data loader '{data_source_name}' failure") b = SlackWebhook.load("slack-integration" if env == "prod" else "slack-staging") u = b.url.get_secret_value() ap = Apprise() ap.add(u) ok = ap.notify( body=( f"Data loader

{data_source_name}

failed at " f"{datetime.now(ZoneInfo('America/New_York'))}\n" f"Flow run URL: {flow_run.get_flow_run_ui_url()}" ), title="Data loader failure", ) print("Apprise sync notify returned:", ok)

Quick confirmation step
To verify cancellation is what you’re seeing, add this in your current except:

import traceback print("Exception type:", type(e)) traceback.print_exc() ``` If it shows a CancelledError traceback, that confirms it. If you’re open to it, I recommend Option A (async hook + shield) to keep using the Prefect block API and still be resilient to engine cancellation. Option B is the pragmatic “just send it” approach. Want me to help turn this into a minimal reproducible example and open a GitHub issue for 2.x hooks cancelling notification sends?

Awesome — here’s a minimal, self-contained Prefect 2.x example you can drop into your environment. It shows two approaches: - Async hook that uses the SlackWebhook block and shields the notify call from cancellation - Synchronous fallback using a direct POST (bypasses Apprise/async entirely) Instructions - Set SLACK_WEBHOOK_URL to a valid Slack Incoming Webhook URL - Run the script in the same Docker image/runtime your worker uses - You can call either flow at the bottom ``` import os import json import requests import anyio from datetime import datetime from zoneinfo import ZoneInfo from functools import partial from pydantic import SecretStr from prefect import flow, task from prefect.runtime import flow_run from prefect.blocks.notifications import SlackWebhook from prefect.blocks.abstract import NotificationError # One-time helper: create/update a SlackWebhook block from env var def setup_slack_block(block_name: str = "slack-test"): url = os.environ.get("SLACK_WEBHOOK_URL") if not url: raise RuntimeError("Please set SLACK_WEBHOOK_URL to a valid Slack incoming webhook URL.") SlackWebhook(url=SecretStr(url)).save(block_name, overwrite=True) return block_name # ------------- Demo task that fails ------------- @task def will_fail(): raise RuntimeError("Boom! This task failed intentionally.") # ------------- Example A: Async hook with shield (uses SlackWebhook block) ------------- async def slack_failure_hook_async(task, task_run, state, env: str, data_source_name: str, block_name: str): print(f"[async hook] Sending Slack message for data loader '{data_source_name}' failure") slack_block: SlackWebhook = SlackWebhook.load(block_name) # Make sure notify_type is something standard; in latest 2.x this should already default to 'info' slack_block.notify_type = "info" try: # Shield the send from cancellation during task failure handling with anyio.CancelScope(shield=True): with slack_block.raise_on_failure(): await slack_block.notify( body=( f"Data loader

{data_source_name}

failed at " f"{datetime.now(ZoneInfo('America/New_York'))}\n" f"Flow run URL: {flow_run.get_flow_run_ui_url()}" ), subject="Data loader failure", ) print(f"[async hook] Sent Slack message for data loader '{data_source_name}' failure") except Exception as e: import traceback print(f"[async hook] Error sending Slack notification: {e!r} ({type(e)})") traceback.print_exc() if isinstance(e, NotificationError): print(f"[async hook] Apprise NotificationError logs:\n{e.log}") @flow(name="demo-async-block-with-shield") def demo_async_block_with_shield(env: str = "staging", block_name: str = "slack-test"): # Ensure the block exists (or create it from SLACK_WEBHOOK_URL) setup_slack_block(block_name) # Attach the async, shielded hook h = partial( slack_failure_hook_async, env=env, data_source_name="example-dataloader", block_name=block_name, ) failing_task = will_fail.with_options(on_failure=[h]) # Trigger the failure and swallow so the flow completes try: failing_task() except Exception as e: print(f"[flow] Task failed as expected: {e!r}") # ------------- Example B: Sync fallback using direct POST (bypasses Apprise) ------------- def slack_failure_hook_sync_direct_post(task, task_run, state, env: str, data_source_name: str, block_name: str): print(f"[sync hook] Sending Slack message for data loader '{data_source_name}' failure (direct POST)") slack_block: SlackWebhook = SlackWebhook.load(block_name) url = slack_block.url.get_secret_value() payload = { "text": ( f"Data loader

{data_source_name}

failed at "

f"{datetime.now(ZoneInfo('America/New_York'))}\n" f"Flow run URL: {flow_run.get_flow_run_ui_url()}" ) } try: r = requests.post(url, data=json.dumps(payload), headers={"Content-Type": "application/json"}, timeout=10) r.raise_for_status() print(f"[sync hook] Sent Slack message for data loader '{data_source_name}' failure") except Exception as e: print(f"[sync hook] Error sending Slack notification: {e!r}") @flow(name="demo-sync-direct-post") def demo_sync_direct_post(env: str = "staging", block_name: str = "slack-test"): # Ensure the block exists (or create it from SLACK_WEBHOOK_URL) setup_slack_block(block_name) h = partial( slack_failure_hook_sync_direct_post, env=env, data_source_name="example-dataloader", block_name=block_name, ) failing_task = will_fail.with_options(on_failure=[h]) try: failing_task() except Exception as e: print(f"[flow] Task failed as expected: {e!r}") if name == "__main__": # Pick one to run: demo_async_block_with_shield() # demo_sync_direct_post() ``` Notes - Example A keeps using the Prefect SlackWebhook block but shields the async send to prevent engine cancellation from interrupting the notification. - Example B is the pragmatic fallback (sync direct POST); since it’s synchronous, it won’t be cancelled by anyio. - If you want dynamic

data_source_name

per item like your original, keep using

functools.partial

like shown. If you hit any snags running this in your container, paste the stdout from the hook section — especially the exception type/traceback — and I’ll help tweak it.