Build a cold email workflow with Python

Build a reply-aware outreach scheduler in Python and prove its stop conditions before connecting a transport. The demo uses SQLite transactions to checkpoint recipient jobs, event IDs, and simulated attempts. Hosted MailKite is unsuitable for unsolicited outreach: its current terms prohibit unsolicited bulk/commercial mail and sending to recipients who haven’t consented. This is a runnable local state-machine simulation, not a cold-email sending integration. No emails are sent.

# Download outreach.py and test_outreach.py into an empty working directory.
python3 outreach.py --state checkpoint.sqlite
python3 outreach.py --state checkpoint.sqlite
python3 -m unittest -v test_outreach.py

Download the Python simulation and tests. Use Python 3.11 or newer with its standard-library sqlite3 module. No package install, API key, SMTP server, or network connection is needed. The CLI has no live option. A second invocation opens the same checkpoint rather than recreating recipient jobs.

Separate orchestration from transport

MailKite, which we build, supplies consent-based sending and inbound receiving. Its acceptable-use terms rule out hosted transport for this guide’s unsolicited-outreach purpose. Neither sendBatch nor a sequence changes that boundary. A subscription list is a newsletter audience; renaming it a cold-email campaign would not implement outreach.

Instantly’s API documentation describes connected accounts, campaigns, leads, OAuth mailbox connections, and webhook events. That is evidence of outreach orchestration surfaces worth evaluating, not an adapter included in this example. Check its current service terms, your mailbox provider’s policies, and the recipient basis for your intended use before building an integration. We make no claim here that any documented endpoint grants blanket permission to contact people.

Pick one owner for follow-up execution. If a vendor campaign engine schedules the mail, your application should synchronize and stop its jobs through that vendor’s actual API. If your worker owns execution, the adapter should dispatch only the individual approved attempts. Running both schedulers without a reconciliation plan makes duplicate follow-ups likely.

Model the recipient lifecycle in SQLite

The demo has five tables: contacts, jobs, events, attempts, and pacing. Contact identity is a normalized email address. A job’s primary key combines the fixed campaign name, address, and step. An event ID is unique, and an attempt key is unique. Those constraints make duplicates visible in the database rather than depending on a process-local set.

from outreach import open_state, tick, apply_event

db = open_state("checkpoint.sqlite")
try:
    first = tick(db, 0)  # simulated attempt; no delivery call
    assert tick(db, 59) is None
    tick(db, 60)
    apply_event(db, "reply-1", "ada@example.com", "reply")
finally:
    db.close()

Opening the database creates missing tables and inserts fixture rows with INSERT OR IGNORE. Existing stops and job statuses remain intact. Each fixture has a name and a specific piece of context, such as a webhook retry write-up. The generated text uses those fields. It contains no fabricated company metrics or invented prior relationship.

A real import would need syntactic validation, source provenance, and a reason the context belongs to the recipient. Don’t treat a display name as an identifier, overwrite a terminal opt-out from a later CSV, or silently enroll every known correspondent. The simulation imports fictional fixtures only; it has no prospect collection feature.

Pace with a controllable clock

tick(db, now) accepts nonnegative integer virtual seconds. Inside BEGIN IMMEDIATE, it reads the shared next slot, selects one due pending job whose contact isn’t stopped, records a simulated attempt, marks the job, and advances the global pacing slot by 60 seconds. A recipient’s follow-up becomes due 86,400 seconds after their initial attempt.

Those intervals are test settings, not a provider throughput recommendation. The scenario moves the virtual clock directly, so you can test a full day without sleeping. It doesn’t demonstrate real mailbox placement, warm-up, or an account’s allowed rate. A future scheduler would need server time, daily budgets, provider-specific limits, and health-based suspension in addition to this state machine.

The transaction keeps the local attempt record, job transition, and pacing update together. SQLite serializes writers for the same database file. That is useful for this small demo, but it isn’t a distributed queue lease across copied databases or different hosts. Keep a single authoritative state store when moving the worker into a service.

Stop follow-ups without stopping the entire campaign

apply_event accepts only reply, opt-out, or bounce and requires a known recipient and nonempty event ID. It inserts the event with a unique key, records the first stop reason, and updates only that recipient’s pending jobs to canceled. The event and cancellation commit in one transaction. Duplicate events return false and do not reopen anything.

The scenario records four initial attempts, then receives Ada’s reply, Bo’s opt-out, and Cy’s bounce. Ada’s reply is submitted again to demonstrate event dedupe. A later tick records Dev’s follow-up because Dev has no stop event. The expected fixture result is five simulated attempts, three canceled jobs, and three unique events. Reopening the same database and rerunning the scenario leaves those counts unchanged.

This stop-first design doesn’t require interpreting whether a reply is positive. A reply ends automatic follow-ups and hands the conversation to a person. Opt-outs and bounces remain terminal for the demo. If you later implement resumption, make it a separate decision with an audit trail rather than a side effect of reimporting contacts.

Feed it authenticated reply events

External feedback needs an adapter before apply_event. Instantly documents reply_received, lead_unsubscribed, and email_bounced, plus workspace, campaign, and lead identifiers. Map them to internal events only after authenticating the delivery and matching the tenant. A documented JSON event schema is not a complete authentication protocol. Don’t publish a receiver that trusts lead_email simply because the body looks plausible.

For an independently permitted MailKite inbound receiving path, use the SDK’s local verifier. Install mailkite-dev==0.20.0 only if exploring this integration sketch; the simulation itself doesn’t need it:

import json
from mailkite import verify_webhook
from outreach import apply_event

def receive_reply(db, raw_body, signature, secret, resolve_reply):
    if not verify_webhook(signature, raw_body, secret):
        raise ValueError("Unauthenticated webhook")
    event = json.loads(raw_body)
    if event.get("type") != "email.received":
        return "review"
    if event.get("auth", {}).get("dmarc") != "pass":
        return "review"
    # Your lookup checks destination inbox, tenant, thread, and expected sender.
    recipient = resolve_reply(event)
    if not recipient:
        return "review"
    return apply_event(db, event["id"], recipient, "reply")

Pass the unmodified body bytes and x-mailkite-signature header. The SDK checks the signature and a five-minute freshness window. Its signature is positional, even though the shared helper input schema uses named fields. The stable inbound message id is the dedupe key. A null email authentication verdict is not a pass.

The signature proves MailKite delivered the event; it doesn’t establish the author’s identity. resolve_reply is an application-owned dependency, not an SDK method. It must match the stored conversation and controlled inbox, check the expected sender, and reject cross-tenant or unmatched messages. Acknowledge only after the cancellation transaction commits. An HTTP endpoint, route setup, schema validation, and reply mapping are not implemented by this sketch.

Know what the tests establish

The tests cover pacing at 59 versus 60 seconds, per-recipient reply/opt-out/bounce cancellation, duplicate events, unknown mappings, and reopening a real temporary SQLite file. No test opens a mail transport. The database proves local dedupe, not exactly-once provider delivery.

A future adapter must record dispatch intent before network calls and reconcile ambiguous outcomes after timeouts. Otherwise a provider may accept an email while the worker fails before writing its result, and an automatic retry can duplicate it. Add provider message IDs, authenticated feedback, remote cancellation, and error classification before describing a transport as deployable. Ordinary custom headers are not guaranteed provider idempotency.

Sources retrieved 2026-10-14: MailKite terms, webhook verification, Python SDK, Instantly API documentation, and webhook schema. Shared inbound and verifier schemas and the Python SDK implementation were inspected. Only the local simulation and mocked boundaries are tested here.

Next: cold email hub, Node.js outreach simulation, Python newsletter, and Node.js newsletter. Share questions about the checkpoint and adapter boundary in community discussions.