Thanks for the snippet and the traceback — this is a classic “async in a flow” gotcha. What’s happening: - In Prefect 3.x, flows run in an async context. -

Block.load(...)

and most notification block methods are async-compatible. When called inside an async context (like a flow),

SlackWebhook.load(...)

returns a coroutine that you must await. The same applies to

notify(...)

. That’s why you’re seeing

'coroutine' object has no attribute 'notify'

—

SlackWebhook.load(...)

returned a coroutine, not a block instance. Fix: make your helper async and await both

load

and

notify

, then make your flow async and await the helper. Here’s a corrected version using the built-in notification block:

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

async def send_slack_message(msg: str):
    webhook = await SlackWebhook.load("ideon-updates-slack-notification")
    await webhook.notify(msg)

@flow
async def my_flow():
    await send_slack_message("this is a test message")

if __name__ == "__main__":
    my_flow()

A few notes: - This will also work the same when running “locally” — the key is that inside a flow you’re in an async context, so you must

await

async-compatible methods. - If you prefer the dedicated Slack integration (attachments/blocks, richer features), you can use

prefect-slack

similarly:

from prefect import flow
  from prefect_slack import SlackWebhook

  async def send_slack_message(msg: str):
      webhook = await SlackWebhook.load("ideon-updates-slack-notification")
      await webhook.notify(msg)

  @flow
  async def my_flow():
      await send_slack_message("this is a test message")

Docs for reference: - Built-in notification blocks: Notification blocks - Prefect Slack integration: prefect-slack docs If you’d prefer to keep

my_flow

as a sync function, we can wrap the awaits with Prefect’s sync utilities — but the async flow pattern above is the most straightforward.

Great question. You don’t need two separate helpers—there are two clean ways to make a single function work in both sync and async flows. Option A (public, simplest): use sync_compatible on a single async helper - Define the helper as async and decorate it with sync_compatible. - Call it normally in a sync flow; await it in an async flow. Example:

from prefect.blocks.notifications import SlackWebhook
from prefect.utilities.asyncutils import sync_compatible
from prefect import flow

@sync_compatible
async def send_slack_message(msg: str):
    block = await SlackWebhook.load("ideon-updates-slack-notification")
    await block.notify(msg)

@flow
def sync_flow():
    send_slack_message("hello from sync")

@flow
async def async_flow():
    await send_slack_message("hello from async")

Docs: sync_compatible API Option B (mirrors Prefect internals): pair sync and async impls with async_dispatch - Keep a true sync implementation and a true async implementation, and let async_dispatch route based on context. - In async flows you must await; in sync flows you just call it. Example:

from prefect.blocks.notifications import SlackWebhook
from prefect._internal.compatibility.async_dispatch import async_dispatch
from prefect import flow

async def _send_slack_message_async(msg: str):
    block = await SlackWebhook.load("ideon-updates-slack-notification")
    await block.notify(msg)

@async_dispatch(_send_slack_message_async)
def send_slack_message(msg: str):
    block = SlackWebhook.load("ideon-updates-slack-notification")
    block.notify(msg)

@flow
def sync_flow():
    send_slack_message("hello from sync")

@flow
async def async_flow():
    await send_slack_message("hello from async")

Why your original code failed in async flows: - In Prefect 3.x,

Block.load()

is context-aware; inside an async context it returns a coroutine you must await. That’s why you saw "'coroutine' object has no attribute 'notify'". - Source for context-aware load: Block.load implementation Either Option A or B will let you reuse a single function across both sync and async flows. If you want to avoid internal APIs, use Option A.