Skip to content


The prefect-email library helps you send emails from your Prefect flows.

Getting Started


  • Prefect installed in a virtual environment.
  • Many email services, such as Gmail, require an App Password to successfully send emails. If you encounter an error similar to smtplib.SMTPAuthenticationError: (535, b'5.7.8 Username and Password not accepted..., it's likely you are not using an App Password.

Install prefect-email

pip install -U prefect-email

Register newly installed blocks types

Register the block types in the prefect-email module to make them available for use.

prefect block register -m prefect_email

Save credentials to an EmailServerCredentials block

Save your email credentials to a block. Replace the placeholders with your email address and password.

from prefect_email import EmailServerCredentials

credentials = EmailServerCredentials(
    password="PASSWORD-PLACEHOLDER",  # must be an app password

In the examples below you load a credentials block to authenticate with the email server.

Send emails

The code below shows how to send an email using the pre-built email_send_message task.

from prefect import flow
from prefect_email import EmailServerCredentials, email_send_message

def example_email_send_message_flow(email_addresses):
    email_server_credentials = EmailServerCredentials.load("BLOCK-NAME-PLACEHOLDER")
    for email_address in email_addresses:
        subject = email_send_message.with_options(name=f"email {email_address}").submit(
            subject="Example Flow Notification using Gmail",
            msg="This proves email_send_message works!",

if __name__ == "__main__":

Capture exceptions and send an email

This example demonstrates how to send an email notification with the details of the exception when a flow run fails.

prefect-email can be wrapped in an except statement to do just that!

from prefect import flow
from prefect.context import get_run_context
from prefect_email import EmailServerCredentials, email_send_message

def notify_exc_by_email(exc):
    context = get_run_context()
    flow_run_name =
    email_server_credentials = EmailServerCredentials.load("email-server-credentials")
        subject=f"Flow run {flow_run_name!r} failed",
        msg=f"Flow run {flow_run_name!r} failed due to {exc}.",

def example_flow():
        1 / 0
    except Exception as exc:

if __name__ == "__main__":


Refer to the prefect-email API documentation linked in the sidebar to explore all the capabilities of the prefect-email library.