Regex-based email parsing works until it doesn't. The moment your emails come from real people instead of a fixed template, pattern matching starts missing fields, mangling values, and breaking every time someone phrases something slightly differently. This is where an LLM-based parser earns its keep: it reads an email the way a person would and returns exactly the structured fields you asked for.
This is a practical, code-first walkthrough for building one yourself.
What we are building
A pipeline that:
- Connects to an inbox and pulls new messages
- Sends each email to an LLM with a defined schema
- Gets back clean, structured JSON
- Validates and normalizes the output
- Writes the result to a database (or CRM, or Airtable)
We will use Python throughout, but the same architecture works in Node.js, Go, or any language with an HTTP client.
Step 1: Connect to the inbox
For Gmail, the cleanest option is the Gmail API rather than raw IMAP, since it supports push notifications and label management natively.
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
def get_gmail_service(token_path="token.json"):
creds = Credentials.from_authorized_user_file(token_path)
return build("gmail", "v1", credentials=creds)
def fetch_unread_messages(service, query="is:unread label:leads"):
results = service.users().messages().list(
userId="me", q=query
).execute()
return results.get("messages", [])
Using a query like label:leads or a dedicated forwarding address (e.g. parse@yourcompany.com) keeps the parser focused on relevant mail instead of processing an entire personal inbox.
Step 2: Extract the raw email content
import base64
def get_message_body(service, message_id):
msg = service.users().messages().get(
userId="me", id=message_id, format="full"
).execute()
payload = msg["payload"]
parts = payload.get("parts", [payload])
body = ""
for part in parts:
if part.get("mimeType") == "text/plain":
data = part["body"].get("data", "")
body += base64.urlsafe_b64decode(data).decode("utf-8", errors="ignore")
return body.strip()
Strip quoted replies and forwarded thread history before this text goes to the model. A simple heuristic (cut everything after the first line starting with On ... wrote: or -----Original Message-----) removes most of the noise.
Step 3: Define your extraction schema
Be specific. The more precisely you describe each field, the more consistent the model's output becomes.
SCHEMA = {
"name": "Full name of the person, or null",
"email": "Email address mentioned in the body, or null",
"phone": "Phone number in E.164 format if possible, or null",
"request_type": "One of: quote_request, support, complaint, follow_up, other",
"urgency": "One of: low, medium, high",
"budget": "Numeric value only, no currency symbol, or null",
"summary": "One sentence summary of what they want",
}
Step 4: Call the model
import json
from openai import OpenAI
client = OpenAI()
def extract_structured_data(email_text: str, schema: dict) -> dict:
system_prompt = f"""
You are a precise data extraction engine. Extract fields from the email
below according to this schema:
{json.dumps(schema, indent=2)}
Rules:
- Only use information explicitly present in the email.
- Never guess or infer a value that is not stated.
- Return valid JSON with exactly these keys, nothing else.
"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": email_text},
],
response_format={"type": "json_object"},
temperature=0,
)
return json.loads(response.choices[0].message.content)
A few things matter here beyond the obvious:
temperature=0keeps output consistent between runs on similar input.- The "never guess" rule is critical. Without it, models will happily fabricate a plausible-looking phone number or date rather than returning null.
response_format={"type": "json_object"}forces valid JSON instead of a model wrapping its answer in explanatory text.
Step 5: Validate and normalize before storing anything
Model output should never go straight into your database. Add a validation layer:
import re
from datetime import datetime
def normalize_phone(phone: str | None) -> str | None:
if not phone:
return None
digits = re.sub(r"[^\d+]", "", phone)
return digits if len(digits) >= 7 else None
def validate_record(record: dict) -> tuple[dict, list[str]]:
warnings = []
record["phone"] = normalize_phone(record.get("phone"))
if record.get("request_type") not in {
"quote_request", "support", "complaint", "follow_up", "other"
}:
warnings.append("unrecognized request_type, defaulting to 'other'")
record["request_type"] = "other"
if not record.get("name") and not record.get("email"):
warnings.append("no identifying contact info found")
record["processed_at"] = datetime.utcnow().isoformat()
return record, warnings
Anything that generates a warning should route to a review queue instead of being written silently. This one habit prevents the slow, quiet data corruption that kills trust in an automated pipeline faster than anything else.
Step 6: Deduplicate before writing
Without deduplication, the same person emailing twice creates two separate records. Match on a stable identifier, not a name:
def find_existing_record(db, email: str | None, phone: str | None):
if email:
existing = db.find_by_email(email)
if existing:
return existing
if phone:
return db.find_by_phone(phone)
return None
If a match is found, update the existing record instead of inserting a new one.
Step 7: Write to your destination
The destination changes the specifics, but the pattern stays the same: take the validated record and call the appropriate API.
def write_to_airtable(record: dict, table):
table.create({
"Name": record.get("name"),
"Email": record.get("email"),
"Phone": record.get("phone"),
"Request Type": record.get("request_type"),
"Urgency": record.get("urgency"),
"Budget": record.get("budget"),
"Summary": record.get("summary"),
})
Swap this function for a HubSpot, Salesforce, or Postgres write and the rest of the pipeline does not need to change.
Step 8: Handle attachments
If the email includes attachments worth extracting data from (a signed PDF, an invoice, a scanned form), download them, run OCR or feed them directly to a multimodal model, and merge the result into the same record before writing it.
def extract_pdf_attachment(file_bytes: bytes, schema: dict) -> dict:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": f"Extract fields matching schema: {schema}"},
{"role": "user", "content": [
{"type": "text", "text": "Extract the data from this document."},
{"type": "image_url", "image_url": {"url": f"data:application/pdf;base64,{base64.b64encode(file_bytes).decode()}"}},
]},
],
response_format={"type": "json_object"},
)
return json.loads(response.choices[0].message.content)
Step 9: Close the loop
Once the record is written, mark the email as processed so it does not get parsed twice:
def mark_processed(service, message_id, label_id):
service.users().messages().modify(
userId="me", id=message_id,
body={"addLabelIds": [label_id], "removeLabelIds": ["UNREAD"]}
).execute()
Testing your parser properly
Before pointing this at live email:
- Collect 30 to 50 real historical emails covering your actual variety of phrasing
- Run extraction on all of them and manually check the output field by field
- Deliberately include edge cases: missing fields, two dates in one email, sarcasm, spam
- Track a simple accuracy score per field, not just overall pass/fail
- Keep the validation warnings visible during this phase so you can tune the schema and prompt before trusting it unattended
Cost and performance notes
A lightweight model like gpt-4o-mini typically costs a fraction of a cent per email for this kind of extraction, so cost rarely becomes the bottleneck even at a few thousand emails per month. Latency is usually one to three seconds per email, which is fine for an asynchronous pipeline but worth knowing if you are building something that needs a near-instant response.
When to stop building it yourself
This architecture works well and is genuinely not hard to get running for a first version. Where teams usually get stuck is production hardening: thread and forward handling, attachment edge cases, deduplication at scale, monitoring for silent failures, and keeping the schema in sync as your business's needs change. If you would rather have this built, tested against your real inbox, and maintained, our email-to-CRM automation service does exactly that. For the broader concepts behind this build, see what an email parser is and how inbox parsing works end to end.
Frequently asked questions
Do I need a fine-tuned model to do this well? No. A general-purpose model with a clear schema and explicit instructions handles most email extraction tasks accurately without any fine-tuning.
What if the email is in a language other than English? Modern LLMs handle multilingual extraction well without extra setup. Just make sure your schema instructions are unambiguous.
Should I use a bigger, more expensive model? For most structured extraction tasks, a smaller, faster model performs nearly as well as a larger one and costs significantly less. Reserve larger models for genuinely ambiguous or high-stakes extractions.
How do I stop the model from hallucinating fields?
Explicitly instruct it to return null rather than guess, set temperature=0, and validate output before storing it. Track any suspicious patterns and refine the prompt over time.