Email Infrastructure for AI-Driven Marketing Automation

import multimail

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

def run_drip_sequence(lead_email: str, mailbox: str, sequence: list[dict]):
    "cm"># Check current sequence position via tags
    contacts = client.manage_contacts(query=lead_email, mailbox=mailbox)
    if not contacts:
        return
    
    contact = contacts[0]
    existing_tags = contact.get("tags", [])
    
    "cm"># Skip if already completed or unsubscribed
    if "drip_complete" in existing_tags or "unsubscribed" in existing_tags:
        return
    
    "cm"># Determine next step
    sent_steps = [t for t in existing_tags if t.startswith("drip_step_")]
    next_step_index = len(sent_steps)
    
    if next_step_index >= len(sequence):
        client.tag_email(contact_id=contact["id"], tags=["drip_complete"])
        return
    
    step = sequence[next_step_index]
    
    "cm"># Send next step — monitored mode, no approval needed
    result = client.send_email(
        mailbox=mailbox,
        to=lead_email,
        subject=step["subject"],
        body=step["body"],
        tags=[f"drip_step_{next_step_index}", "campaign_drip"]
    )
    
    print(f"Sent drip step {next_step_index} to {lead_email}: {result[&"cm">#039;message_id']}")

# Example sequence
sequence = [
    {"subject": "Welcome to the platform", "body": "Here&"cm">#039;s how to get started..."},
    {"subject": "Your first integration guide", "body": "Most teams start with..."},
    {"subject": "Are you getting value?", "body": "Here&"cm">#039;s what teams like yours typically do next..."}
]

run_drip_sequence("[email protected]", "[email protected]", sequence)

import multimail
from datetime import datetime, timedelta

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

def queue_winback_drafts(churned_accounts: list[dict], mailbox: str):
    queued = []
    
    for account in churned_accounts:
        days_since_churn = (datetime.now() - account["churned_at"]).days
        
        "cm"># Personalize based on churn duration
        if days_since_churn < 30:
            subject = f"We noticed you left — here&"cm">#039;s what changed"
            body = f"Hi {account[&"cm">#039;first_name']},\n\nIt's been {days_since_churn} days since your subscription ended..."
        else:
            subject = f"A lot has changed since you left"
            body = f"Hi {account[&"cm">#039;first_name']},\n\nWe've shipped significant updates since you last used the platform..."
        
        # Queue as pending (gated_send mode — agent drafts, human approves)
        result = client.send_email(
            mailbox=mailbox,
            to=account["email"],
            subject=subject,
            body=body,
            tags=["win_back", f"churn_day_{days_since_churn}"],
            oversight_mode="gated_send"  # holds for approval
        )
        queued.append(result["message_id"])
    
    print(f"Queued {len(queued)} win-back drafts for review")
    return queued

def review_winback_batch(mailbox: str):
    # Marketing ops reviews pending sends
    pending = client.list_pending(mailbox=mailbox, tag="win_back")
    print(f"{len(pending[&"cm">#039;messages'])} win-back messages awaiting approval")
    
    for msg in pending["messages"]:
        print(f"  To: {msg[&"cm">#039;to']} | Subject: {msg['subject']}")
        # In practice: route to approval UI
        # client.decide_email(message_id=msg["id"], decision="approve")
        # client.decide_email(message_id=msg["id"], decision="reject")

import multimail
import re

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

BUYING_SIGNALS = [
    r"\bpric(e|ing)\b",
    r"\bdemo\b",
    r"\btrial\b",
    r"\bbudget\b",
    r"\bcontract\b",
    r"\bteam size\b",
]

def qualify_inbound_replies(mailbox: str):
    inbox = client.check_inbox(
        mailbox=mailbox,
        filter={"tags": ["campaign_drip"], "unread": True}
    )
    
    qualified = []
    
    for email_summary in inbox["emails"]:
        email = client.read_email(
            mailbox=mailbox,
            message_id=email_summary["id"]
        )
        
        body_lower = email["body"].lower()
        matched_signals = [
            pattern for pattern in BUYING_SIGNALS
            if re.search(pattern, body_lower)
        ]
        
        if len(matched_signals) >= 2:
            "cm"># Tag as MQL — buying intent detected
            client.tag_email(
                message_id=email["id"],
                tags=["mql", "buying_intent", f"signals_{len(matched_signals)}"]
            )
            qualified.append({
                "from": email["from"],
                "signals": matched_signals,
                "thread_id": email["thread_id"]
            })
            print(f"MQL: {email[&"cm">#039;from']} — {len(matched_signals)} buying signals detected")
    
    return qualified

qualify_inbound_replies("[email protected]")

# MCP tool sequence for feedback request campaign
# Run from Claude Desktop or any MCP-compatible client

# Step 1: Send feedback request to recent customer
send_email(
  mailbox="[email protected]",
  to="[email protected]",
  subject="Quick question about your experience",
  body="Hi Sarah,\n\nYou've been using the platform for 30 days now. We'd value your honest take — what's working well, and what's frustrating?\n\nThis goes directly to our product team, not a survey form.\n\nThanks,\nThe Team",
  tags=["nps_request", "cohort_2026_q2"]
)

# Step 2: Check for replies (run 3-5 days later)
check_inbox(
  mailbox="[email protected]",
  filter={"tags": ["nps_request"], "has_replies": true}
)

# Step 3: Read full thread to extract sentiment
get_thread(
  mailbox="[email protected]",
  thread_id="<thread_id_from_step_2>"
)

# Step 4: Tag by sentiment for downstream routing
tag_email(
  message_id="<reply_message_id>",
  tags=["nps_response", "sentiment_positive"]
)

import multimail

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

def send_product_announcement(recipients: list[str], announcement: dict, mailbox: str):
    results = []
    
    for recipient in recipients:
        result = client.send_email(
            mailbox=mailbox,
            to=recipient,
            subject=announcement["subject"],
            body=f"""{announcement[&"cm">#039;body']}

---
You&"cm">#039;re receiving this because you're a {announcement['product_name']} user.
Unsubscribe: https://yourdomain.com/unsubscribe?email={recipient}
Your Company, Inc. · 123 Main St, San Francisco CA 94105
""",
            headers={
                "List-Unsubscribe": f"<mailto:[email protected]?subject=unsubscribe>, <https://yourdomain.com/unsubscribe?email={recipient}>",
                "List-Unsubscribe-Post": "List-Unsubscribe=One-Click",
                "X-Campaign-ID": announcement["campaign_id"]
            },
            tags=["product_announcement", announcement["campaign_id"]]
        )
        results.append(result["message_id"])
    
    print(f"Sent announcement to {len(results)} recipients")
    return results

announcement = {
    "subject": "New: Bulk export and custom webhook routing",
    "body": "We shipped two features this week that came directly from your feedback...",
    "product_name": "Platform",
    "campaign_id": "announce_2026_q2_bulk_export"
}

send_product_announcement(
    recipients=["[email protected]", "[email protected]"],
    announcement=announcement,
    mailbox="[email protected]"
)