How to Integrate Replit with Mailchimp

1import os
2import hashlib
3import requests
4from typing import Optional
5
6API_KEY = os.environ["MAILCHIMP_API_KEY"]
7SERVER = os.environ["MAILCHIMP_SERVER_PREFIX"]
8LIST_ID = os.environ["MAILCHIMP_LIST_ID"]
9BASE_URL = f"https://{SERVER}.api.mailchimp.com/3.0"
10
11# Mailchimp uses Basic Auth: any username + API key as password
12AUTH = ("anystring", API_KEY)
13
14def get_subscriber_hash(email: str) -> str:
15    """Mailchimp identifies subscribers by MD5 hash of lowercase email."""
16    return hashlib.md5(email.lower().encode()).hexdigest()
17
18def subscribe_contact(email: str, first_name: str = "", last_name: str = "") -> dict:
19    """Add or update a subscriber in the audience (upsert via PATCH)."""
20    subscriber_hash = get_subscriber_hash(email)
21    url = f"{BASE_URL}/lists/{LIST_ID}/members/{subscriber_hash}"
22    data = {
23        "email_address": email,
24        "status_if_new": "subscribed",  # Status for new subscribers
25        "status": "subscribed",          # Status for existing subscribers
26        "merge_fields": {
27            "FNAME": first_name,
28            "LNAME": last_name
29        }
30    }
31    response = requests.put(url, json=data, auth=AUTH)
32    response.raise_for_status()
33    return response.json()
34
35def unsubscribe_contact(email: str) -> dict:
36    """Unsubscribe a contact (does not delete — preserves history)."""
37    subscriber_hash = get_subscriber_hash(email)
38    url = f"{BASE_URL}/lists/{LIST_ID}/members/{subscriber_hash}"
39    response = requests.patch(url, json={"status": "unsubscribed"}, auth=AUTH)
40    response.raise_for_status()
41    return response.json()
42
43def get_contact_status(email: str) -> Optional[str]:
44    """Check if an email is subscribed, unsubscribed, or not in the list."""
45    subscriber_hash = get_subscriber_hash(email)
46    url = f"{BASE_URL}/lists/{LIST_ID}/members/{subscriber_hash}"
47    response = requests.get(url, auth=AUTH)
48    if response.status_code == 404:
49        return None  # Not in list
50    response.raise_for_status()
51    return response.json().get("status")
52
53def add_tag_to_contact(email: str, tag_name: str) -> None:
54    """Add a tag to a subscriber for segmentation."""
55    subscriber_hash = get_subscriber_hash(email)
56    url = f"{BASE_URL}/lists/{LIST_ID}/members/{subscriber_hash}/tags"
57    data = {"tags": [{"name": tag_name, "status": "active"}]}
58    response = requests.post(url, json=data, auth=AUTH)
59    response.raise_for_status()
60
61# Example usage
62if __name__ == "__main__":
63    result = subscribe_contact("user@example.com", "Jane", "Doe")
64    print(f"Subscribed: {result['email_address']} — status: {result['status']}")
65
66    add_tag_to_contact("user@example.com", "early-adopter")
67    print("Tag added successfully")
68
69    status = get_contact_status("user@example.com")
70    print(f"Current status: {status}")

1const express = require('express');
2const mailchimp = require('@mailchimp/mailchimp_marketing');
3const crypto = require('crypto');
4
5const app = express();
6app.use(express.json());
7
8// Configure Mailchimp SDK from Replit Secrets
9mailchimp.setConfig({
10  apiKey: process.env.MAILCHIMP_API_KEY,
11  server: process.env.MAILCHIMP_SERVER_PREFIX
12});
13
14const LIST_ID = process.env.MAILCHIMP_LIST_ID;
15
16function getSubscriberHash(email) {
17  return crypto.createHash('md5').update(email.toLowerCase()).digest('hex');
18}
19
20// Verify API connection on startup
21async function checkConnection() {
22  try {
23    const response = await mailchimp.ping.get();
24    console.log('Mailchimp connected:', response.health_status);
25  } catch (err) {
26    console.error('Mailchimp connection failed:', err.response?.text);
27  }
28}
29checkConnection();
30
31// Subscribe or update a contact
32app.post('/subscribe', async (req, res) => {
33  const { email, firstName, lastName, tags } = req.body;
34  if (!email) return res.status(400).json({ error: 'email is required' });
35
36  const subscriberHash = getSubscriberHash(email);
37  try {
38    const response = await mailchimp.lists.setListMember(LIST_ID, subscriberHash, {
39      email_address: email,
40      status_if_new: 'subscribed',
41      status: 'subscribed',
42      merge_fields: {
43        FNAME: firstName || '',
44        LNAME: lastName || ''
45      },
46      tags: tags || []
47    });
48    res.json({ success: true, id: response.id, status: response.status });
49  } catch (err) {
50    console.error('Subscribe error:', err.response?.text);
51    res.status(500).json({ error: err.message });
52  }
53});
54
55// Unsubscribe a contact
56app.post('/unsubscribe', async (req, res) => {
57  const { email } = req.body;
58  if (!email) return res.status(400).json({ error: 'email is required' });
59
60  const subscriberHash = getSubscriberHash(email);
61  try {
62    await mailchimp.lists.updateListMember(LIST_ID, subscriberHash, {
63      status: 'unsubscribed'
64    });
65    res.json({ success: true });
66  } catch (err) {
67    res.status(500).json({ error: err.message });
68  }
69});
70
71// Create and send a campaign
72app.post('/campaign', async (req, res) => {
73  const { subject, previewText, fromName, replyTo, htmlContent } = req.body;
74  try {
75    // Step 1: Create campaign
76    const campaign = await mailchimp.campaigns.create({
77      type: 'regular',
78      recipients: { list_id: LIST_ID },
79      settings: {
80        subject_line: subject,
81        preview_text: previewText || '',
82        from_name: fromName,
83        reply_to: replyTo
84      }
85    });
86
87    // Step 2: Set campaign content
88    await mailchimp.campaigns.setContent(campaign.id, {
89      html: htmlContent
90    });
91
92    // Step 3: Send campaign
93    await mailchimp.campaigns.send(campaign.id);
94    res.json({ success: true, campaignId: campaign.id });
95  } catch (err) {
96    console.error('Campaign error:', err.response?.text);
97    res.status(500).json({ error: err.message });
98  }
99});
100
101app.listen(3000, '0.0.0.0', () => {
102  console.log('Mailchimp integration server running on port 3000');
103});

1# Flask webhook receiver for Mailchimp events
2from flask import Flask, request, jsonify
3import os
4
5app = Flask(__name__)
6
7@app.route('/mailchimp/webhook', methods=['GET', 'POST'])
8def mailchimp_webhook():
9    # Mailchimp sends a GET request to verify the endpoint
10    if request.method == 'GET':
11        return 'OK', 200
12
13    # POST contains the event data
14    data = request.form  # Mailchimp sends form-encoded data, not JSON
15    event_type = data.get('type')
16    email = data.get('data[email]')
17    list_id = data.get('data[list_id]')
18
19    print(f"Mailchimp webhook: {event_type} for {email}")
20
21    if event_type == 'unsubscribe':
22        # Update your database to mark user as unsubscribed
23        # update_user_subscription_status(email, subscribed=False)
24        print(f"User unsubscribed: {email}")
25    elif event_type == 'subscribe':
26        print(f"User subscribed: {email}")
27    elif event_type == 'profile':
28        new_email = data.get('data[new_email]')
29        print(f"Profile updated: {email} -> {new_email}")
30
31    return jsonify({'status': 'received'}), 200
32
33if __name__ == '__main__':
34    app.run(host='0.0.0.0', port=3000)