An agent that works over email accumulates a record of everything it has been told: vendor replies, confirmations, a customer's answer to a question from last week. That record already has structure (sender, subject, thread, timestamp), which makes it a reasonable RAG source. This post loads a Lumbox inbox into a LlamaIndex VectorStoreIndex, keeps the index current as mail arrives, and queries it in a way that points back to the exact email an answer came from.

The examples use Python, llama-index-core (0.14 at the time of writing) with its default in-memory vector store, and httpx against the Lumbox REST API. Embeddings and the LLM use whatever you have configured in LlamaIndex's Settings.

One email, one Document

import os
import httpx
from llama_index.core import Document, VectorStoreIndex

INBOX_ID = os.environ["LUMBOX_INBOX_ID"]
http = httpx.Client(
    base_url="https://api.lumbox.co/v1",
    headers={"X-API-Key": os.environ["LUMBOX_API_KEY"]},
    timeout=30,
)

FENCE = "=" * 40

def unfence(text: str | None) -> str:
    if not text:
        return ""
    lines = text.split("\n")
    if len(lines) >= 3 and lines[0] == FENCE and lines[-1] == FENCE:
        lines = lines[2:-1]
    return "\n".join(lines)

def to_document(email: dict) -> Document:
    body = unfence(email.get("stripped_text") or email.get("text_body"))
    sender = email["from"]
    return Document(
        id_=email["id"],
        text=f"Subject: {email.get('subject') or ''}\n\n{body}",
        metadata={
            "email_id": email["id"],
            "thread_id": email.get("thread_id") or "",
            "from": sender.get("address") or "",
            "from_name": sender.get("name") or "",
            "category": email["parsed"]["category"],
            "received_at": email["received_at"],
        },
        excluded_embed_metadata_keys=["email_id", "thread_id", "received_at"],
        excluded_llm_metadata_keys=["email_id", "thread_id"],
    )

def load_recent() -> list[Document]:
    r = http.get(f"/inboxes/{INBOX_ID}/emails", params={"limit": 100})
    r.raise_for_status()
    return [to_document(e) for e in r.json()["data"]]

index = VectorStoreIndex.from_documents(load_recent())

Each choice in to_document is there for retrieval quality.

The Lumbox email ID is the document ID. Setting id_ means the same email always maps to the same document, which is what makes the incremental updates below safe to repeat.

stripped_text over text_body. In a long thread, every reply quotes the previous ones. Embedding text_body stores the same paragraphs many times, and retrieval returns five copies of one message. stripped_text drops lines starting with > and stops at a signature delimiter or an "On ... wrote:" line. The cost is that it also stops at a line of dashes, so a forwarded message's content is cut; if your inbox receives forwards, fall back to text_body for those.

The fence is removed for embedding. Lumbox wraps both text fields in a line of = characters and an untrusted-content label, so an LLM reading a raw API response knows it is data. That label is identical in every email and adds nothing to an embedding, so unfence strips it. The content is still untrusted; the query section below handles that.

Metadata is flat and partly hidden. Vector stores generally want string or number values, so the from object is split into two strings. IDs and timestamps stay out of the embedding text (they would only add noise to similarity), and IDs stay out of what the LLM sees, while remaining available for filters and citations.

The list endpoint returns newest first, up to 100 emails per call, with spam excluded unless you pass include_spam=true. That makes load_recent a starting point, not an archive import. For an inbox your agent owns from day one, the better plan is to index each email as it arrives.

Keeping the index current

Create a webhook with POST /v1/webhooks (default event email.received) and keep the whsec_... secret from the response, which is shown once. The webhook payload carries the email ID, sender, subject, category and extracted codes, but not the body, so the handler fetches the full email before indexing it.

import hashlib, hmac, json

def handle_webhook(raw: bytes, signature: str) -> None:
    expected = hmac.new(
        os.environ["LUMBOX_WEBHOOK_SECRET"].encode(), raw, hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(expected, signature):
        raise PermissionError("bad signature")

    payload = json.loads(raw)
    data = payload["data"]
    if payload["event"] != "email.received" or data["inbox_id"] != INBOX_ID or data["is_spam"]:
        return

    email_id = data["email_id"]
    risk = http.get(f"/emails/{email_id}/safety").json()["injection"]["risk"]
    if risk == "high":
        return

    r = http.get(f"/emails/{email_id}")
    r.raise_for_status()
    index.refresh_ref_docs([to_document(r.json())])

Call it from your web framework's POST handler with the raw request body and the X-Lumbox-Signature header. The signature is HMAC-SHA256 over the exact bytes Lumbox sent, keyed with the full secret including its whsec_ prefix, so verify before parsing. Webhooks are per org, which is why the handler checks inbox_id.

refresh_ref_docs inserts documents it has not seen and skips ones whose content hash has not changed, so a duplicate delivery is a no-op. That matters for the other half of this setup: webhook deliveries for received mail are not retried today, so an endpoint that is down when an email arrives misses it. A small job that runs every few minutes, lists emails with since set to the last run time, and passes them through the same refresh_ref_docs call closes that gap without double-indexing.

The safety check is optional but cheap. GET /v1/emails/:id/safety returns an injection score built from override phrases, hidden text, zero-width characters and similar patterns, with risk set to high at 60 out of 100. Skipping those emails keeps the most obvious attempts out of the index. It is a filter on known patterns, not a guarantee.

Querying with filters and citations

from llama_index.core.vector_stores import ExactMatchFilter, MetadataFilters

engine = index.as_query_engine(
    similarity_top_k=5,
    filters=MetadataFilters(filters=[ExactMatchFilter(key="from", value="pm@acme.com")]),
)
resp = engine.query("What did Acme say about the Q3 timeline?")
print(resp)
for n in resp.source_nodes:
    print(n.node.metadata["email_id"], n.node.metadata["received_at"], n.score)

Filters narrow the search before similarity ranking, which is the difference between "what did anyone say about Q3" and "what did this sender say". A thread_id filter does the same for one conversation. The email_id on each source node is the useful part for an agent: it can quote the source, and it can answer in the same thread with POST /v1/inboxes/:id/reply, which takes that email_id and sets the threading headers.

http.post(
    f"/inboxes/{INBOX_ID}/reply",
    json={"email_id": email_id, "text": str(resp)},
    headers={"Idempotency-Key": f"answer-{email_id}"},
)

Because retrieved text came from strangers, keep the query step read-only. Let the LLM draft an answer, and put any action (sending, clicking a link, calling another tool) behind code or a person that checks it, so a planted instruction in an email cannot trigger a send on its own.

What this setup does not cover

Attachments are only partly searchable. Lumbox extracts text from text/* attachments (in the single-email response as attachments[].extracted_text), but not from PDFs or office files; for those, download the file from the attachment's download_url and run it through a LlamaIndex reader yourself. Exact lookups such as an invoice number are often better served by Lumbox's keyword search, GET /v1/emails?q=INV-2291&inbox_id=..., which matches subject and body text directly. And the in-memory store here disappears with the process: persist the StorageContext or use a hosted vector store once this runs for real.

The webhooks docs list every event and field, the emails docs cover the email object, and email as agent memory looks at the same idea without a vector index.