1pip install mailchimp-marketing

1export MAILCHIMP_API_KEY="your-api-key-here"
2export MAILCHIMP_DC="us21"

1import os
2import mailchimp_marketing as MailchimpMarketing
3from mailchimp_marketing.api_client import ApiClientError
4
5def get_mailchimp_client():
6    """Initialize and return an authenticated Mailchimp client."""
7    client = MailchimpMarketing.Client()
8    client.set_config({
9        "api_key": os.environ.get("MAILCHIMP_API_KEY"),
10        "server": os.environ.get("MAILCHIMP_DC", "us21"),
11    })
12
13    # Verify connection
14    try:
15        response = client.ping.get()
16        print(f"Mailchimp connected: {response}")
17    except ApiClientError as error:
18        print(f"Connection failed: {error.text}")
19        raise
20
21    return client
22
23client = get_mailchimp_client()
24# Output: Mailchimp connected: {'health_status': "Everything's Chimpy!"}

1def create_audience(client, name: str, from_name: str, reply_to: str) -> str:
2    """
3    Create a new Mailchimp audience.
4    Returns the list_id of the newly created audience.
5    """
6    try:
7        response = client.lists.create_list({
8            "name": name,
9            "permission_reminder": "You signed up for updates from our website.",
10            "email_type_option": False,
11            "contact": {
12                "company": "Your Company Name",
13                "address1": "123 Main Street",
14                "city": "San Francisco",
15                "state": "CA",
16                "zip": "94102",
17                "country": "US",
18            },
19            "campaign_defaults": {
20                "from_name": from_name,
21                "from_email": reply_to,
22                "subject": "",
23                "language": "EN",
24            },
25        })
26        list_id = response["id"]
27        print(f"Audience created: {name} (ID: {list_id})")
28        return list_id
29    except ApiClientError as error:
30        print(f"Error creating audience: {error.text}")
31        raise
32
33# Create a drip campaign audience
34list_id = create_audience(
35    client=client,
36    name="Product Launch Drip Sequence",
37    from_name="Jennifer from AutomateMyJob",
38    reply_to="jennifer@automatemyjob.com"
39)

1import hashlib
2
3def add_subscriber(client, list_id: str, email: str, first_name: str, last_name: str,
4                   tags: list = None) -> dict:
5    """Add or update a subscriber in the audience."""
6    # Mailchimp uses MD5 hash of lowercase email as subscriber ID
7    subscriber_hash = hashlib.md5(email.lower().encode()).hexdigest()
8
9    member_data = {
10        "email_address": email,
11        "status": "subscribed",
12        "merge_fields": {
13            "FNAME": first_name,
14            "LNAME": last_name,
15        },
16    }
17
18    if tags:
19        member_data["tags"] = [{"name": tag, "status": "active"} for tag in tags]
20
21    try:
22        response = client.lists.set_list_member(list_id, subscriber_hash, member_data)
23        print(f"Added: {email} ({response['status']})")
24        return response
25    except ApiClientError as error:
26        print(f"Error adding subscriber {email}: {error.text}")
27        raise
28
29# Add a subscriber with tags for segmentation
30add_subscriber(
31    client=client,
32    list_id=list_id,
33    email="jane.doe@example.com",
34    first_name="Jane",
35    last_name="Doe",
36    tags=["trial-user", "source-webinar"]
37)

1def batch_add_subscribers(client, list_id: str, subscribers: list[dict]) -> dict:
2    """
3    Batch add subscribers. Each subscriber dict should have:
4    email, first_name, last_name, and optionally tags (list of strings).
5    """
6    members = []
7    for sub in subscribers:
8        member = {
9            "email_address": sub["email"],
10            "status": "subscribed",
11            "merge_fields": {
12                "FNAME": sub.get("first_name", ""),
13                "LNAME": sub.get("last_name", ""),
14            },
15        }
16        if sub.get("tags"):
17            member["tags"] = [{"name": t, "status": "active"} for t in sub["tags"]]
18        members.append(member)
19
20    response = client.lists.batch_list_members(list_id, {
21        "members": members,
22        "update_existing": True,  # Update instead of error on duplicate email
23    })
24
25    print(f"Added: {response['new_members']} | Updated: {response['updated_members']} "
26          f"| Errors: {len(response['errors'])}")
27    return response
28
29# Example usage
30new_subscribers = [
31    {"email": "alice@example.com", "first_name": "Alice", "last_name": "Smith",
32     "tags": ["trial-user"]},
33    {"email": "bob@example.com", "first_name": "Bob", "last_name": "Jones",
34     "tags": ["paid-user", "source-organic"]},
35]
36batch_add_subscribers(client, list_id, new_subscribers)

1def create_drip_campaign(client, list_id: str, subject: str, preview_text: str,
2                         from_name: str, reply_to: str, html_content: str,
3                         schedule_time: str = None) -> str:
4    """
5    Create a Mailchimp campaign.
6    schedule_time: ISO 8601 datetime string (e.g., '2026-03-05T09:00:00+00:00')
7                   Pass None to save as draft.
8    Returns the campaign_id.
9    """
10    # Step 1: Create the campaign object
11    try:
12        campaign = client.campaigns.create({
13            "type": "regular",
14            "recipients": {"list_id": list_id},
15            "settings": {
16                "subject_line": subject,
17                "preview_text": preview_text,
18                "title": f"Drip - {subject}",
19                "from_name": from_name,
20                "reply_to": reply_to,
21                "auto_footer": False,
22            },
23        })
24        campaign_id = campaign["id"]
25        print(f"Campaign created: {campaign_id}")
26    except ApiClientError as error:
27        print(f"Error creating campaign: {error.text}")
28        raise
29
30    # Step 2: Set the email content
31    try:
32        client.campaigns.set_content(campaign_id, {"html": html_content})
33        print(f"Content set for campaign: {campaign_id}")
34    except ApiClientError as error:
35        print(f"Error setting content: {error.text}")
36        raise
37
38    # Step 3: Schedule or keep as draft
39    if schedule_time:
40        try:
41            client.campaigns.schedule(campaign_id, {"schedule_time": schedule_time})
42            print(f"Campaign scheduled for: {schedule_time}")
43        except ApiClientError as error:
44            print(f"Error scheduling campaign: {error.text}")
45            raise
46
47    return campaign_id

1from datetime import datetime, timedelta, timezone
2
3def build_drip_sequence(client, list_id: str, base_date: datetime):
4    """
5    Create a complete 4-email drip sequence starting from base_date.
6    base_date: datetime object for when the sequence should begin (Day 0).
7    """
8    from_name = "Jennifer at AutomateMyJob"
9    reply_to = "jennifer@automatemyjob.com"
10
11    emails = [
12        {
13            "day_offset": 0,
14            "subject": "Welcome to AutomateMyJob — here's what to expect 👋",
15            "preview_text": "Your automation journey starts right now.",
16            "html": """
17                <h1>Welcome, *|FNAME|*!</h1>
18                <p>You've just taken the best step toward getting your time back.</p>
19                <p>Over the next two weeks, we'll send you our most practical automation guides — no fluff, just tools and techniques that work immediately.</p>
20                <p>First up: check out our <a href="https://automatemyjob.com/posts/no-code-automation-tools-beginners">beginner's guide to no-code automation</a> to see where to start.</p>
21                <p>See you in a few days,<br>Jennifer</p>
22            """,
23        },
24        {
25            "day_offset": 3,
26            "subject": "The one automation that saves 5 hours a week (Day 3)",
27            "preview_text": "Most people overlook this completely.",
28            "html": """
29                <h1>Hey *|FNAME|*, Day 3 tip incoming 🔧</h1>
30                <p>If you do only one thing this week, automate your email sorting.</p>
31                <p>With three Gmail filters and 20 minutes of setup, you can cut your
32                   daily email time by 30%. Here's exactly how to do it...</p>
33                <p><a href="https://automatemyjob.com">Read the full guide →</a></p>
34                <p>Jennifer</p>
35            """,
36        },
37        {
38            "day_offset": 7,
39            "subject": "How Sarah saved 12 hours a week with Python automation",
40            "preview_text": "Real story, real numbers.",
41            "html": """
42                <h1>A quick story, *|FNAME|* 📖</h1>
43                <p>Sarah was spending 3 hours every Monday manually compiling reports
44                   from 6 different spreadsheets.</p>
45                <p>One Python script later: 8 minutes. Every week. Automatically.</p>
46                <p>We broke down exactly how she did it in this tutorial: <a href="https://automatemyjob.com/posts/automate-weekly-reports-python-ai-tutorial">Automate Your Weekly Reports with Python</a></p>
47                <p>You can do the same thing. No prior Python experience needed.</p>
48                <p>Jennifer</p>
49            """,
50        },
51        {
52            "day_offset": 14,
53            "subject": "You've been with us 2 weeks — here's what comes next",
54            "preview_text": "Ready to take automation seriously?",
55            "html": """
56                <h1>Two weeks in, *|FNAME|* 🎯</h1>
57                <p>You've been learning about automation for two weeks now.
58                   Here's a question: what's the one task you do every week
59                   that you'd most like to eliminate?</p>
60                <p>Reply to this email and tell us. We read every response and
61                   often turn reader questions into tutorials.</p>
62                <p>And if you're ready to go deeper, our full automation course
63                   covers everything from beginner scripts to enterprise AI agents.
64                   <a href="https://automatemyjob.com">Explore the curriculum →</a>
65                </p>
66                <p>Thanks for being here,<br>Jennifer</p>
67            """,
68        },
69    ]
70
71    campaign_ids = []
72    for email in emails:
73        send_time = base_date + timedelta(days=email["day_offset"])
74        # Schedule at 9 AM UTC
75        send_time = send_time.replace(hour=9, minute=0, second=0, microsecond=0,
76                                      tzinfo=timezone.utc)
77        schedule_str = send_time.strftime("%Y-%m-%dT%H:%M:%S+00:00")
78
79        campaign_id = create_drip_campaign(
80            client=client,
81            list_id=list_id,
82            subject=email["subject"],
83            preview_text=email["preview_text"],
84            from_name=from_name,
85            reply_to=reply_to,
86            html_content=email["html"],
87            schedule_time=schedule_str,
88        )
89        campaign_ids.append(campaign_id)
90        print(f"Day {email['day_offset']} email scheduled: {schedule_str}")
91
92    return campaign_ids
93
94# Launch the sequence starting tomorrow
95start_date = datetime.now(timezone.utc) + timedelta(days=1)
96campaign_ids = build_drip_sequence(client, list_id, start_date)
97print(f"\nDrip sequence ready: {len(campaign_ids)} campaigns scheduled")

1def get_campaign_stats(client, campaign_id: str) -> dict:
2    """Retrieve open rate, click rate, and key metrics for a campaign."""
3    try:
4        report = client.reports.get_campaign_report(campaign_id)
5        stats = {
6            "campaign_id": campaign_id,
7            "subject": report["campaign_title"],
8            "emails_sent": report["emails_sent"],
9            "open_rate": round(report["opens"]["open_rate"] * 100, 2),
10            "click_rate": round(report["clicks"]["click_rate"] * 100, 2),
11            "unsubscribe_count": report["unsubscribes"]["unsubscribe_count"],
12            "bounce_count": report["bounces"]["hard_bounces"] + report["bounces"]["soft_bounces"],
13        }
14        return stats
15    except ApiClientError as error:
16        print(f"Error fetching report for {campaign_id}: {error.text}")
17        return {}
18
19def get_drip_sequence_report(client, campaign_ids: list[str]) -> list[dict]:
20    """Get a full performance report for every email in the drip sequence."""
21    all_stats = []
22    for i, campaign_id in enumerate(campaign_ids):
23        stats = get_campaign_stats(client, campaign_id)
24        if stats:
25            stats["sequence_position"] = i + 1
26            all_stats.append(stats)
27            print(f"Email {i+1}: {stats['open_rate']}% open rate | "
28                  f"{stats['click_rate']}% click rate | "
29                  f"{stats['emails_sent']} sent")
30    return all_stats
31
32# Pull the report after campaigns have sent
33report = get_drip_sequence_report(client, campaign_ids)

1def get_non_openers(client, campaign_id: str, list_id: str) -> list[str]:
2    """
3    Return email addresses of subscribers who were sent a campaign but didn't open it.
4    Useful for sending a re-engagement or different follow-up to non-openers.
5    """
6    try:
7        # Get all members who were sent this campaign
8        sent_report = client.reports.get_email_activity_for_campaign(
9            campaign_id, fields=["activity.email_address", "activity.action"]
10        )
11
12        non_openers = []
13        for member in sent_report.get("emails", []):
14            actions = [a["action"] for a in member.get("activity", [])]
15            if "open" not in actions:
16                non_openers.append(member["email_address"])
17
18        print(f"Non-openers for campaign {campaign_id}: {len(non_openers)}")
19        return non_openers
20
21    except ApiClientError as error:
22        print(f"Error fetching activity: {error.text}")
23        return []
24
25def tag_subscribers(client, list_id: str, emails: list[str], tag_name: str):
26    """Apply a tag to a list of subscribers for segmentation."""
27    members = []
28    for email in emails:
29        subscriber_hash = hashlib.md5(email.lower().encode()).hexdigest()
30        members.append({
31            "email_address": email,
32            "status": "subscribed",
33            "tags": [{"name": tag_name, "status": "active"}]
34        })
35
36    # Use batch update for efficiency
37    client.lists.batch_list_members(list_id, {
38        "members": members,
39        "update_existing": True,
40    })
41    print(f"Tagged {len(emails)} subscribers with '{tag_name}'")
42
43# Tag non-openers for a re-engagement campaign
44non_openers = get_non_openers(client, campaign_ids[0], list_id)
45if non_openers:
46    tag_subscribers(client, list_id, non_openers, "day1-non-opener")

1import time
2import json
3
4def create_campaign_with_retry(client, payload: dict, max_retries: int = 3) -> dict:
5    """
6    Create a campaign with exponential backoff retry on transient errors.
7    Retries on 429 (rate limit) and 5xx (server errors). Raises on 4xx.
8    """
9    for attempt in range(max_retries):
10        try:
11            return client.campaigns.create(payload)
12        except ApiClientError as error:
13            error_data = json.loads(error.text)
14            status_code = error_data.get("status", 0)
15
16            if status_code == 429:
17                # Rate limited — wait and retry
18                wait_time = 2 ** attempt
19                print(f"Rate limited. Waiting {wait_time}s before retry {attempt + 1}")
20                time.sleep(wait_time)
21            elif status_code >= 500:
22                # Server error — retry
23                wait_time = 2 ** attempt
24                print(f"Server error ({status_code}). Retrying in {wait_time}s")
25                time.sleep(wait_time)
26            else:
27                # Client error (400, 401, 404) — don't retry, raise immediately
28                print(f"Client error ({status_code}): {error_data.get('detail')}")
29                raise
30
31    raise RuntimeError(f"Failed after {max_retries} attempts")