Database
First reply
Send an automatic acknowledgement in the same Gmail thread, with guards that stop the bot from answering itself, auto-replies, or bulk mail.
By the end of this page every customer email gets an automatic "we received your message" in the same thread. This is the first full round trip, before any AI is involved.
An auto-reply without guards is dangerous. If the bot answers its own replies, or another mailbox's out-of-office message, two systems email each other until someone notices. The guards run first, in plain Python.
Write the guards
from email.utils import parseaddr
def skip_reason(message: dict, headers: dict[str, str], own_address: str) -> str | None:
"""Return why the bot must not answer this message, or None when it may."""
if "SENT" in message.get("labelIds", []):
return "sent by us"
if parseaddr(headers.get("from", ""))[1].lower() == own_address.lower():
return "from our own address"
auto_submitted = headers.get("auto-submitted", "no").strip().lower()
if auto_submitted != "no":
return f"auto-submitted: {auto_submitted}"
if headers.get("precedence", "").strip().lower() in {"bulk", "list", "junk"}:
return "bulk or list mail"
return NoneOut-of-office replies and mailer daemons set Auto-Submitted (RFC 3834). Newsletters and mailing lists often set Precedence. The duplicate check already lives in db.start_event.
Send the acknowledgement
Replace app/pipeline.py. handle_message is unchanged. process now runs the guards and sends the reply.
import logging
from email.utils import parseaddr
from app import db
from app.gmail import PROCESSED_LABEL, Gmail, get_body, get_headers
from app.guards import skip_reason
logger = logging.getLogger(__name__)
MAX_ATTEMPTS = 3
ACKNOWLEDGEMENT = """Hi,
Thanks for your email. We received your message and will get back to you soon.
Harbor & Pine support"""
def handle_message(gmail: Gmail, message_id: str) -> None:
event = db.start_event(message_id)
if event is None:
gmail.add_label(message_id, PROCESSED_LABEL)
return
try:
status, result = process(gmail, message_id)
gmail.add_label(message_id, PROCESSED_LABEL)
db.finish_event(event["id"], status, result)
logger.info("%s %s %s", message_id, status, result)
except Exception as error:
db.fail_event(event["id"], str(error))
if event["attempts"] >= MAX_ATTEMPTS:
gmail.add_label(message_id, PROCESSED_LABEL)
raise
def process(gmail: Gmail, message_id: str) -> tuple[str, dict]:
message = gmail.get_message(message_id)
headers = get_headers(message)
reason = skip_reason(message, headers, gmail.address)
if reason:
return "skipped", {"reason": reason}
sender = parseaddr(headers.get("from", ""))[1]
ticket_id = db.upsert_ticket(message["threadId"], sender, headers.get("subject", ""))
db.save_message(ticket_id, message_id, "in", sender, get_body(message))
sent = gmail.send_reply(message, ACKNOWLEDGEMENT)
db.save_message(ticket_id, sent["id"], "out", gmail.address, ACKNOWLEDGEMENT)
db.update_ticket(ticket_id, "answered")
return "processed", {"action": "acknowledged"}The reply is saved as an outgoing message on the same ticket, so the dashboard later shows the whole conversation.
Check
Start the poller, then run three tests.
| Test | Expected |
|---|---|
| Email the demo address from your personal account | One acknowledgement in the same thread, event processed |
| Reply to the acknowledgement | A second acknowledgement in the same thread |
| From the demo account, email the demo address itself | No reply, event skipped with reason sent by us |
INFO app.pipeline 19a2f... processed {'action': 'acknowledged'}
INFO app.pipeline 19a30... skipped {'reason': 'sent by us'}Your own acknowledgement never comes back as new mail, because Gmail files sent mail under SENT, not in the inbox. The guard covers the case where it does.