The poller

Run a separate process that checks the inbox on an interval, handles each new message, and labels it so it is never picked up twice.

By the end of this page a second process runs next to the API. Every POLL_INTERVAL_SECONDS it lists new inbox mail, logs it, and adds the ai-processed label.

Polling or webhooks

Gmail can also push new mail to your server through a webhook. That needs Google Pub/Sub, a public HTTPS address, and a domain. Polling only makes outgoing calls, so nothing has to reach your server. For email support a delay of a minute or ten is fine. Switch to webhooks when you need instant replies or high volume.

Handle one message

For now the pipeline logs the email and labels it. The next pages add the database and the replies.

app/pipeline.py

import logging

from app.gmail import PROCESSED_LABEL, Gmail, get_headers

logger = logging.getLogger(__name__)


def handle_message(gmail: Gmail, message_id: str) -> None:
    message = gmail.get_message(message_id)
    headers = get_headers(message)
    logger.info("New email from %s: %s", headers.get("from"), headers.get("subject"))
    gmail.add_label(message_id, PROCESSED_LABEL)

Write the poller

app/poller.py

import logging
import time

from app.config import settings
from app.gmail import Gmail
from app.pipeline import handle_message

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
logger = logging.getLogger("poller")


def poll_once() -> int:
    gmail = Gmail()
    message_ids = gmail.list_new_message_ids()
    for message_id in message_ids:
        try:
            handle_message(gmail, message_id)
        except Exception:
            # One bad email must not stop the loop.
            logger.exception("Failed to process %s", message_id)
    return len(message_ids)


def main() -> None:
    logger.info("Polling every %s seconds", settings.poll_interval_seconds)
    while True:
        try:
            count = poll_once()
            if count:
                logger.info("Handled %s new messages", count)
        except Exception:
            logger.exception("Poll failed")
        time.sleep(settings.poll_interval_seconds)


if __name__ == "__main__":
    main()

poll_once does one pass, and main repeats it forever. A message that fails keeps no label, so the next pass tries it again.

Run it

Use two terminals from now on, one for the API and one for the poller.

uv run python -m app.poller

The poller runs as its own process, not inside FastAPI. With several API workers you would get several pollers, and the Gmail client is synchronous, so it would block the API while it waits.

Check

Send two emails to the demo address and watch the poller terminal.

2026-10-28 10:00:00 INFO poller Polling every 30 seconds
2026-10-28 10:00:31 INFO app.pipeline New email from Alex <alex@example.com>: Paddle board size
2026-10-28 10:00:31 INFO app.pipeline New email from Alex <alex@example.com>: Missing screws
2026-10-28 10:00:31 INFO poller Handled 2 new messages

In Gmail both emails carry the ai-processed label. On the next pass the poller finds nothing, because the search skips that label. Stop the poller with Ctrl+C.