Inbound Email Webhook Parsing with LLMs: Converting Unstructured Email to JSON

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel, Field, EmailStr
from openai import OpenAI
import os

app = FastAPI(title="Inbound Email LLM Ingestion Gateway")
client = OpenAI()

class StructuredSupportTicket(BaseModel):
    category: str = Field(description="Billing, Bug Report, Security, or General Inquiry")
    sentiment: str = Field(description="Positive, Neutral, Frustrated, or Urgent")
    customer_intent: str = Field(description="Core objective of the customer message")
    requires_human_reply: bool = Field(description="Whether automated response suffices or human triage needed")
    extracted_entities: dict = Field(default_factory=dict, description="Extracted account IDs, invoice numbers, or URLs")
    suggested_reply: str = Field(description="Drafted empathetic and accurate reply to customer")

@app.post("/webhooks/inbound")
async def handle_inbound_email_webhook(request: Request):
    payload = await request.json()
    from_email = payload.get("from")
    subject = payload.get("subject", "")
    raw_body = payload.get("text") or payload.get("html", "")

    if not from_email or not raw_body:
        raise HTTPException(status_code=400, detail="Missing required email fields")

    # Pass sanitized raw body to LLM structured parser
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "You are a specialized enterprise email parser. Extract structured facts, categorize intent, and ignore email disclaimers and signature logos.",
            },
            {
                "role": "user",
                "content": f"From: {from_email}\nSubject: {subject}\n\nBody:\n{raw_body[:4000]}",
            },
        ],
        response_format=StructuredSupportTicket,
    )

    ticket = completion.choices[0].message.parsed
    print(f"[PARSED] {from_email} -> {ticket.category} (Urgency: {ticket.sentiment})")

    # Store structured ticket in Postgres or trigger internal workflow
    return {"status": "success", "ticket": ticket.dict()}