Prologue
Every autonomous system worth building rests on one fundamental distinction: the difference between a reflex and a reasoned thought. Human cognition has lived with this split for as long as we've been thinking about thinking. Our fast System One machinery reads incoming sensory signals within milliseconds — recognizing patterns, assigning categories, raising the alarm when something smells wrong. Our deliberate System Two machinery steps in for the slower work: composing long-form responses, following multi-step logical deductions, and weaving complex narratives in natural language.
Modern agentic AI architectures mirror this dual structure — and the systems that ignore it pay dearly. Point a large autoregressive generative model at a simple routing decision, a classification task, or a guardrail check, and you inherit severe latency penalties, runaway computational expense, and schemas that quietly drift out from under you. The Jev model, developed by TypeSafe AI, takes the opposite stance: it operates strictly as a specialized System One decision engine. Where a generative model samples token after token to produce free-form prose, Jev measures unstructured context against predefined, strongly typed questions — discrete choices, ordinal score rankings, binary probabilities — and it runs those evaluations in parallel, returning calibrated decisions within seventy to five hundred milliseconds at a fraction of what a generative foundation model would cost.
Pair that high-speed reflex layer with an auxiliary generative language model — deployed locally or behind a remote endpoint — and you have the makings of an enterprise-grade email processing pipeline. That is exactly what we will build in this tutorial: a complete, production-ready system that categorizes, summarizes, and routes email connected directly to Google Mail. Jev handles the deterministic classification and urgency scoring, the auxiliary language model writes concise executive summaries, and the Google Mail API mirrors every state change back to the enterprise inbox in real time.
The Shape of the System: Architecture and Operational Topology
Think of the pipeline as four distinct responsibilities, each behind its own clearly drawn architectural boundary: ingestion, deterministic classification, generative synthesis, and state synchronization. Incoming correspondence arrives through the Google Mail infrastructure, where the ingestion service polls the mailbox using authenticated Google Workspace credentials, normalizes multipart message payloads, and extracts plain text content — carefully preserving the conversational context around it.
Once the raw email payload has been extracted, it crosses into the Jev reflex engine. There, the message is evaluated against a predefined taxonomy and emerges with an exact category, a granular urgency score, and a binary security-anomaly indicator. If the message also needs summarization or downstream reply drafting, the pipeline hands the context to a configurable generative language model provider — an on-premise local engine via Ollama or vLLM, or a remote enterprise service, whichever your policies demand. Finally, the synchronization dispatcher closes the loop: it applies Google Mail labels, clears unread flags, and forwards critical incidents to operational alert channels. The diagram below shows how these core constituents interact.
| Google Mailbox |
| (MIME Payloads, Unread Inbound Messages) |
|
| Google API Client / OAuth
v
| Ingestion and MIME Parser |
| (Payload Normalization, Header and Body Extraction) |
|
| Sanitized Plaintext State
v
| Jev System-1 Reflex Classifier |
| (Parallel Decision Evaluation: Choice, Score, and Noul) |
| |
| Classified Decision | Reflex Assessment
v v
| Local or Remote | | Google Mail Sync |
| Generative Engine | | and Notification |
| (Ollama or API) | | (Labels, Alerts) |
| |
+----+ +----+
Synthesized Executive Summary
The Domain Model and Its Schema Contracts
Clean architecture asks one thing of us up front: the data structures at the heart of the domain must never be held hostage by an external framework. An email analysis has to hold two very different kinds of truth side by side — the deterministic classifications produced by Jev, and the generative summaries written by the language model. Modeling both with modern Pydantic v2 structures keeps the application strictly typed through the entire execution lifecycle, from the moment a message arrives to the moment a decision is synced back.
The code below establishes the fundamental domain enumerations and models: message categories, urgency classifications, and the consolidated analysis result.
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, EmailStr
class EmailCategory(str, Enum):
SUPPORT_URGENT = "support_urgent"
SUPPORT_GENERAL = "support_general"
BILLING_INQUIRY = "billing_inquiry"
SECURITY_ALERT = "security_alert"
MARKETING_NEWSLETTER = "marketing_newsletter"
SPAM = "spam"
class UrgencyLevel(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class ProcessingDecision(BaseModel):
category: EmailCategory = Field(
description="Primary operational category determined by Jev reflex evaluation."
)
category_confidence: float = Field(
ge=0.0, le=1.0, description="Calibrated decision confidence score."
)
urgency: UrgencyLevel = Field(
description="Assessed operational urgency level derived from Jev scoring."
)
is_security_incident: bool = Field(
description="Binary reflex flag indicating potential unauthorized access or attack."
)
extracted_account_id: Optional[str] = Field(
default=None, description="Extracted account identifier if identified in message context."
)
summary: str = Field(
description="Executive summary generated by the local or remote generative model."
)
Notice how cleanly the model divides responsibility. Category, confidence, urgency, and the security-incident flag are strict outputs of the Jev decision engine — no ambiguity, no prose. The summary field belongs to the auxiliary language model, the one place where natural language is welcome. And the extracted account identifier records the structured entity IDs discovered during payload normalization, so nothing downstream ever has to parse for them again.
Building the Jev Reflex Classification Engine
Working with the Jev client is disarmingly simple: you submit unstructured state text alongside typed question objects to the TypeSafe AI API. Jev understands three kinds of questions. A Choice question measures the state against a constrained set of discrete categorical strings and returns the winning option together with calibrated probability distributions. A Score question places the input along an integer scale — an ideal mechanism for calculating urgency. A Noul question resolves a binary boolean decision, complete with confidence metrics.
The implementation below constructs the reflex engine itself: it packages an email's sender, subject, and body into a single evaluation state and asks Jev for an instantaneous classification.
import os
import logging
import re
from typing import Optional, Tuple
from jev_sdk import JevClient, Choice, Score, Noul
class JevReflexEngine:
def __init__(self, api_key: Optional[str] = None):
self.logger = logging.getLogger(self.__class__.__name__)
resolved_key = api_key or os.getenv("TYPESAFE_API_KEY") or os.getenv("JEV_API_KEY")
if not resolved_key:
self.logger.error("Typesafe API key is missing from environment and initialization arguments.")
raise ValueError("A valid Typesafe API key must be provided via TYPESAFE_API_KEY.")
self.client = JevClient(api_key=resolved_key)
self.logger.info("Jev reflex engine initialized successfully.")
def evaluate_message(self, sender: str, subject: str, body: str) -> Tuple[EmailCategory, float, UrgencyLevel, bool, Optional[str]]:
state = f"Sender: {sender}\nSubject: {subject}\nBody:\n{body}"
category_question = Choice(
name="email_category",
options=[c.value for c in EmailCategory]
)
urgency_question = Score(
name="operational_urgency",
min_score=1,
max_score=4
)
security_question = Noul(
name="security_threat_detected"
)
results = self.client.evaluate(
state=state,
questions=[category_question, urgency_question, security_question]
)
raw_cat = results["email_category"].selected_option
cat_conf = results["email_category"].confidence
category = EmailCategory(raw_cat)
score_val = results["operational_urgency"].score
score_to_urgency = {
1: UrgencyLevel.LOW,
2: UrgencyLevel.MEDIUM,
3: UrgencyLevel.HIGH,
4: UrgencyLevel.CRITICAL
}
urgency = score_to_urgency.get(score_val, UrgencyLevel.MEDIUM)
is_security = results["security_threat_detected"].value
account_id = None
account_match = re.search(r"\b(ACC-[A-Z0-9]{4,10})\b", body, re.IGNORECASE)
if account_match:
account_id = account_match.group(1).upper()
self.logger.info(
f"Jev reflex evaluated: category={category.value} (conf={cat_conf:.2f}), "
f"urgency={urgency.value}, security_incident={is_security}"
)
return category, cat_conf, urgency, is_security, account_id
The evaluate_message method does its work in a single network interaction: it formats the contextual state and fires all three typed questions simultaneously. It then translates the numeric score into an operational UrgencyLevel and parses the binary security flag. Quietly, in the same pass, a deterministic regular expression lifts structured account tokens straight out of the text — and that last detail matters, because it completely bypasses the non-deterministic hallucination tendencies that make pure text-generation models so hard to trust with identifiers.
Configuring Generative Language Providers, Local and Remote
Jev is brilliant at what it does — rapid, low-latency discriminative classification — but it will not write a sentence a human being wants to read. Synthesizing human-readable summaries calls for an autoregressive generative language model. And here the real world pushes back: in modern enterprise environments, architecture teams often enforce strict data-privacy standards, which means confidential corporate communications must be summarized on local infrastructure rather than dispatched to an external public API. Other workloads carry no such sensitivity and can safely be delegated to high-capacity cloud endpoints.
Rather than choose for you, we define an abstract base class for generative language providers and implement two production-ready adaptations: a local provider that talks to an Ollama or vLLM daemon on your own hardware, and a remote provider compatible with cloud API endpoints. Both lean on nothing heavier than the standard library's networking utilities, so no bulky dependencies creep in.
import json
import urllib.request
import urllib.error
from abc import ABC, abstractmethod
class GenerativeLanguageProvider(ABC):
@abstractmethod
def generate_summary(self, subject: str, body: str) -> str:
pass
class LocalOllamaProvider(GenerativeLanguageProvider):
def __init__(self, endpoint_url: str = "http://localhost:11434/api/generate", model_name: str = "llama3.2:latest"):
self.endpoint_url = endpoint_url
self.model_name = model_name
self.logger = logging.getLogger(self.__class__.__name__)
def generate_summary(self, subject: str, body: str) -> str:
prompt = (
f"You are a professional executive email summarizer. "
f"Write exactly one concise, factual sentence summarizing the following email.\n\n"
f"Subject: {subject}\n"
f"Body:\n{body}\n\n"
f"Summary:"
)
payload = {
"model": self.model_name,
"prompt": prompt,
"stream": False
}
req_data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
self.endpoint_url,
data=req_data,
headers={"Content-Type": "application/json"}
)
try:
with urllib.request.urlopen(req, timeout=30.0) as resp:
resp_data = json.loads(resp.read().decode("utf-8"))
return resp_data.get("response", "").strip()
except Exception as e:
self.logger.warning(f"Local LLM summarization encountered an issue: {str(e)}. Using fallback summary.")
return f"Inbound message regarding '{subject}' requiring operational attention."
class RemoteOpenAICompatibleProvider(GenerativeLanguageProvider):
def __init__(self, api_key: str, endpoint_url: str = "https://api.openai.com/v1/chat/completions", model_name: str = "gpt-4o-mini"):
self.api_key = api_key
self.endpoint_url = endpoint_url
self.model_name = model_name
self.logger = logging.getLogger(self.__class__.__name__)
def generate_summary(self, subject: str, body: str) -> str:
payload = {
"model": self.model_name,
"messages": [
{"role": "system", "content": "You are an executive assistant. Summarize incoming emails in one clear sentence."},
{"role": "user", "content": f"Subject: {subject}\n\nBody:\n{body}"}
],
"temperature": 0.2,
"max_tokens": 100
}
req_data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
self.endpoint_url,
data=req_data,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
)
try:
with urllib.request.urlopen(req, timeout=30.0) as resp:
resp_data = json.loads(resp.read().decode("utf-8"))
choices = resp_data.get("choices", [])
if choices:
return choices[0]["message"]["content"].strip()
return f"Email regarding '{subject}'."
except Exception as e:
self.logger.warning(f"Remote LLM summarization failed: {str(e)}. Using fallback summary.")
return f"Inbound communication regarding '{subject}'."
The provider pattern is what keeps this architecture honest — it decouples Jev's split-second reflex decisions from the slower art of text generation. The LocalOllamaProvider posts directly to the local Ollama REST interface, which guarantees that private email content never crosses the corporate firewall. Organizations running on hosted infrastructure can swap in the RemoteOpenAICompatibleProvider without touching a single line of business orchestration logic. And because networks fail at the worst possible moments, both classes ship with defensive timeout protections and fallback summaries, so a partition becomes a log line rather than an outage.
Taming Google Mail: Robust Authentication and MIME Ingestion
Anyone who has wired an automated enterprise pipeline into Google Mail remembers the two places it hurts. The first is headless OAuth token handling: an interactive development script can spin up a local web server and catch the consent redirect, but a production worker running inside a containerized cloud instance has no browser to lean on — it must rely on securely persisted refresh tokens or service-account credentials. The second is MIME itself. Real email arrives in every conceivable configuration: nested multipart alternative bodies carrying plain text and rich HTML side by side, embedded inline attachments, and character encodings from across the decades.
The implementation below resolves both challenges head-on, pairing resilient token loading with a recursive multipart traversal algorithm that walks the payload tree until it finds readable text.
import base64
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build, Resource
class GoogleMailIngestionService:
def __init__(self, token_path: str = "token.json"):
self.logger = logging.getLogger(self.__class__.__name__)
self.service = self._initialize_service(token_path)
def _initialize_service(self, token_path: str) -> Resource:
scopes = ["https://www.googleapis.com/auth/gmail.modify"]
if not os.path.exists(token_path):
self.logger.error(f"Required Gmail authorization file not found at {token_path}.")
raise FileNotFoundError(f"Missing authorized token file: {token_path}")
creds = Credentials.from_authorized_user_file(token_path, scopes)
if creds.expired and creds.refresh_token:
self.logger.info("Refreshing expired Gmail API credentials using stored refresh token.")
creds.refresh(Request())
with open(token_path, "w") as token_file:
token_file.write(creds.to_json())
if not creds.valid:
self.logger.critical("Loaded Google Mail credentials are invalid and cannot be refreshed automatically.")
raise PermissionError("Invalid Google Mail credentials provided.")
return build("gmail", "v1", credentials=creds)
def retrieve_unread_messages(self, limit: int = 10) -> list[dict]:
query = "is:unread"
response = self.service.users().messages().list(
userId="me",
q=query,
maxResults=limit
).execute()
messages_meta = response.get("messages", [])
self.logger.info(f"Discovered {len(messages_meta)} unread email items in mailbox.")
parsed_batch = []
for meta in messages_meta:
full_item = self.service.users().messages().get(
userId="me",
id=meta["id"],
format="full"
).execute()
parsed = self._extract_email_components(full_item)
parsed_batch.append(parsed)
return parsed_batch
def _extract_email_components(self, message: dict) -> dict:
headers = message.get("payload", {}).get("headers", [])
subject = "No Subject"
sender = "unknown@domain.com"
for header in headers:
name = header.get("name", "").lower()
if name == "subject":
subject = header.get("value", "")
elif name == "from":
sender = header.get("value", "")
body = self._traverse_payload_for_body(message.get("payload", {}))
return {
"id": message.get("id"),
"sender": sender,
"subject": subject,
"body": body
}
def _traverse_payload_for_body(self, payload: dict) -> str:
if "body" in payload and payload["body"].get("data"):
raw_bytes = base64.urlsafe_b64decode(payload["body"]["data"].encode("utf-8"))
return raw_bytes.decode("utf-8", errors="replace")
parts = payload.get("parts", [])
plain_candidate = ""
html_candidate = ""
for part in parts:
mime_type = part.get("mimeType", "")
if mime_type == "text/plain" and part.get("body", {}).get("data"):
raw_bytes = base64.urlsafe_b64decode(part["body"]["data"].encode("utf-8"))
return raw_bytes.decode("utf-8", errors="replace")
elif mime_type == "text/html" and part.get("body", {}).get("data"):
raw_bytes = base64.urlsafe_b64decode(part["body"]["data"].encode("utf-8"))
html_candidate = raw_bytes.decode("utf-8", errors="replace")
elif "parts" in part:
nested_result = self._traverse_payload_for_body(part)
if nested_result:
return nested_result
fallback_text = plain_candidate or html_candidate
return fallback_text if fallback_text else "No extractable text content available."
The GoogleMailIngestionService is built to be run repeatedly and trusted. At startup it verifies that the authorized token file exists, then seamlessly refreshes expired access tokens through the stored refresh token — no human in the loop, no manual re-consent. The recursive helper _traverse_payload_for_body prioritizes plain text payloads, falls back gracefully to HTML representations, descends into nested multipart parts when it must, and replaces unrecognized byte sequences rather than letting a decoding crash take the worker down.
Closing the Loop: Label Synchronization and Alerting in Google Mail
A decision that never leaves the pipeline is a decision that never happened. Once an email has been evaluated by the Jev reflex engine and summarized by the language provider, the application must write those judgments back into the mailbox environment — so that human operators can locate classified emails at a glance, see urgent correspondence surface in its designated filtered views, and trust that processed messages stop cluttering the unread inbox queue.
Google Mail organizes all of this through hierarchical labels. Creating labels dynamically means first checking whether a label already exists and caching its identifier; otherwise every processed message costs you redundant network roundtrips. The class below implements exactly that discipline: label creation, caching, and batch synchronization.
class GoogleMailActionDispatcher:
def __init__(self, gmail_service: Resource):
self.service = gmail_service
self.logger = logging.getLogger(self.__class__.__name__)
self._label_cache: dict[str, str] = {}
self._prime_label_cache()
def _prime_label_cache(self) -> None:
response = self.service.users().labels().list(userId="me").execute()
for label in response.get("labels", []):
self._label_cache[label["name"].lower()] = label["id"]
def _resolve_or_create_label(self, label_name: str) -> str:
cached_id = self._label_cache.get(label_name.lower())
if cached_id:
return cached_id
label_payload = {
"name": label_name,
"labelListVisibility": "labelShow",
"messageListVisibility": "show"
}
created = self.service.users().labels().create(
userId="me",
body=label_payload
).execute()
label_id = created["id"]
self._label_cache[label_name.lower()] = label_id
self.logger.info(f"Created new Google Mail label '{label_name}' with ID {label_id}.")
return label_id
def synchronize_decision(self, message_id: str, decision: ProcessingDecision) -> None:
category_label = f"JEV/{decision.category.value.upper()}"
category_label_id = self._resolve_or_create_label(category_label)
add_labels = [category_label_id]
if decision.urgency in (UrgencyLevel.HIGH, UrgencyLevel.CRITICAL):
urgency_label_id = self._resolve_or_create_label("JEV/URGENT_ATTENTION")
add_labels.append(urgency_label_id)
if decision.is_security_incident:
security_label_id = self._resolve_or_create_label("JEV/SECURITY_THREAT")
add_labels.append(security_label_id)
self.logger.critical(
f"IMMEDIATE SECURITY ACTION: Message {message_id} flagged as attack or intrusion. "
f"Account ID: {decision.extracted_account_id}. Summary: {decision.summary}"
)
self.service.users().messages().batchModify(
userId="me",
body={
"ids": [message_id],
"addLabelIds": add_labels,
"removeLabelIds": ["UNREAD"]
}
).execute()
self.logger.info(
f"Successfully synchronized message {message_id}: applied labels {add_labels} and cleared UNREAD status."
)
The GoogleMailActionDispatcher keeps an in-memory cache of every label identifier it has established. When an email needs a category label, the dispatcher consults the cache first and creates the label on demand only if it's absent. And when a message comes back marked as critically urgent — or worse, as a security threat — secondary escalation labels are appended on the spot, triggering immediate audit logs and operational alerts.
The Complete, Running Implementation
Everything discussed so far, assembled for real. The standalone Python program below unifies every architectural component into a single executable pipeline — full error handling, clean architecture interfaces, support for both local and remote language models, genuine Jev SDK integration, and live Google Mail synchronization. No mocks, no placeholders.
import os
import sys
import json
import base64
import re
import logging
import urllib.request
import urllib.error
from abc import ABC, abstractmethod
from enum import Enum
from typing import Optional, Tuple, List, Dict, Any
from pydantic import BaseModel, Field
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build, Resource
from jev_sdk import JevClient, Choice, Score, Noul
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
stream=sys.stdout
)
class EmailCategory(str, Enum):
SUPPORT_URGENT = "support_urgent"
SUPPORT_GENERAL = "support_general"
BILLING_INQUIRY = "billing_inquiry"
SECURITY_ALERT = "security_alert"
MARKETING_NEWSLETTER = "marketing_newsletter"
SPAM = "spam"
class UrgencyLevel(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class ProcessingDecision(BaseModel):
category: EmailCategory = Field(
description="Primary operational category determined by Jev reflex evaluation."
)
category_confidence: float = Field(
ge=0.0, le=1.0, description="Calibrated decision confidence score."
)
urgency: UrgencyLevel = Field(
description="Assessed operational urgency level derived from Jev scoring."
)
is_security_incident: bool = Field(
description="Binary reflex flag indicating potential unauthorized access or attack."
)
extracted_account_id: Optional[str] = Field(
default=None, description="Extracted account identifier if identified in message context."
)
summary: str = Field(
description="Executive summary generated by the local or remote generative model."
)
class GenerativeLanguageProvider(ABC):
@abstractmethod
def generate_summary(self, subject: str, body: str) -> str:
pass
class LocalOllamaProvider(GenerativeLanguageProvider):
def __init__(self, endpoint_url: str = "http://localhost:11434/api/generate", model_name: str = "llama3.2:latest"):
self.endpoint_url = endpoint_url
self.model_name = model_name
self.logger = logging.getLogger(self.__class__.__name__)
def generate_summary(self, subject: str, body: str) -> str:
prompt = (
f"You are a professional executive email summarizer. "
f"Write exactly one concise, factual sentence summarizing the following email.\n\n"
f"Subject: {subject}\n"
f"Body:\n{body}\n\n"
f"Summary:"
)
payload = {
"model": self.model_name,
"prompt": prompt,
"stream": False
}
req_data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
self.endpoint_url,
data=req_data,
headers={"Content-Type": "application/json"}
)
try:
with urllib.request.urlopen(req, timeout=30.0) as resp:
resp_data = json.loads(resp.read().decode("utf-8"))
return resp_data.get("response", "").strip()
except Exception as e:
self.logger.warning(f"Local LLM summarization encountered an issue: {str(e)}. Using fallback summary.")
return f"Inbound message regarding '{subject}' requiring operational attention."
class RemoteOpenAICompatibleProvider(GenerativeLanguageProvider):
def __init__(self, api_key: str, endpoint_url: str = "https://api.openai.com/v1/chat/completions", model_name: str = "gpt-4o-mini"):
self.api_key = api_key
self.endpoint_url = endpoint_url
self.model_name = model_name
self.logger = logging.getLogger(self.__class__.__name__)
def generate_summary(self, subject: str, body: str) -> str:
payload = {
"model": self.model_name,
"messages": [
{"role": "system", "content": "You are an executive assistant. Summarize incoming emails in one clear sentence."},
{"role": "user", "content": f"Subject: {subject}\n\nBody:\n{body}"}
],
"temperature": 0.2,
"max_tokens": 100
}
req_data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
self.endpoint_url,
data=req_data,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
)
try:
with urllib.request.urlopen(req, timeout=30.0) as resp:
resp_data = json.loads(resp.read().decode("utf-8"))
choices = resp_data.get("choices", [])
if choices:
return choices[0]["message"]["content"].strip()
return f"Email regarding '{subject}'."
except Exception as e:
self.logger.warning(f"Remote LLM summarization failed: {str(e)}. Using fallback summary.")
return f"Inbound communication regarding '{subject}'."
class JevReflexEngine:
def __init__(self, api_key: Optional[str] = None):
self.logger = logging.getLogger(self.__class__.__name__)
resolved_key = api_key or os.getenv("TYPESAFE_API_KEY") or os.getenv("JEV_API_KEY")
if not resolved_key:
self.logger.error("Typesafe API key is missing from environment and initialization arguments.")
raise ValueError("A valid Typesafe API key must be provided via TYPESAFE_API_KEY.")
self.client = JevClient(api_key=resolved_key)
self.logger.info("Jev reflex engine initialized successfully.")
def evaluate_message(self, sender: str, subject: str, body: str) -> Tuple[EmailCategory, float, UrgencyLevel, bool, Optional[str]]:
state = f"Sender: {sender}\nSubject: {subject}\nBody:\n{body}"
category_question = Choice(
name="email_category",
options=[c.value for c in EmailCategory]
)
urgency_question = Score(
name="operational_urgency",
min_score=1,
max_score=4
)
security_question = Noul(
name="security_threat_detected"
)
results = self.client.evaluate(
state=state,
questions=[category_question, urgency_question, security_question]
)
raw_cat = results["email_category"].selected_option
cat_conf = results["email_category"].confidence
category = EmailCategory(raw_cat)
score_val = results["operational_urgency"].score
score_to_urgency = {
1: UrgencyLevel.LOW,
2: UrgencyLevel.MEDIUM,
3: UrgencyLevel.HIGH,
4: UrgencyLevel.CRITICAL
}
urgency = score_to_urgency.get(score_val, UrgencyLevel.MEDIUM)
is_security = results["security_threat_detected"].value
account_id = None
account_match = re.search(r"\b(ACC-[A-Z0-9]{4,10})\b", body, re.IGNORECASE)
if account_match:
account_id = account_match.group(1).upper()
self.logger.info(
f"Jev reflex evaluated: category={category.value} (conf={cat_conf:.2f}), "
f"urgency={urgency.value}, security_incident={is_security}"
)
return category, cat_conf, urgency, is_security, account_id
class GoogleMailIngestionService:
def __init__(self, token_path: str = "token.json"):
self.logger = logging.getLogger(self.__class__.__name__)
self.service = self._initialize_service(token_path)
def _initialize_service(self, token_path: str) -> Resource:
scopes = ["https://www.googleapis.com/auth/gmail.modify"]
if not os.path.exists(token_path):
self.logger.error(f"Required Gmail authorization file not found at {token_path}.")
raise FileNotFoundError(f"Missing authorized token file: {token_path}")
creds = Credentials.from_authorized_user_file(token_path, scopes)
if creds.expired and creds.refresh_token:
self.logger.info("Refreshing expired Gmail API credentials using stored refresh token.")
creds.refresh(Request())
with open(token_path, "w") as token_file:
token_file.write(creds.to_json())
if not creds.valid:
self.logger.critical("Loaded Google Mail credentials are invalid and cannot be refreshed automatically.")
raise PermissionError("Invalid Google Mail credentials provided.")
return build("gmail", "v1", credentials=creds)
def retrieve_unread_messages(self, limit: int = 10) -> List[Dict[str, Any]]:
query = "is:unread"
response = self.service.users().messages().list(
userId="me",
q=query,
maxResults=limit
).execute()
messages_meta = response.get("messages", [])
self.logger.info(f"Discovered {len(messages_meta)} unread email items in mailbox.")
parsed_batch = []
for meta in messages_meta:
full_item = self.service.users().messages().get(
userId="me",
id=meta["id"],
format="full"
).execute()
parsed = self._extract_email_components(full_item)
parsed_batch.append(parsed)
return parsed_batch
def _extract_email_components(self, message: Dict[str, Any]) -> Dict[str, Any]:
headers = message.get("payload", {}).get("headers", [])
subject = "No Subject"
sender = "unknown@domain.com"
for header in headers:
name = header.get("name", "").lower()
if name == "subject":
subject = header.get("value", "")
elif name == "from":
sender = header.get("value", "")
body = self._traverse_payload_for_body(message.get("payload", {}))
return {
"id": message.get("id"),
"sender": sender,
"subject": subject,
"body": body
}
def _traverse_payload_for_body(self, payload: Dict[str, Any]) -> str:
if "body" in payload and payload["body"].get("data"):
raw_bytes = base64.urlsafe_b64decode(payload["body"]["data"].encode("utf-8"))
return raw_bytes.decode("utf-8", errors="replace")
parts = payload.get("parts", [])
plain_candidate = ""
html_candidate = ""
for part in parts:
mime_type = part.get("mimeType", "")
if mime_type == "text/plain" and part.get("body", {}).get("data"):
raw_bytes = base64.urlsafe_b64decode(part["body"]["data"].encode("utf-8"))
return raw_bytes.decode("utf-8", errors="replace")
elif mime_type == "text/html" and part.get("body", {}).get("data"):
raw_bytes = base64.urlsafe_b64decode(part["body"]["data"].encode("utf-8"))
html_candidate = raw_bytes.decode("utf-8", errors="replace")
elif "parts" in part:
nested_result = self._traverse_payload_for_body(part)
if nested_result:
return nested_result
fallback_text = plain_candidate or html_candidate
return fallback_text if fallback_text else "No extractable text content available."
class GoogleMailActionDispatcher:
def __init__(self, gmail_service: Resource):
self.service = gmail_service
self.logger = logging.getLogger(self.__class__.__name__)
self._label_cache: Dict[str, str] = {}
self._prime_label_cache()
def _prime_label_cache(self) -> None:
response = self.service.users().labels().list(userId="me").execute()
for label in response.get("labels", []):
self._label_cache[label["name"].lower()] = label["id"]
def _resolve_or_create_label(self, label_name: str) -> str:
cached_id = self._label_cache.get(label_name.lower())
if cached_id:
return cached_id
label_payload = {
"name": label_name,
"labelListVisibility": "labelShow",
"messageListVisibility": "show"
}
created = self.service.users().labels().create(
userId="me",
body=label_payload
).execute()
label_id = created["id"]
self._label_cache[label_name.lower()] = label_id
self.logger.info(f"Created new Google Mail label '{label_name}' with ID {label_id}.")
return label_id
def synchronize_decision(self, message_id: str, decision: ProcessingDecision) -> None:
category_label = f"JEV/{decision.category.value.upper()}"
category_label_id = self._resolve_or_create_label(category_label)
add_labels = [category_label_id]
if decision.urgency in (UrgencyLevel.HIGH, UrgencyLevel.CRITICAL):
urgency_label_id = self._resolve_or_create_label("JEV/URGENT_ATTENTION")
add_labels.append(urgency_label_id)
if decision.is_security_incident:
security_label_id = self._resolve_or_create_label("JEV/SECURITY_THREAT")
add_labels.append(security_label_id)
self.logger.critical(
f"IMMEDIATE SECURITY ACTION: Message {message_id} flagged as attack or intrusion. "
f"Account ID: {decision.extracted_account_id}. Summary: {decision.summary}"
)
self.service.users().messages().batchModify(
userId="me",
body={
"ids": [message_id],
"addLabelIds": add_labels,
"removeLabelIds": ["UNREAD"]
}
).execute()
self.logger.info(
f"Successfully synchronized message {message_id}: applied labels {add_labels} and cleared UNREAD status."
)
class IntelligentEmailOrchestrator:
def __init__(
self,
ingestion_service: GoogleMailIngestionService,
reflex_engine: JevReflexEngine,
generative_provider: GenerativeLanguageProvider,
dispatcher: GoogleMailActionDispatcher
):
self.ingestion = ingestion_service
self.reflex = reflex_engine
self.llm = generative_provider
self.dispatcher = dispatcher
self.logger = logging.getLogger(self.__class__.__name__)
def execute_processing_cycle(self, batch_size: int = 10) -> int:
self.logger.info("Initiating orchestrated email processing cycle.")
messages = self.ingestion.retrieve_unread_messages(limit=batch_size)
if not messages:
self.logger.info("No unread correspondence found. Processing cycle complete.")
return 0
failed = 0
for msg in messages:
msg_id = msg["id"]
try:
sender = msg["sender"]
subject = msg["subject"]
body = msg["body"]
self.logger.info(f"Processing message {msg_id} from {sender} with subject '{subject}'.")
category, conf, urgency, is_security, account_id = self.reflex.evaluate_message(
sender=sender,
subject=subject,
body=body
)
summary_text = self.llm.generate_summary(subject=subject, body=body)
decision = ProcessingDecision(
category=category,
category_confidence=conf,
urgency=urgency,
is_security_incident=is_security,
extracted_account_id=account_id,
summary=summary_text
)
self.logger.info(f"Consolidated Decision Payload: {decision.model_dump_json()}")
self.dispatcher.synchronize_decision(message_id=msg_id, decision=decision)
except Exception as ex:
failed += 1
self.logger.error(f"Message {msg_id} failed and stays UNREAD for retry: {ex}")
total = len(messages)
if failed:
self.logger.warning(f"Cycle finished: {total - failed}/{total} succeeded, {failed} failed.")
else:
self.logger.info(f"All {total} messages processed successfully.")
return failed
def run_production_pipeline() -> None:
logger = logging.getLogger("ApplicationBootstrap")
logger.info("Bootstrapping Intelligent Email Processing Service.")
try:
reflex_engine = JevReflexEngine()
use_local_llm = os.getenv("USE_LOCAL_LLM", "false").lower() in ("true", "1", "yes")
if use_local_llm:
logger.info("Configuring local Ollama language provider for privacy-preserving summarization.")
llm_provider = LocalOllamaProvider(
endpoint_url=os.getenv("OLLAMA_ENDPOINT", "http://localhost:11434/api/generate"),
model_name=os.getenv("OLLAMA_MODEL", "llama3.2:latest")
)
else:
remote_key = os.getenv("OPENAI_API_KEY", "").strip()
if not remote_key:
raise ValueError("OPENAI_API_KEY must be set when USE_LOCAL_LLM is not enabled.")
logger.info("Configuring remote cloud language provider for summarization.")
llm_provider = RemoteOpenAICompatibleProvider(
api_key=remote_key,
endpoint_url=os.getenv("OPENAI_ENDPOINT", "https://api.openai.com/v1/chat/completions"),
model_name=os.getenv("OPENAI_MODEL", "gpt-4o-mini")
)
token_path = os.getenv("GMAIL_TOKEN_PATH", "token.json")
ingestion_service = GoogleMailIngestionService(token_path=token_path)
dispatcher = GoogleMailActionDispatcher(gmail_service=ingestion_service.service)
orchestrator = IntelligentEmailOrchestrator(
ingestion_service=ingestion_service,
reflex_engine=reflex_engine,
generative_provider=llm_provider,
dispatcher=dispatcher
)
failed = orchestrator.execute_processing_cycle(batch_size=10)
if failed:
logger.error(f"Pipeline completed with {failed} failed message(s).")
sys.exit(2)
logger.info("Pipeline execution completed without errors.")
except Exception as ex:
logger.critical(f"Fatal error during pipeline execution: {str(ex)}")
sys.exit(1)
if __name__ == "__main__":
run_production_pipeline()
Where We Go from Here: Conclusion and Future Outlook
The architectural separation of high-speed discriminative reflexes from deliberate generative synthesis is more than an implementation detail — it marks a genuine milestone in how agentic AI systems ought to be designed. Hand high-volume classification, urgency scoring, and guardrail validation to Jev, and an enterprise system gains deterministic, sub-second decision-making at a fraction of standard operational costs. Leave the auxiliary generative models — whether hosted locally via Ollama or reached through remote cloud endpoints — free to do the one thing they do best: producing natural language summaries and syntheses. Wire the two into a productivity platform like Google Mail, and what emerges is a resilient, scalable, genuinely production-ready foundation for modern autonomous operations — a system that thinks fast when it should, and thinks deeply only when it must.
