Email Intake Configuration for CMMS Work Order Routing

Email intake configuration is the capture stage of the Work Order Ingestion & Parsing Pipelines domain — the component that connects to a maintenance mailbox, decodes each inbound message, and emits a single canonical work order payload for everything downstream to consume.

Most facilities still receive the majority of their service requests as free-form email: a building manager forwarding a tenant complaint, a vendor attaching an inspection report, an alarm system blasting a fault notification. Left unstructured, that traffic forces manual triage, drops data on the floor, and lets service-level windows slip before anyone has even logged a ticket. This component closes that gap. It runs as a stateless polling consumer that isolates transport mechanics — IMAP sessions, message flagging, MIME decoding, attachment extraction — from the parsing and routing logic that follows, so a flaky mail server can never corrupt the work order state machine. This guide builds that stage end to end: prerequisites, the input/output data contract, a step-by-step Python implementation, a configuration reference, validation checks, and the failure modes you will actually encounter against a live mailbox.

Scope and Prerequisites

This component is responsible for exactly one thing: converting raw RFC 822 messages into validated WorkOrderPayload objects and handing them off. It does not score priority, group work by trade, or call the CMMS dispatch endpoint — that is the job of async batch processing further down the line. Keeping the boundary that tight is what lets you redeploy the mail poller without touching dispatch, and vice versa.

Before you deploy the intake service, confirm the following are in place.

• Python 3.11+ using only the standard library for transport and parsing — imaplib for the IMAP4rev1 session, email with the modern email.policy.default for MIME decoding, ssl for TLS negotiation. Add pydantic>=2.6 if you want the envelope validated at the boundary (recommended) and redis>=5.0 for the seen-message cache that enforces exactly-once capture.
• A dedicated maintenance mailbox with IMAP4rev1 access enabled and TLS 1.2 or higher enforced on the server side. Never point the poller at a human’s shared inbox; capture semantics depend on the service owning the read/flag state of every message.
• A scoped service credential. The account must have read and flag (or move) permission on the target folder and nothing more — no send, no admin. For modern providers, use OAuth2 with a refreshable token rather than basic auth, which most large mail platforms have now disabled. Store the secret in a manager (Vault, AWS Secrets Manager) and rotate on a 90-day cycle.
• CMMS REST API v1 is not called directly here, but the field vocabulary you map onto must match it. Align your mapping with work order schema standards so the payload this stage emits validates downstream without rework.
• Environment variables: MAINT_EMAIL_HOST, MAINT_EMAIL_USER, MAINT_EMAIL_PASS (or MAINT_OAUTH_TOKEN), MAINT_MAILBOX_FOLDER (default INBOX), POLL_INTERVAL_SECONDS (default 120), ATTACHMENT_STAGING_DIR, and BROKER_URL for the handoff queue.

Architecture and Data Contract

The intake layer sits at the very front of the pipeline and consumes from the network, not from another stage. Its input is a stream of raw message bytes pulled from the mailbox over IMAP; its output is a stream of canonical WorkOrderPayload objects published to the broker. Three boundaries keep the stage honest and stop mail-server volatility from leaking into the work order layer.

• Transport boundary: the poller owns the IMAP session, TLS negotiation, connection health checks, and message flagging. A dropped connection or an auth refresh is retried here and never surfaces as a malformed work order.
• Normalization boundary: raw bytes are decoded into headers, a clean text body, and a list of staged attachments, then mapped to the canonical schema. Anything that cannot be mapped to the mandatory fields is rejected at this boundary, not passed along half-formed.
• Handoff boundary: the validated payload is published to the routing queue under at-least-once delivery with a stable message_id derived from the email Message-ID, so a redelivered message can be deduplicated rather than double-captured. The intake scope terminates the moment the broker accepts the payload.

The contract across the normalization boundary is explicit. The input is opaque bytes (one fetched RFC 822 message); the output is a WorkOrderPayload whose mandatory fields — work_order_id, asset_id, and the SLA fields — are populated or the message is dead-lettered. Encoding the output as a typed model means an empty body, a missing asset tag, or an unparseable date is caught at capture time instead of failing three stages later in dispatch.

Step-by-Step Implementation

1. Define the canonical work order payload

Intake emits the same WorkOrderPayload used everywhere on this site, imported rather than redefined so the SLA fields stay identical across every stage. The SLA fields — priority, requested_completion, and escalation_tier — are populated here from the email’s signals (subject keywords, sender domain, headers) and are what the routing stage later reads to decide urgency.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Dict, List, Optional


class Priority(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"


@dataclass
class WorkOrderPayload:
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    priority: Priority = Priority.STANDARD
    requested_completion: Optional[datetime] = None
    escalation_tier: int = 0
    status: str = "open"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

2. Open a secure, health-checked IMAP session

Initialize the client with an explicit TLS context and verify the connection is live before each polling cycle. Wrapping the session in a small class keeps credential handling, folder selection, and logout in one place and makes the poller easy to test against a mock server.

import imaplib
import os
import ssl
from typing import Optional


class IMAPIntakeSession:
    def __init__(self, host: str, port: int = 993) -> None:
        self.host = host
        self.port = port
        self._context = ssl.create_default_context()
        self._conn: Optional[imaplib.IMAP4_SSL] = None

    def connect(self) -> None:
        """Open a TLS session and select the maintenance folder read-write."""
        self._conn = imaplib.IMAP4_SSL(self.host, self.port, ssl_context=self._context)
        self._conn.login(os.environ["MAINT_EMAIL_USER"], os.environ["MAINT_EMAIL_PASS"])
        # Read-write so the poller can flag processed messages for exactly-once capture.
        self._conn.select(os.environ.get("MAINT_MAILBOX_FOLDER", "INBOX"), readonly=False)

    def is_healthy(self) -> bool:
        """NOOP is the cheapest way to confirm the session is still authenticated."""
        if self._conn is None:
            return False
        try:
            status, _ = self._conn.noop()
            return status == "OK"
        except (imaplib.IMAP4.abort, OSError):
            return False

    def close(self) -> None:
        if self._conn is not None:
            try:
                self._conn.close()
                self._conn.logout()
            finally:
                self._conn = None

3. Poll and select only unprocessed maintenance mail

Drive the session from a scheduler that wakes on a fixed interval — 60 to 300 seconds suits most maintenance queues. Use IMAP search criteria to fetch only what you have not handled, and reconnect on a stale session rather than letting a half-dead socket silently skip a cycle. The detailed scheduling, backoff, and deduplication strategy for high-volume mailboxes is covered in configuring IMAP polling for maintenance email queues.

from typing import Iterator, Tuple


def fetch_unseen(session: IMAPIntakeSession) -> Iterator[Tuple[bytes, bytes]]:
    """Yield (uid, raw_bytes) for every UNSEEN message; reconnects if stale."""
    if not session.is_healthy():
        session.connect()
    conn = session._conn

    # UID SEARCH so identifiers stay stable even if the mailbox is re-indexed.
    status, data = conn.uid("search", None, "UNSEEN")
    if status != "OK" or not data or not data[0]:
        return

    for uid in data[0].split():
        status, msg_data = conn.uid("fetch", uid, "(RFC822)")
        if status == "OK" and msg_data and msg_data[0]:
            yield uid, msg_data[0][1]

4. Normalize the message into the canonical payload

Decode the MIME tree, prefer the text/plain alternative, fall back to stripped text/html, and map the header and body signals onto the canonical fields. Subject keywords and sender domain drive an initial priority; ambiguous free-text bodies are routed to NLP intent classification downstream rather than guessed at here.

import email
import re
from email.message import EmailMessage
from email.policy import default
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone


_HTML_TAG = re.compile(r"<[^>]+>")
_ASSET_TAG = re.compile(r"\b([A-Z]{3}-\d{2,6})\b")  # e.g. HVAC-12, PMP-204
_URGENT = ("leak", "flood", "outage", "failure", "down", "urgent", "no heat")


def _best_body(msg: EmailMessage) -> str:
    """Prefer text/plain; fall back to HTML with tags stripped."""
    if msg.is_multipart():
        plain = msg.get_body(preferencelist=("plain",))
        if plain is not None:
            return plain.get_content().strip()
        html = msg.get_body(preferencelist=("html",))
        if html is not None:
            return _HTML_TAG.sub("", html.get_content()).strip()
        return ""
    return msg.get_content().strip()


def normalize(raw_bytes: bytes, attachments: list[str]) -> WorkOrderPayload:
    """Map one raw RFC 822 message onto the canonical work order schema."""
    msg = email.message_from_bytes(raw_bytes, policy=default)
    subject = (msg["Subject"] or "").strip()
    body = _best_body(msg)
    haystack = f"{subject}\n{body}".lower()

    asset_match = _ASSET_TAG.search(f"{subject} {body}")
    if asset_match is None:
        raise ValueError("no asset tag found in subject or body")

    priority = Priority.HIGH if any(k in haystack for k in _URGENT) else Priority.STANDARD

    received = msg["Date"]
    created_at = parsedate_to_datetime(received) if received else datetime.now(timezone.utc)

    return WorkOrderPayload(
        work_order_id=msg["Message-ID"].strip("<>"),
        asset_id=asset_match.group(1),
        part_skus=[],
        required_quantities={},
        priority=priority,
        requested_completion=None,
        escalation_tier=1 if priority is Priority.HIGH else 0,
        status="open",
        created_at=created_at,
    )

5. Stage attachments out of the message path

Pull binaries — inspection PDFs, fault photos, meter-reading spreadsheets — out of the MIME tree and write them to staging storage with deterministic names, leaving only references on the payload. Documents that need their contents extracted are handed to PDF parsing with Python as a separate stage; intake only stages and references them. Enforce a size ceiling so a runaway attachment cannot exhaust the worker’s disk.

import uuid
from email.message import EmailMessage
from pathlib import Path

MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024  # 25 MB hard ceiling per file


def stage_attachments(msg: EmailMessage, staging_dir: Path) -> list[str]:
    """Write each attachment to staging storage and return its path references."""
    staging_dir.mkdir(parents=True, exist_ok=True)
    paths: list[str] = []

    for part in msg.iter_attachments():
        filename = part.get_filename()
        if not filename:
            continue
        payload = part.get_payload(decode=True) or b""
        if len(payload) > MAX_ATTACHMENT_BYTES:
            continue  # Oversized — skip rather than risk filling the volume.

        safe_name = f"{uuid.uuid4().hex}_{Path(filename).name}"
        target = staging_dir / safe_name
        target.write_bytes(payload)
        paths.append(str(target))

    return paths

6. Validate, hand off, and flag the message as processed

Publish the validated payload to the routing queue with the email Message-ID as the broker message_id, then flag the source message \Seen (or move it to an archive folder) so the next cycle never re-captures it. Order matters: publish first, flag second. If the worker crashes between the two, the message stays unflagged and is retried — at-least-once capture — and the downstream deduplication on message_id absorbs the rare double-publish.

import json
import logging

import pika

logger = logging.getLogger(__name__)


def handoff(payload: WorkOrderPayload, uid: bytes, session: IMAPIntakeSession,
            broker_url: str, queue: str = "wo.intake") -> None:
    """Publish the payload durably, then flag the source message processed."""
    connection = pika.BlockingConnection(pika.URLParameters(broker_url))
    try:
        channel = connection.channel()
        channel.queue_declare(queue=queue, durable=True)
        body = json.dumps({**payload.__dict__, "priority": payload.priority.value},
                          default=str)
        channel.basic_publish(
            exchange="",
            routing_key=queue,
            body=body,
            properties=pika.BasicProperties(
                delivery_mode=2,                 # Persistent across broker restart.
                content_type="application/json",
                message_id=payload.work_order_id,  # Stable dedup key downstream.
            ),
        )
    finally:
        connection.close()

    # Only after a successful publish do we mark the email consumed.
    session._conn.uid("store", uid, "+FLAGS", "(\\Seen)")
    logger.info("captured work order %s for asset %s", payload.work_order_id, payload.asset_id)

Configuration Reference

Keep every tunable in a version-controlled configuration registry, not hard-coded in the poller. The defaults below are conservative starting points for a single multi-building facility.

Validation and Testing

The highest-value test asserts that a representative raw email maps to exactly the canonical fields you expect, with no live mailbox required — feed bytes straight into normalize. This catches a broken regex, a changed sender format, or a timezone bug at the boundary instead of in dispatch.

from datetime import datetime


def test_urgent_email_maps_to_canonical_payload():
    raw = (
        b"Message-ID: <[email protected]>\r\n"
        b"From: [email protected]\r\n"
        b"Subject: URGENT water leak at pump PMP-204\r\n"
        b"Date: Mon, 28 Jun 2026 09:15:00 +0000\r\n"
        b"Content-Type: text/plain; charset=utf-8\r\n\r\n"
        b"There is water pooling under pump PMP-204, please send someone.\r\n"
    )
    payload = normalize(raw, attachments=[])

    assert payload.work_order_id == "[email protected]"
    assert payload.asset_id == "PMP-204"
    assert payload.priority is Priority.HIGH      # 'urgent'/'leak' triggered escalation
    assert payload.escalation_tier == 1
    assert isinstance(payload.created_at, datetime)

On a healthy production run, each captured message emits a single structured log line — captured work order abc-123@... for asset PMP-204 — which is the canonical signal that intake handed a payload to the broker. A message that cannot be mapped logs a ValueError and lands in the dead-letter queue rather than blocking the cycle. Assert against both the log line and a non-empty broker queue in integration tests, and confirm the source message is flagged \Seen only after a successful publish.

Failure Modes and Troubleshooting

Expand each scenario for the root cause, the diagnostic signal, and the fix. The checklist items render as interactive checkboxes — work through them in order.

The same email is captured twice

• Confirm the poller publishes before it flags \Seen; flagging first means a crash mid-publish loses the message, while never flagging means it re-captures every cycle.
• Verify the broker message_id is set to the email Message-ID so downstream deduplication can recognize the redelivery.
• Check that select() is called read-write — a session opened readonly=True silently ignores the STORE flag write, so every message stays UNSEEN forever.

Messages arrive but bodies come through empty or full of HTML tags

• Confirm you parse with policy=default; the legacy compat32 policy returns raw payloads and breaks get_body() preference selection.
• Check the text/plain preference path runs before the HTML fallback — some senders ship an empty plain part and the real content only in HTML.
• Verify the charset is honored by get_content() rather than decoding bytes manually, which mangles non-ASCII technician notes.

Authentication fails after working for weeks

• Look for a basic-auth deprecation on the mail provider; migrate to an OAuth2 token with refresh rather than a static password.
• Confirm the secret in the manager was rotated and the running service was restarted to pick up the new value — a stale in-memory credential outlives a rotation.
• Check the service account still has folder read/flag scope; an over-tightened policy revoking STORE looks like an auth failure only when flagging.

The dead-letter queue is filling with valid-looking requests

• Inspect a dead-lettered body — if real asset tags are present but unmatched, the _ASSET_TAG regex does not cover your site’s naming scheme; widen it against your asset hierarchy design.
• Check whether requests legitimately lack an asset tag and should be routed to triage with an asset-resolution step rather than rejected outright.
• Confirm the mandatory-field set matches the current CMMS schema; a new required field upstream dead-letters everything until the mapping is updated.

Frequently Asked Questions

Should the poller use IMAP IDLE instead of interval polling?

For most maintenance mailboxes, interval polling on a 60–300 second cycle is simpler to operate and entirely adequate — work orders rarely need sub-minute capture. IMAP IDLE gives near-real-time delivery but holds a long-lived connection that you must babysit with keepalives and reconnect logic, and many hosted providers drop idle sessions aggressively. Start with polling and move to IDLE only if a measured latency requirement justifies the added connection management.

How do I guarantee each email becomes exactly one work order?

Capture is at-least-once by design: the poller publishes the payload, then flags the source message, so a crash between the two retries the message rather than losing it. Exactly-once effect comes from setting the broker message_id to the email Message-ID and deduplicating on it downstream, exactly as the routing stage does for its envelopes. Together they ensure a redelivered email never spawns a second work order.

What belongs in this stage versus the parsing stages after it?

This stage owns transport and normalization only — connecting, decoding MIME, stripping HTML, staging attachments, and mapping to the canonical schema. Interpreting ambiguous free text belongs to NLP intent classification, extracting figures from attached documents belongs to PDF parsing, and scoring or grouping the work belongs to async batch processing. Keeping intake thin means a mail-server change never ripples into the parsing logic.

Where does part and inventory data come from if the email rarely lists SKUs?

Intake deliberately leaves part_skus and required_quantities empty when the email does not state them — guessing would corrupt the payload. The asset tag it captures is the join key a later enrichment step uses to resolve required parts through parts availability checks, so the work order arrives at dispatch with its parts already confirmed rather than fabricated at intake.

Related

Feed captured payloads into async batch processing, disambiguate free-text requests with NLP intent classification, extract staged documents through PDF parsing with Python, tune the scheduler in configuring IMAP polling for maintenance email queues, and align the field mapping with work order schema standards.

Part of: Work Order Ingestion & Parsing Pipelines.