Notifications

A nightly job that fails at 2 AM should reach someone on Slack, instead of waiting for a person to notice in the UI. Attach notifications to the trigger and every run it starts reports its outcome.

You can attach notifications to a trigger using the notifications parameter of flyte.Trigger. Notifications fire when a triggered run reaches a terminal execution phase.

triggers.py
                
                    
                
            
                
            

                
                    
                
            
import flyte
from flyte import notify
from flyte.models import ActionPhase

env = flyte.TaskEnvironment(name="my_task_env")

trigger_with_notifications = flyte.Trigger(
    name="daily_report",
    automation=flyte.Cron("0 9 * * 1-5"),
    notifications=(
        notify.Slack(
            on_phase=ActionPhase.FAILED,
            webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
            message="Run {{.Run.Name}} failed with: {{.Error}}",
        ),
        notify.Email(
            on_phase=ActionPhase.SUCCEEDED,
            recipients=["[email protected]"],
            subject="Run {{.Run.Name}} succeeded",
            body="Run: {{.Run.Name}}",
        ),
    ),
)

@env.task(triggers=trigger_with_notifications)
def process_data(date: str) -> str:
    return f"Processed {date}"

Execution phases

The on_phase parameter accepts a single phase or a tuple of terminal phases from flyte.models.ActionPhase:

Phase	Description
`ActionPhase.SUCCEEDED`	Run completed successfully
`ActionPhase.FAILED`	Run failed with an error
`ActionPhase.TIMED_OUT`	Run exceeded its timeout
`ActionPhase.ABORTED`	Run was manually aborted

To notify on multiple phases with the same notification:

notifications.py
                
notify.Email(
    on_phase=(ActionPhase.FAILED, ActionPhase.ABORTED),
    recipients=["[email protected]"],
    subject="Alert: Run completed with phase {{.Phase}}",
    body="Run: {{.Run.Name}}\nError: {{.Error}}",
)

Template variables

All message fields support template variables that are substituted at delivery time:

Variable	Description
`{{.Run.Project}}`	Project name
`{{.Run.Domain}}`	Domain name
`{{.Run.Name}}`	Run ID
`{{.Phase}}`	Execution phase
`{{.Error}}`	Error message when failed or abort reason when aborted

Slack notifications

notify.Slack sends a message to a Slack channel via an incoming webhook.

Simple message:

notifications.py
                
notify.Slack(
    on_phase=ActionPhase.FAILED,
    webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
    message="Run {{.Run.Name}} failed in {{.Run.Project}}/{{.Run.Domain}}: {{.Error}}",
)

Rich formatting with Block Kit:

Use blocks instead of message for structured layouts. When blocks is provided, message is ignored.

notifications.py
                
                    
                
            
                
            

                
                    
                
            
notify.Slack(
    on_phase=ActionPhase.SUCCEEDED,
    webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
    blocks=[
        {
            "type": "header",
            "text": {"type": "plain_text", "text": "Task Succeeded"},
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": "*Run:*\n{{.Run.Name}}"},
                {"type": "mrkdwn", "text": "*Phase:*\n{{.Phase}}"},
            ],
        },
        {"type": "divider"},
        {
            "type": "context",
            "elements": [
                {"type": "mrkdwn", "text": "{{.Run.Project}}/{{.Run.Domain}}"},
            ],
        },
    ],
)

Email notifications

notify.Email sends an email notification. You can provide a plain-text body, an html_body, or both (the email is sent as multipart when both are present).

notifications.py
                
notify.Email(
    on_phase=ActionPhase.FAILED,
    recipients=["[email protected]"],
    cc=["[email protected]"],
    subject="ALERT: Run {{.Run.Name}} failed",
    body="Run: {{.Run.Name}}\nError: {{.Error}}",
    html_body="<b>Error:</b> {{.Error}}<br>",
)

Microsoft Teams notifications

notify.Teams sends a message to a Teams channel via an incoming webhook. Use card for Adaptive Card formatting; when card is set, title and message are ignored.

notifications.py
                
notify.Teams(
    on_phase=ActionPhase.FAILED,
    webhook_url="https://outlook.office.com/webhook/YOUR_WEBHOOK_URL",
    title="Task Failed",
    message="Run {{.Run.Name}} failed: {{.Error}}\n",
)

Custom webhook notifications

notify.Webhook sends an HTTP request to any endpoint. All string values in headers and body support template variables.

notifications.py
                
notify.Webhook(
    on_phase=ActionPhase.SUCCEEDED,
    url="https://api.example.com/events",
    method="POST",
    headers={"Authorization": "Bearer my-token"},
    body={
        "event": "task_succeeded",
        "run": "{{.Run.Name}}",
    },
)

To attach notifications to a single run instead of a trigger, see Run with notifications.