Stop Bounces Before They Happen

import multimail
from multimail.exceptions import DomainBlockedError, SuppressedAddressError

client = multimail.Client(api_key="mm_live_...")

def send_with_domain_check(mailbox_id, recipient, subject, body):
    try:
        result = client.send_email(
            mailbox_id=mailbox_id,
            to=recipient,
            subject=subject,
            body=body,
        )
        if result.domain_risk == "yellow":
            "cm"># Flagged but not blocked — agent decides
            print(f"Yellow domain signals: {result.domain_signals}")
        return result

    except DomainBlockedError as e:
        "cm"># Red domain — blocked before send, zero bounce impact
        print(f"Blocked: {e.recipient} — reason: {e.reason}")
        return None

    except SuppressedAddressError as e:
        "cm"># Prior hard bounce — suppressed at platform level
        print(f"Suppressed: {e.recipient} — suppressed_at: {e.suppressed_at}")
        return None

result = send_with_domain_check(
    mailbox_id="mbx_01JDEF456",
    recipient="[email protected]",
    subject="Q2 update from Acme",
    body="Hi, wanted to share our Q2 summary with you.",
)

import { createHmac } from 'crypto';

export async function handleBounceWebhook(req, res) {
  const sig = req.headers['x-multimail-signature'];
  const expected = createHmac('sha256', process.env.MM_WEBHOOK_SECRET)
    .update(JSON.stringify(req.body))
    .digest('hex');

  if (sig !== expected) {
    return res.status(401).json({ error: 'invalid_signature' });
  }

  const event = req.body;

  if (event.type === 'email.bounced') {
    const { recipient, bounce_type, smtp_code, message_id, suppressed_at } = event.data;

    if (bounce_type === 'hard') {
      "cm">// Platform has already suppressed this address.
      "cm">// Sync the event to your CRM or contact store.
      await markContactInvalid(recipient, {
        reason: 'hard_bounce',
        smtp_code,
        bounced_message_id: message_id,
        suppressed_at,
      });
      console.log(`Hard bounce suppressed: ${recipient} (SMTP ${smtp_code})`);
    } else if (bounce_type === 'soft') {
      "cm">// Temporary failure — platform will retry automatically.
      "cm">// Log for monitoring but no action needed.
      console.log(`Soft bounce on ${message_id}, smtp_code=${smtp_code}`);
    }
  }

  res.status(200).json({ received: true });
}

"cm"># Validate addresses before campaign send
curl -X POST https://api.multimail.dev/v1/suppression/check \
  -H "Authorization: Bearer $MULTIMAIL_API_KEY..." \
  -H "Content-Type: application/json" \
  -d &"cm">#039;{
    "addresses": [
      "[email protected]",
      "[email protected]",
      "[email protected]"
    ]
  }&"cm">#039;

"cm"># Response:
"cm"># {
"cm">#   "results": [
"cm">#     {
"cm">#       "address": "[email protected]",
"cm">#       "suppressed": false,
"cm">#       "domain_risk": "green"
"cm">#     },
"cm">#     {
"cm">#       "address": "[email protected]",
"cm">#       "suppressed": true,
"cm">#       "reason": "hard_bounce",
"cm">#       "domain_risk": "red",
"cm">#       "signals": ["no_mx_records", "platform_bounce_rate_exceeded"]
"cm">#     },
"cm">#     {
"cm">#       "address": "[email protected]",
"cm">#       "suppressed": false,
"cm">#       "domain_risk": "yellow",
"cm">#       "signals": ["missing_dmarc"]
"cm">#     }
"cm">#   ],
"cm">#   "suppressed_count": 1,
"cm">#   "sendable_count": 2
"cm"># }

import multimail
from multimail.exceptions import DomainBlockedError, SuppressedAddressError

client = multimail.Client(
    api_key="mm_live_...",
    oversight_mode="autonomous",
)

def run_bounce_safe_campaign(mailbox_id, contacts, subject, body_template):
    results = {"sent": [], "blocked": [], "suppressed": [], "yellow_flagged": []}

    for contact in contacts:
        try:
            result = client.send_email(
                mailbox_id=mailbox_id,
                to=contact["email"],
                subject=subject,
                body=body_template.format(name=contact["name"]),
            )
            if result.domain_risk == "yellow":
                results["yellow_flagged"].append({
                    "address": contact["email"],
                    "signals": result.domain_signals,
                })
            else:
                results["sent"].append(contact["email"])

        except DomainBlockedError as e:
            results["blocked"].append({
                "address": contact["email"],
                "reason": e.reason,
                "signals": e.domain_signals,
            })

        except SuppressedAddressError as e:
            results["suppressed"].append(contact["email"])

    print(f"Sent: {len(results[&"cm">#039;sent'])}")
    print(f"Blocked (red domain): {len(results[&"cm">#039;blocked'])}")
    print(f"Suppressed (prior bounce): {len(results[&"cm">#039;suppressed'])}")
    print(f"Flagged (yellow domain): {len(results[&"cm">#039;yellow_flagged'])}")
    return results