Great question! In Prefect 3.x, there are two good ways to send flow run status messages to Microsoft Teams: Option A — No code: use an Automation that sends a notification via a Teams Webhook block Option B — In code: add a flow hook (on_completion/on_failure/etc.) that posts to Teams Below are the steps for both. 1) Set up an Incoming Webhook in Microsoft Teams - In the Teams channel where you want messages: - Click the channel’s … menu > Connectors (or Apps) > search for “Incoming Webhook” - Add the Incoming Webhook connector - Give it a name and icon, then copy the generated webhook URL 2) Create a MicrosoftTeamsWebhook block in Prefect - Via the UI: Blocks > Create block > Notifications > Microsoft Teams Webhook - Paste the Teams webhook URL - Save the block with a name, e.g. ms-teams - Or via Python:

from prefect.blocks.notifications import MicrosoftTeamsWebhook

  teams = MicrosoftTeamsWebhook(url="<https://outlook.office.com/webhook/>…")
  teams.save("ms-teams", overwrite=True)

Option A: Send notifications with an Automation (recommended for “all flows” or org-wide rules) - In Prefect UI: Automations > New Automation - Trigger: When a flow run enters a state (e.g., Completed, Failed, Crashed, Cancelled) - Filters: Choose specific flows/work pools/tags if you want to scope it - Action: Send notification - Select your MicrosoftTeamsWebhook block (ms-teams) - Customize the message body/subject if desired - Save This will post to Teams automatically whenever the chosen event(s) occur. Option B: Send notifications from your flow using hooks (good for per-flow custom messaging) - Define hook functions and register them on your flow. Example:

from prefect import flow
  from prefect.blocks.notifications import MicrosoftTeamsWebhook

  async def notify_on_completion(flow, flow_run, state):
    teams = await MicrosoftTeamsWebhook.load("ms-teams")
    teams.notify(
      subject=f"[Completed] {flow.name}",
      body=(
        f"Flow '{flow.name}' completed.\n"
        f"Run ID: {flow_run.id}\n"
        f"State: {state.name}\n"
        f"Start: {flow_run.start_time}\n"
        f"End: {flow_run.end_time}\n"
      ),
    )

  async def notify_on_failure(flow, flow_run, state):
    teams = await MicrosoftTeamsWebhook.load("ms-teams")
    teams.notify(
      subject=f"[Failed] {flow.name}",
      body=(
        f"Flow '{flow.name}' failed.\n"
        f"Run ID: {flow_run.id}\n"
        f"State: {state.name}\n"
        f"Message: {state.message or 'No message'}\n"
      ),
    )

  @flow(
    name="example-flow",
    on_completion=[notify_on_completion],
    on_failure=[notify_on_failure],
  )
  def example_flow():
    # your flow logic here
    pass

  if __name__ == "__main__":
    example_flow()

Notes - The

MicrosoftTeamsWebhook

block’s

notify(body, subject=None)

method is synchronous; using it in an async hook is fine. - If you prefer a single hook that fires on every state change, you can use

on_running

,

on_crashed

,

on_cancellation

, etc. See

@flow

signature for available hooks. - If you see an error about apprise, install it:

pip install apprise

- Since you’re running Prefect Server on EC2 with RDS Postgres, there’s nothing special you need to change for notifications—the block will post directly to Teams over HTTPS. Docs - Blocks overview: Prefect 3: Blocks - Notification blocks API: Notification Blocks API Would you like: - UI automation (no code) that posts on Completed/Failed for all flows, or - Per-flow hooks (code) with custom messages? Also, which states do you want to notify on: only failures, or all state changes?

Got it — let’s get Slack notifications working. Here’s a quick setup + a checklist to debug why it “didn’t work.” Assumptions: Prefect 3.x, using the SlackWebhook notification block. Quick setup (works for both Automations and in-flow hooks) 1) Ensure Apprise is installed (required by notification blocks) - On the machine running the Prefect flow/worker:

pip install apprise

2) Create a Slack Incoming Webhook and copy the URL - In Slack: Apps > search “Incoming WebHooks” > Add to Slack - Pick the channel and copy the webhook URL (looks like https://hooks.slack.com/services/T…/B…/…) 3) Create and test a Prefect SlackWebhook block - One-time create:

from prefect.blocks.notifications import SlackWebhook

  slack = SlackWebhook(url="<https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>")
  slack.save("slack-default", overwrite=True)

- Test send:

from prefect.blocks.notifications import SlackWebhook

  slack = SlackWebhook.load("slack-default")
  try:
      slack.notify(body="Test message from Prefect")
      print("Sent!")
  except Exception as e:
      print("Failed:", e)

If that “Test message” appears in Slack, wiring into Prefect is next: Option A: Send via an Automation (no code) - In Prefect UI: Automations > New Automation - Trigger: “Flow run enters state” (choose Completed, Failed, Crashed, etc.) - Filters: scope to specific flows/tags as needed - Action: “Send notification” -> choose your SlackWebhook block (slack-default) - Enable/save the automation Option B: Send from the flow via hooks (code)

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

async def notify_success(flow, flow_run, state):
    slack = await SlackWebhook.load("slack-default")
    slack.notify(
        subject=f"[Completed] {flow.name}",
        body=f"Flow '{flow.name}' completed. Run ID: {flow_run.id}"
    )

async def notify_failure(flow, flow_run, state):
    slack = await SlackWebhook.load("slack-default")
    slack.notify(
        subject=f"[Failed] {flow.name}",
        body=f"Flow '{flow.name}' failed. Run ID: {flow_run.id} | {state.message or ''}"
    )

@flow(on_completion=[notify_success], on_failure=[notify_failure])
def my_flow():
    # your logic here
    pass

if __name__ == "__main__":
    my_flow()

Troubleshooting checklist - Apprise installed? If not,

SlackWebhook.notify()

will fail. - Webhook URL correct and active? Try a direct curl:

curl -X POST -H 'Content-type: application/json' \
       --data '{"text":"Hello from curl"}' \
       <https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>

If curl fails, fix the Slack webhook first (channel/app permissions, re-create webhook). - Network egress allowed from your EC2 host? Ensure outbound HTTPS to hooks.slack.com is not blocked (VPC, security groups, proxies). - Using Automations? - Is the automation enabled (not paused)? - Does the trigger actually fire (correct states/filters)? - Check Automations page to see recent runs and any errors. - Using hooks? - Confirm your flow actually reached the hook state (e.g., a failure to trigger on_failure). - Check Prefect logs for exceptions from

notify()

. Docs - Notification Blocks: Prefect 3: Notification Blocks API - Blocks concept: Prefect 3: Blocks If it’s still not working, can you share: - How you’re triggering notifications (Automation vs. in-flow hook)? - Any error output from the test

slack.notify(...)

snippet? - Whether the curl test to the webhook succeeds? - The block name you used and a rough screenshot of your Automation setup (if applicable)?

Thanks for the detail — since curl works but your Python script prints “Sent!” without a Slack message arriving, here’s how to isolate and fix it quickly. Why this happens -

SlackWebhook.notify()

doesn’t return a value and may not raise on a soft failure. A “Sent!” print after

notify()

isn’t proof the message reached Slack. - Common culprits: missing/old Apprise, using a different environment than curl, webhook URL format, or Slack rejecting the payload silently. Step-by-step checks 1) Make sure Apprise is installed and up to date in the SAME environment running your Python script

pip show apprise
pip install -U apprise

2) Minimal Prefect Slack block test (same URL you used with curl)

from prefect.blocks.notifications import SlackWebhook

slack = SlackWebhook(url="<https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>")
slack.notify(body="Test message from Prefect via SlackWebhook")
print("Called SlackWebhook.notify()")

- If nothing shows in Slack, try the Apprise-native URL format too: - Example:

slack = SlackWebhook(url="<slack://TOKEN_A/TOKEN_B/TOKEN_C>")

(same tokens as in your hooks URL) - Also try adding a subject (some renderers format differently):

slack.notify(subject="Prefect", body="Test with subject")

3) Test Apprise directly (bypasses Prefect’s block)

import apprise
a = apprise.Apprise()
a.add("<https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>")
ok = a.notify(title="Apprise direct", body="Hello from Apprise")
print("Apprise notify returned:", ok)

- If this doesn’t land in Slack, it’s an Apprise<->Slack issue (version/format). - If this works but the block doesn’t, it’s how the block is used or the environment. 4) Confirm you’re running the test in the same environment where curl succeeded - Same machine/container? - Same proxy settings? If curl uses a proxy, ensure Python has

HTTP_PROXY/HTTPS_PROXY

env vars set as needed. 5) Quick unblock: use a direct POST (no Apprise) in your hook If you need it working immediately, post directly with requests:

import requests

def send_slack(webhook_url: str, text: str):
    r = <http://requests.post|requests.post>(webhook_url, json={"text": text}, timeout=10)
    r.raise_for_status()

# Example hook usage:
from prefect import flow
async def on_failure(flow, flow_run, state):
    send_slack("<https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>",
               f"[Failed] {flow.name} | run {flow_run.id} | {state.message or ''}")

@flow(on_failure=[on_failure])
def my_flow():
    ...

if __name__ == "__main__":
    my_flow()

This mirrors your working curl payload and avoids Apprise entirely. If still not working, please share: - Your exact Python snippet (redact the webhook) - Output of

prefect version

and

pip show apprise

- Whether the Apprise-direct test returns True and whether that message appears - Confirmation that the Python test runs on the same EC2 host as the successful curl Docs - Notification blocks: Prefect 3: Notification Blocks API - Blocks overview: Prefect 3: Blocks

Thanks—that narrows it down. Since: - curl works - apprise direct works - but SlackWebhook.notify() prints “Sent” with no Slack message It’s almost certainly a configuration issue with the Prefect SlackWebhook block (URL/flags/storage), or you’re loading a different/misconfigured block. Try these quick isolates/fixes: 1) Bypass the saved block and construct inline

from prefect.blocks.notifications import SlackWebhook

slack = SlackWebhook(url="<https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>")
slack.notify(body="Test via inline SlackWebhook")
print("Called SlackWebhook.notify()")

- If this lands in Slack, your runtime can send messages and Apprise is fine. The issue is with the saved block’s contents. 2) Inspect the saved block you’re loading

from prefect.blocks.notifications import SlackWebhook

blk = SlackWebhook.load("slack-default")  # replace with your block name
print("allow_private_urls:", getattr(blk, "allow_private_urls", None))
# DO NOT print the full secret in logs; just verify it's set and looks right
url = blk.url.get_secret_value()
print("url looks set:", bool(url), "starts with <https://hooks.slack.com/services/>", url.startswith("<https://hooks.slack.com/services/>"))

Common gotchas: - allow_private_urls False: Apprise rejects private URLs; set it True. - URL saved with angle brackets <...>: remove the brackets; use the plain https URL. - A stray space/newline in the saved URL. If needed, re-create the block cleanly

from prefect.blocks.notifications import SlackWebhook

slack = SlackWebhook(
    url="<https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>",
    # explicitly ensure this is true
    allow_private_urls=True,
)
slack.save("slack-default", overwrite=True)

Then retest:

SlackWebhook.load("slack-default").notify(body="Test from re-created block")

3) Make sure you’re in the same environment/profile - If you have multiple virtualenvs or Prefect profiles,

SlackWebhook.load("name")

may pick up a different block registry. - Run

prefect version

in the exact environment and confirm Prefect 3.x. - Set

PREFECT_LOGGING_LEVEL=DEBUG

and rerun to see if Apprise logs a URL rejection. 4) If you’re using Automations (server-side) - Automations send notifications from the Prefect Server process, not your flow env. - Ensure Apprise is installed wherever the Prefect Server runs. - Confirm outbound HTTPS from the server process to hooks.slack.com. - Check Automations page/logs to see if the notification action errored and why. Immediate workaround (no Apprise/blocks): direct POST in your hook

import requests
from prefect import flow

def send_slack(webhook_url: str, text: str):
    r = <http://requests.post|requests.post>(webhook_url, json={"text": text}, timeout=10)
    r.raise_for_status()

async def on_failure(flow, flow_run, state):
    send_slack("<https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>",
               f"[Failed] {flow.name} | run {flow_run.id} | {state.message or ''}")

@flow(on_failure=[on_failure])
def my_flow():
    ...

if __name__ == "__main__":
    my_flow()

If it’s still not sending via the block, please share: - How the block was created (UI vs code) - Output of the “inspect the saved block” snippet (esp. allow_private_urls and URL prefix check) - Whether the inline SlackWebhook(url=...).notify() test sends successfully - Prefect version and where your Prefect Server runs (if using Automations) Refs - Notification Blocks API: Prefect 3: Notification Blocks API - Blocks concept: Prefect 3: Blocks

Thanks for the snippet — I see the issue. Your saved Prefect block is using the webhook URL with angle brackets. Slack URLs in code must NOT be wrapped in < >. Slack (the chat app) adds <…> as display formatting, but passing those characters to the block can cause Apprise/Prefect to skip or reject the URL silently. What to do 1) Re-create the block with a clean URL (no angle brackets)

from prefect.blocks.notifications import SlackWebhook

raw_url = "<https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>"  # no < >
slack = SlackWebhook(url=raw_url, allow_private_urls=True)
slack.save("slack-default", overwrite=True)

slack = SlackWebhook.load("slack-default")
slack.notify(body="Test message from Prefect (clean URL)")
print("Called SlackWebhook.notify()")

If you only have the bracketed form, sanitize it:

url_with_brackets = "<https://hooks.slack.com/services/...>"
clean_url = url_with_brackets.strip().lstrip("<").rstrip(">")

2) Double-check the saved block actually holds the clean URL

blk = SlackWebhook.load("slack-default")
print("allow_private_urls:", blk.allow_private_urls)
print("url starts with hooks.slack:", blk.url.get_secret_value().startswith("<https://hooks.slack.com/services/>"))

3) If it still doesn’t post, try the Apprise “slack://” scheme Some environments behave better with Apprise’s native scheme:

# tokens are the 3 parts from your webhook URL
slack = SlackWebhook(url="<slack://TXXXX/BXXXX/XXXXXXXX>", allow_private_urls=True)
slack.notify(body="Test via slack:// scheme")

4) Quick alternative that always works: CustomWebhookNotificationBlock (direct POST)

from prefect.blocks.notifications import CustomWebhookNotificationBlock

cw = CustomWebhookNotificationBlock(
    name="slack-direct",
    url="<https://hooks.slack.com/services/XXXXXXXX/XXXXXXXX/XXXXXXXXXXXXXXXX>",
    method="POST",
    json_data={"text": "Hello from Prefect via CustomWebhookNotificationBlock"},
    headers={"Content-Type": "application/json"},
)
cw.save("slack-direct", overwrite=True)
CustomWebhookNotificationBlock.load("slack-direct").notify(body="ignored", subject=None)

Why your earlier tests behaved differently - curl and your “apprise direct” test worked despite < > because Apprise sometimes tolerates or strips them when URLs are added directly. - The Prefect SlackWebhook block stores the URL as a secret and may not strip < >, leading to a no-op post. -

SlackWebhook.notify()

returns None even on success; printing “Sent!” only means the method returned, not that Slack accepted the message. If issues persist, please share: - Output of: -

prefect version

-

pip show apprise

- Whether

SlackWebhook(url=raw_url).notify(...)

works after removing the brackets - Any DEBUG logs (

PREFECT_LOGGING_LEVEL=DEBUG

) around the notification Refs - Notification blocks API: Prefect 3: Notification Blocks API - Blocks overview: Prefect 3: Blocks

Got it — the reason you’re seeing the same message each time is that your CustomWebhookNotificationBlock has a hard-coded JSON payload. The block’s

json_data={"text": "Hello from Prefect via CustomWebhookNotificationBlock"}

is what gets sent by Automations every time, and the

body

passed by the automation is not used to overwrite that. You’ve got two good ways to get per–flow-run, per–state messages: Option 1 (recommended): Use an Automation “Call a webhook” action with a templated JSON body - In Prefect UI > Automations: - Trigger: “Flow run enters state” (pick Completed/Failed/Crached/Running/etc.) - Action: “Call a webhook” - URL: your Slack webhook URL - Method: POST - Headers: Content-Type: application/json - Body: use a Jinja template to include flow/run/state context. Example:

{
        "text": "[{{ state.name }}] {{ flow.name }} — run {{ flow_run.name or flow_run.id }}\n"
                "Deployment: {{ deployment.name if deployment else 'ad-hoc' }}\n"
                "State: {{ state.name }}{% if state.message %} — {{ state.message }}{% endif %}\n"
                "Start: {{ flow_run.start_time }}  End: {{ flow_run.end_time }}\n"
                "Parameters: {{ flow_run.parameters | tojson }}"
      }

- Prefect will render the template on each event so Slack gets a unique message per flow run/state. - Templating docs: Automation templates and parameters Option 2: Use “Send notification” with SlackWebhook block (not CustomWebhook), and template the message in the Automation - Fix your SlackWebhook block (no angle brackets in URL). - In the Automation action “Send notification,” put a templated message body (same content as above, but without the outer JSON since SlackWebhook handles formatting):

[{{ state.name }}] {{ flow.name }} — run {{ flow_run.name or flow_run.id }}
  Deployment: {{ deployment.name if deployment else 'ad-hoc' }}
  State: {{ state.name }}{% if state.message %} — {{ state.message }}{% endif %}
  Start: {{ flow_run.start_time }}  End: {{ flow_run.end_time }}
  Parameters: {{ flow_run.parameters | tojson }}

- This ensures each notification is run-specific. If you really want to keep using CustomWebhookNotificationBlock with “Send notification” - Remove the hard-coded

json_data

from the block and rely on the Automation’s “body” to become the payload. However, the CustomWebhookNotificationBlock is designed to use its own stored request config; in practice, the “Call a webhook” action (Option 1) is the clean way to send templated JSON to Slack. Which route would you like to take? If you share which states you want to notify on (e.g., Failed + Crashed only), I can craft the exact Automation templates for you.

You’re right — that payload wasn’t valid JSON. In JSON you can’t split one string across multiple adjacent quoted segments. Use a single string for the text value. Try this minimal, valid JSON body for “Call a webhook” action:

{
  "text": "[{{ state.name }}] {{ flow.name }} — run {{ flow_run.name or flow_run.id }}\nDeployment: {{ deployment.name if deployment else 'ad-hoc' }}\nState: {{ state.name }}{% if state.message %} — {{ state.message }}{% endif %}\nStart: {{ flow_run.start_time }}  End: {{ flow_run.end_time }}"
}

Tips: - In your Automation action: - Method: POST - Headers: Content-Type: application/json - Body: paste the JSON above exactly as-is (one single string for "text") - Avoid embedding complex JSON like parameters inside the text unless you escape quotes. If you also want to include parameters, escape quotes so the final rendered JSON stays valid. Example:

{
  "text": "[{{ state.name }}] {{ flow.name }} — run {{ flow_run.name or flow_run.id }}\nDeployment: {{ deployment.name if deployment else 'ad-hoc' }}\nState: {{ state.name }}{% if state.message %} — {{ state.message }}{% endif %}\nStart: {{ flow_run.start_time }}  End: {{ flow_run.end_time }}\nParameters: ```{{ flow_run.parameters | tojson | replace('\"','\\\"') }}```"
}

Alternatively, use “Send notification” with a SlackWebhook block (no JSON needed) and just paste a templated message body:

[{{ state.name }}] {{ flow.name }} — run {{ flow_run.name or flow_run.id }}
Deployment: {{ deployment.name if deployment else 'ad-hoc' }}
State: {{ state.name }}{% if state.message %} — {{ state.message }}{% endif %}
Start: {{ flow_run.start_time }}  End: {{ flow_run.end_time }}

Docs: Automation templating guide If this still errors, paste the exact validation message you see and I’ll adjust the payload accordingly.