millionsend

import millionsend

millionsend.api_key = "ms_123"
millionsend.base_url = "https://mail.acme.dev"  # self-hosted only; omit for MillionSend Cloud

email = millionsend.Emails.send({
    "from": "Acme <[email protected]>",
    "to": "[email protected]",
    "subject": "Hello from MillionSend",
    "html": "<strong>It works!</strong>",
})

print(email.id)  # responses support both email["id"] and email.id

import millionsend

millionsend.api_key  = "ms_123"                  # or env MILLIONSEND_API_KEY
millionsend.base_url = "https://mail.acme.dev"   # or env MILLIONSEND_BASE_URL
millionsend.timeout  = 30                        # optional, seconds (default 60)
millionsend.allow_insecure_http = False          # accept a non-loopback http:// base_url

millionsend.Emails.send(payload, {"idempotency_key": "order-42"})
millionsend.Emails.send(payload, idempotency_key="order-42")

millionsend.Batch.send(payloads, {"idempotency_key": "batch-1", "batch_validation": "permissive"})
millionsend.Batch.send(payloads, batch_validation="permissive")

from millionsend import NotFoundError, MillionSendError

try:
    contact = millionsend.Contacts.get(email="[email protected]")
except NotFoundError:
    ...  # 404
except MillionSendError as e:
    print(e.code, e.status_code, e.message)

millionsend.Emails.send({
    "from": "Acme <[email protected]>",
    "to": ["[email protected]"],
    "cc": "[email protected]",
    "bcc": ["[email protected]"],
    "reply_to": "[email protected]",
    "subject": "Your receipt",
    "html": "<p>Thanks!</p>",
    "text": "Thanks!",
    "scheduled_at": "in 2 hours",                        # or ISO 8601 with offset
    "tags": [{"name": "category", "value": "receipt"}],
    "topic_id": topic.id,                                # skip recipients opted out of the topic
    "headers": {"X-Entity-Ref-ID": "order-42"},
    "attachments": [{
        "filename": "receipt.pdf",
        "content": base64_pdf,                           # base64 string
        "content_type": "application/pdf",               # optional
        "content_id": "receipt",                         # optional, for cid: references
    }],
}, idempotency_key="order-42")

millionsend.Emails.get(email_id)                              # GET /emails/{id} (includes a nullable 0-10 `score`)
millionsend.Emails.list(limit=50, after=cursor)               # GET /emails
millionsend.Emails.update({"id": email_id, "scheduled_at": "2026-09-01T09:00:00Z"})  # PATCH, scheduled only
millionsend.Emails.cancel(email_id)                           # POST /emails/{id}/cancel (scheduled only)
millionsend.Emails.remove(email_id)                           # DELETE /emails/{id}
millionsend.Emails.get_insights(email_id)                     # GET /emails/{id}/insights (404 until computed)

millionsend.Batch.send([payload_a, payload_b], batch_validation="permissive")  # up to 100; see `errors`

millionsend.Contacts.create({
    "email": "[email protected]",
    "first_name": "Ada",
    "last_name": "Lovelace",
    "unsubscribed": False,
    "properties": {"plan": "pro", "seats": 3},
    "segments": [{"id": segment.id}],
    "topics": [{"id": topic.id, "subscription": "opt_in"}],
})
millionsend.Contacts.get(email="[email protected]")  # by id or email (email wins)
millionsend.Contacts.get("contact-id")          # bare id works too, as does resend's id="contact-id"
millionsend.Contacts.update({"id": "contact-id", "unsubscribed": True, "first_name": None})  # None clears
millionsend.Contacts.update({"email": "[email protected]", "properties": {"plan": None}})       # None removes the key
millionsend.Contacts.remove(email="[email protected]")              # the contact's emails stay in the send log
millionsend.Contacts.remove(email="[email protected]", erase=True)  # ?erase=true — also scrubs the address from email history, event payloads and API logs (GDPR/LGPD erasure)
millionsend.Contacts.preferences_link(email="[email protected]").url  # hosted preference page (MillionSend extension); no expiry, hand it only to the contact
millionsend.Contacts.list(limit=50)
millionsend.Contacts.list(segment_id=segment.id)  # GET /segments/{id}/contacts
# Bulk read (MillionSend extension): attach properties and topic subscriptions to every item,
# so an audience reads in one request per 100 contacts instead of one per contact
millionsend.Contacts.list(limit=100, include=["properties", "topics"])  # ?include=properties,topics

# Bulk create (MillionSend extension) — up to 1000 per call
result = millionsend.Contacts.Batch.create(
    [{"email": "[email protected]"}, {"email": "[email protected]", "first_name": "B"}],
    on_conflict="upsert",            # error (default) | skip | upsert
    batch_validation="permissive",   # strict (default) | permissive
)
result.data[0].status  # created | updated | skipped
result.counts.failed
result.errors          # permissive mode: [{index, message}]

# Bulk lookup (MillionSend extension) — up to 1000 contacts by id or email in one request, in request order;
# unknown entries are listed under `missing`, not errors — one request against the rate limit
result = millionsend.Contacts.Batch.get(["contact-id", {"email": "[email protected]"}], include=["topics"])
result.data     # [{object: "contact", id, email, ..., topics}] — the contacts found
result.missing  # [{index, email}] — request entries that matched nobody

# Bulk delete (MillionSend extension) — {"ids": [...]} or {"emails": [...]}, up to 1000; data lists only the rows deleted.
# Emails stay in the send log; erase=True also scrubs each address from email history, event payloads and API logs
millionsend.Contacts.Batch.remove({"emails": ["[email protected]", "[email protected]"]})
millionsend.Contacts.Batch.remove({"emails": ["[email protected]"]}, erase=True)  # body {"emails": [...], "erase": true}

# Segment membership — mirrors resend's contacts.segments
millionsend.Contacts.Segments.add({"contact_id": "contact-id", "segment_id": segment.id})
millionsend.Contacts.Segments.remove({"email": "[email protected]", "segment_id": segment.id})

# Topic subscriptions (granular unsubscribe) — mirrors resend's contacts.topics
millionsend.Contacts.Topics.update({
    "email": "[email protected]",
    "topics": [{"id": "topic-id", "subscription": "opt_out"}],
})
topics = millionsend.Contacts.Topics.list(email="[email protected]")  # GET /contacts/{idOrEmail}/topics
for t in topics.data:
    print(t.name, t.subscription, t.explicit, t.visibility)  # subscription is the effective choice; explicit=False means the topic default applies; the preference page lists public topics only

prop = millionsend.ContactProperties.create({"key": "plan", "type": "string", "fallback_value": "free"})
millionsend.ContactProperties.list()
millionsend.ContactProperties.get(prop.id)
millionsend.ContactProperties.update({"id": prop.id, "fallback_value": None})  # None clears
millionsend.ContactProperties.remove(prop.id)

millionsend.Topics.create({"name": "Product updates", "default_subscription": "opt_in"})
millionsend.Topics.get(topic_id)
millionsend.Topics.list()      # bare {"data": [...]} — topics are unpaginated
millionsend.Topics.update(topic_id, {"name": "Product news", "visibility": "public"})
millionsend.Topics.remove(topic_id)

broadcast = millionsend.Broadcasts.create({
    "name": "September launch",   # internal, optional
    "segment_id": segment.id,     # optional
    "topic_id": topic.id,         # optional
    "from": "Acme <[email protected]>",
    "reply_to": "[email protected]",
    "subject": "Launch",
    "preview_text": "It's here",
    "html": "<p>Hi {{{FIRST_NAME|there}}}</p>",
    "text": "Hi there",
    "send": False,                # True sends (or schedules) instead of saving a draft
    "scheduled_at": "in 1 hour",  # with send: True
})
millionsend.Broadcasts.list()
millionsend.Broadcasts.get(broadcast.id)
millionsend.Broadcasts.update(broadcast.id, {"subject": "Launch 🚀", "topic_id": None})  # draft only; None clears
millionsend.Broadcasts.update({"broadcast_id": broadcast.id, "subject": "Launch"})        # resend-python shape
millionsend.Broadcasts.send(broadcast.id, scheduled_at="2026-09-01T09:00:00Z")  # omit to send now
millionsend.Broadcasts.send({"broadcast_id": broadcast.id})                      # resend-python shape
millionsend.Broadcasts.cancel(broadcast.id)  # scheduled only
millionsend.Broadcasts.remove(broadcast.id)  # draft only

segment = millionsend.Segments.create({
    "name": "Pro plan",
    "filter": {"match": "all", "conditions": [   # optional; omit or None = every contact
        {"field": "property:plan", "op": "equals", "value": "pro"},
    ]},
})
millionsend.Segments.get(segment.id)   # includes a live contact_count
millionsend.Segments.list()
millionsend.Segments.update(segment.id, {"name": "Pro tier"})
millionsend.Segments.remove(segment.id)

millionsend.Suppressions.add({"email": "[email protected]", "origin": "manual"})  # origin: bounce | complaint | manual | unsubscribe
millionsend.Suppressions.get("[email protected]")
millionsend.Suppressions.list(origin="bounce", limit=50)
millionsend.Suppressions.remove("[email protected]")

millionsend.Suppressions.Batch.add({"emails": ["[email protected]", "[email protected]"], "origin": "unsubscribe"})  # up to 1000
millionsend.Suppressions.Batch.remove({"emails": ["[email protected]"]})   # or {"ids": [...]}

domain = millionsend.Domains.create({
    "name": "acme.dev",
    "region": "us-east-1",           # optional; must match the deployment's SES region
    "custom_return_path": "send",    # optional
    "open_tracking": True,           # optional
    "click_tracking": True,          # optional
    "tracking_subdomain": "links",   # optional; links.acme.dev
})
for record in domain.records:        # DNS records to publish
    print(record.type, record.name, record.value)

millionsend.Domains.list()
millionsend.Domains.get(domain.id)
millionsend.Domains.verify(domain.id)
millionsend.Domains.update({"id": domain.id, "open_tracking": False, "tracking_subdomain": None})
millionsend.Domains.remove(domain.id)

webhook = millionsend.Webhooks.create({
    "endpoint": "https://acme.dev/hooks/millionsend",
    "events": ["email.delivered", "email.bounced", "email.complained"],
    "signing_secret": "whsec_...",   # optional: reuse a secret instead of minting one
})
webhook.signing_secret               # also returned by get()

millionsend.Webhooks.list()
millionsend.Webhooks.get(webhook.id)
millionsend.Webhooks.update({"webhook_id": webhook.id, "status": "disabled"})  # endpoint, events, status
millionsend.Webhooks.remove(webhook.id)

# Rotate the signing secret (MillionSend extension). For overlap_hours (default 24, max 72) deliveries
# carry both signatures, so the receiver can switch without a gap; 0 drops the old secret at once.
rotated = millionsend.Webhooks.rotate(webhook.id, {"overlap_hours": 24})  # or {"signing_secret": "whsec_..."}
rotated.signing_secret
rotated.previous_secret_expires_at   # None once the old secret stops signing; also on get()

key = millionsend.ApiKeys.create({"name": "ci", "permission": "sending_access", "domain_id": domain.id})
key.token                            # shown once
millionsend.ApiKeys.list()
millionsend.ApiKeys.remove(key.id)

template = millionsend.Templates.create({
    "name": "Welcome",
    "alias": "welcome-v1",           # optional, unique per team
    "subject": "Welcome aboard",     # optional
    "html": "<p>Hi {{{FIRST_NAME}}}</p>",
    "text": "Hi",                    # optional
})
millionsend.Templates.get("welcome-v1")
millionsend.Templates.list()
millionsend.Templates.update({"id": "welcome-v1", "subject": None, "alias": None})  # None clears
millionsend.Templates.duplicate(template.id)
millionsend.Templates.publish(template.id)   # templates are always published; kept for resend compatibility
millionsend.Templates.remove(template.id)

usage = millionsend.Usage.get()
usage.plan                     # None when self-hosted
usage.limits.emails_per_day    # None = unlimited
usage.today.emails_sent

insights = millionsend.Emails.get_insights(email.id)  # raises NotFoundError until computed
print(insights.score, insights.band)                  # 8.5 "excellent"
for check in insights.checks:
    print(check.id, check.status, check.penalty)

account = millionsend.Deliverability.get()            # trailing-30-day account score
print(account.score, account.band, account.guardrail_status)  # scores are None until enough data