ENG-3301: Add threading header support to email providers (Story 2)

changelog/8122-threading-header-support.yaml

@@ -0,0 +1,4 @@
+type: Added
+description: Added threading header support (Reply-To, Message-ID, In-Reply-To, References) and plaintext alternative to all email providers
+pr: 8122
+labels: []

src/fides/api/schemas/messaging/messaging.py

@@ -309,8 +309,14 @@ class EmailForActionType(BaseModel):
     """
 
     subject: str
-    body: str
+    body: str  # HTML body
     template_variables: Optional[Dict[str, Any]] = {}
+    # Threading / envelope fields (all optional, backward compatible)
+    reply_to: str | None = None
+    message_id: str | None = None  # RFC 5322 Message-ID
+    in_reply_to: str | None = None
+    references: str | None = None
+    body_text: str | None = None  # plaintext alternative for multipart/alternative
 
 
 class MessagingServiceDetails(Enum):

src/fides/api/service/messaging/messaging_providers/aws_ses_service.py

@@ -29,12 +29,12 @@ def get_identity_verification_attributes(
         self, Identities: list[str]
     ) -> dict[str, dict[str, dict[str, str]]]: ...
 
-    def send_email(
+    def send_raw_email(
         self,
         Source: str,
-        Destination: dict[str, list[str]],
-        Message: dict[str, Any],
-    ) -> None: ...
+        Destinations: list[str],
+        RawMessage: dict[str, bytes],
+    ) -> dict[str, Any]: ...
 
 
 def _sanitize_aws_error(exc: Exception) -> str:
@@ -133,10 +133,10 @@ def validate_on_save(self) -> None:
                 raise MessageDispatchException(f"{identity} is not verified in SES.")
 
     def send_email(self, to: str, message: EmailForActionType) -> None:
-        """Send an email using AWS SES simple API.
+        """Send an email using AWS SES raw API for custom header support.
 
-        Does NOT call validate_email_and_domain_status() — that is done at
-        config save/test time. SES rejects sends from unverified identities.
+        Builds a MIME message using ``email.message.EmailMessage`` (modern
+        Python 3.6+ API).
         """
         ses_client = self.get_ses_client()
 
@@ -149,14 +149,14 @@ def send_email(self, to: str, message: EmailForActionType) -> None:
             from_address = f"noreply@{self.details.domain}"
 
         try:
-            ses_client.send_email(
+            msg = self.build_mime(from_address, to.strip(), message)
+            ses_client.send_raw_email(
                 Source=from_address,
-                Destination={"ToAddresses": [to.strip()]},
-                Message={
-                    "Subject": {"Data": message.subject},
-                    "Body": {"Html": {"Data": message.body}},
-                },
+                Destinations=[to.strip()],
+                RawMessage={"Data": msg.as_bytes()},
             )
+        except MessageDispatchException:
+            raise
         except Exception as exc:
             logger.error("Email failed to send: {}", str(exc))
             raise MessageDispatchException(

src/fides/api/service/messaging/messaging_providers/base.py

@@ -1,4 +1,6 @@
 from abc import ABC, abstractmethod
+from email import policy as email_policy
+from email.message import EmailMessage
 from typing import ClassVar
 
 from loguru import logger
@@ -78,9 +80,56 @@ def _get_optional_secret(self, key: MessagingServiceSecrets) -> str | None:
 class BaseEmailProviderService(BaseMessageProviderService):
     """Base class for email provider services."""
 
+    HEADER_REPLY_TO = "Reply-To"
+    HEADER_MESSAGE_ID = "Message-ID"
+    HEADER_IN_REPLY_TO = "In-Reply-To"
+    HEADER_REFERENCES = "References"
+
     @abstractmethod
     def send_email(self, to: str, message: EmailForActionType) -> None: ...
 
+    @staticmethod
+    def get_threading_headers(
+        message: EmailForActionType, header_prefix: str = ""
+    ) -> dict[str, str]:
+        """Return non-None threading headers from the message."""
+        candidates = {
+            BaseEmailProviderService.HEADER_REPLY_TO: message.reply_to,
+            BaseEmailProviderService.HEADER_MESSAGE_ID: message.message_id,
+            BaseEmailProviderService.HEADER_IN_REPLY_TO: message.in_reply_to,
+            BaseEmailProviderService.HEADER_REFERENCES: message.references,
+        }
+        return {f"{header_prefix}{k}": v for k, v in candidates.items() if v}
+
+    @staticmethod
+    def build_mime(
+        from_address: str, to: str, message: EmailForActionType
+    ) -> EmailMessage:
+        """Build a MIME EmailMessage with optional threading headers.
+
+        Reusable by any provider that sends raw MIME (e.g., SES, SMTP).
+        """
+        msg = EmailMessage(policy=email_policy.SMTP)
+        msg["From"] = from_address
+        msg["To"] = to
+        msg["Subject"] = message.subject
+
+        for header, value in BaseEmailProviderService.get_threading_headers(
+            message
+        ).items():
+            msg[header] = value
+
+        if message.body_text:
+            # The RFC 2046 multipart/alternative spec says parts should be
+            # ordered from simplest to richest. The email client picks the
+            # richest part it can render successfully.
+            msg.set_content(message.body_text)
+            msg.add_alternative(message.body, subtype="html")
+        else:
+            msg.set_content(message.body, subtype="html")
+
+        return msg
+
 
 class BaseSMSProviderService(BaseMessageProviderService):
     """Base class for SMS provider services."""

src/fides/api/service/messaging/messaging_providers/mailchimp_transactional_service.py

@@ -28,15 +28,29 @@ def __init__(self, messaging_config: MessagingConfig):
         )
 
     def send_email(self, to: str, message: EmailForActionType) -> None:
+        msg_payload: dict = {
+            "from_email": self.from_email,
+            "subject": message.subject,
+            "html": message.body,
+            "to": [{"email": to.strip(), "type": "to"}],
+        }
+
+        # Reply-to uses Mandrill's native field
+        if message.reply_to:
+            msg_payload["reply_to"] = message.reply_to
+
+        # Threading headers (exclude Reply-To — handled natively above)
+        threading_headers = self.get_threading_headers(message)
+        threading_headers.pop(BaseEmailProviderService.HEADER_REPLY_TO, None)
+        if threading_headers:
+            msg_payload["headers"] = threading_headers
+        if message.body_text:
+            msg_payload["text"] = message.body_text
+
         data = json.dumps(
             {
                 "key": self.api_key,
-                "message": {
-                    "from_email": self.from_email,
-                    "subject": message.subject,
-                    "html": message.body,
-                    "to": [{"email": to.strip(), "type": "to"}],
-                },
+                "message": msg_payload,
             }
         )
 

src/fides/api/service/messaging/messaging_providers/mailgun_service.py

@@ -57,6 +57,11 @@ def send_email(self, to: str, message: EmailForActionType) -> None:
             else:
                 data["html"] = message.body
 
+            # Threading / envelope headers
+            data.update(self.get_threading_headers(message, header_prefix="h:"))
+            if message.body_text:
+                data["text"] = message.body_text
+
             response = requests.post(
                 f"{self.base_url}/{self.api_version}/{self.domain}/messages",
                 auth=("api", self.api_key),

src/fides/api/service/messaging/messaging_providers/twilio_email_service.py

@@ -2,7 +2,16 @@
 
 import sendgrid
 from loguru import logger
-from sendgrid.helpers.mail import Content, Email, Mail, Personalization, TemplateId, To
+from sendgrid.helpers.mail import (
+    Content,
+    Email,
+    Header,
+    Mail,
+    Personalization,
+    ReplyTo,
+    TemplateId,
+    To,
+)
 
 from fides.api.common_exceptions import MessageDispatchException
 from fides.api.models.messaging import MessagingConfig
@@ -41,9 +50,21 @@ def send_email(self, to: str, message: EmailForActionType) -> None:
             from_email = Email(self.from_email)
             to_email = To(to.strip())
             mail = self._compose_mail(
-                from_email, to_email, message.subject, message.body, template_id
+                from_email,
+                to_email,
+                message.subject,
+                message.body,
+                template_id,
+                body_text=message.body_text,
             )
 
+            # Threading / envelope headers
+            for key, value in self.get_threading_headers(message).items():
+                if key == BaseEmailProviderService.HEADER_REPLY_TO:
+                    mail.reply_to = ReplyTo(value)
+                else:
+                    mail.header = Header(key, value)
+
             response = sg.client.mail.send.post(request_body=mail.get())
             if response.status_code >= 400:
                 logger.error(
@@ -77,15 +98,26 @@ def _compose_mail(
         subject: str,
         message_body: str,
         template_id: str | None = None,
+        body_text: str | None = None,
     ) -> Mail:
-        """Composes a SendGrid Mail object, using a template if one exists."""
+        """Composes a SendGrid Mail object, using a template if one exists.
+
+        When body_text is provided, builds multipart/alternative with text/plain
+        before text/html (RFC 2046: last part is most preferred).
+        """
         if template_id:
             mail = Mail(from_email=from_email, subject=subject)
             mail.template_id = TemplateId(template_id)
             personalization = Personalization()
             personalization.dynamic_template_data = {"fides_email_body": message_body}
             personalization.add_email(to_email)
             mail.add_personalization(personalization)
+        elif body_text:
+            mail = Mail(from_email=from_email, subject=subject)
+            mail.add_personalization(Personalization())
+            mail.personalizations[0].add_email(to_email)
+            mail.content = Content("text/plain", body_text)
+            mail.content = Content("text/html", message_body)
         else:
             content = Content("text/html", message_body)
             mail = Mail(from_email, to_email, subject, content)