Notifications

To set up automated alerts for task phase transitions in vik-advani-flyte-sdk-9b3ce04, you can attach notification objects to your task executions or scheduled triggers.

from flyte.notify import Slack, Email
from flyte.models import ActionPhase
import flyte

# Define notifications for specific phases
notifications = [
    Slack(
        on_phase=ActionPhase.FAILED,
        webhook_url="https://hooks.slack.com/services/T000/B000/XXXX",
        message="🚨 Task {task.name} failed in {project}/{domain}!\nError: {run.error}"
    ),
    Email(
        on_phase=(ActionPhase.SUCCEEDED, ActionPhase.ABORTED),
        recipients=["dev-alerts@example.com"],
        subject="Task {task.name} reached phase {run.phase}"
    )
]

# Apply notifications to a manual run
flyte.with_runcontext(notifications=notifications).run(my_task, arg1="value")

Notification Phases and Templates

Every notification requires an on_phase argument, which can be a single ActionPhase or a tuple of phases. Supported phases from flyte.models.ActionPhase include:

• FAILED
• SUCCEEDED
• ABORTED
• TIMED_OUT

Template Variables

You can use curly-brace placeholders in messages, subjects, and URLs. These are automatically populated at runtime:

• {task.name}: The name of the task being executed.
• {run.name}: The unique execution name.
• {run.phase}: The current phase (e.g., "FAILED").
• {run.error}: The error message (only available on failure).
• {run.url}: A direct link to the execution in the Flyte console.
• {run.duration}: How long the task ran.
• {project} / {domain}: The Flyte project and domain.

Slack with Block Kit

For richer formatting, use the blocks argument with Slack's Block Kit. If blocks is provided, the message field is ignored.

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

Slack(
    on_phase=ActionPhase.FAILED,
    webhook_url="https://hooks.slack.com/services/...",
    blocks=[
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": "*Task Failure Alert*\nTask: {task.name}\nError: {run.error}"
            }
        },
        {
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "View Execution"},
                    "url": "{run.url}"
                }
            ]
        }
    ]
)

Microsoft Teams and Adaptive Cards

The Teams notifier supports simple messages or complex card definitions using Adaptive Cards.

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

Teams(
    on_phase=ActionPhase.SUCCEEDED,
    webhook_url="https://outlook.office.com/webhook/...",
    title="✅ {task.name} Success",
    message="Execution {run.name} completed successfully in {run.duration}."
)

Custom Webhooks

The Webhook class allows you to send arbitrary HTTP requests to external services. It supports recursive template rendering in the request body.

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

Webhook(
    on_phase=ActionPhase.FAILED,
    url="https://api.internal.tool/v1/alerts",
    method="POST",
    headers={"X-API-Key": "secret-token"},
    body={
        "alert_type": "flyte_failure",
        "metadata": {
            "task": "{task.name}",
            "execution_id": "{run.name}",
            "logs": "{run.url}"
        }
    }
)

Scheduled Task Notifications

To receive alerts for automated tasks, pass the notifications to a Trigger.

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

flyte.Trigger(
    name="daily-sync",
    automation=flyte.Cron("0 9 * * *"), # Every day at 9 AM
    notifications=[
        Slack(on_phase=ActionPhase.FAILED, webhook_url="...")
    ]
).run(my_sync_task)

Named Notifications (Remote Only)

If your Flyte platform has pre-configured notification rules or delivery channels, you can reference them by name. These are ignored during local execution.

• NamedRule: References a complete rule (phases + delivery) defined on the server.
• NamedDelivery: References a specific delivery channel (e.g., a shared Slack channel) while you define the phases locally.

from flyte.notify import NamedRule, NamedDelivery
from flyte.models import ActionPhase

# Use a server-side rule
rule = NamedRule(name="oncall-critical-alerts")

# Use a server-side delivery channel with local phase logic
delivery = NamedDelivery(on_phase=ActionPhase.FAILED, name="slack-main-channel")

Troubleshooting and Local Development

• Local Email: When running locally, Email notifications attempt to connect to an SMTP server at localhost:25. If no server is available, the SDK logs a warning but does not fail the task.
• Non-Blocking: Notification delivery is designed to be non-blocking. If a notification fails to send (e.g., due to a network error or invalid webhook URL), the error is logged as a warning, and the task execution continues.
• Missing Variables: If a template variable like {run.error} is used in a success notification where it doesn't exist, the placeholder will remain as literal text in the output.