What is an Email Automation Agent?
An Email Automation Agent is an intelligent system that can read, triage, draft, and send emails autonomously based on predefined rules and natural language instructions. Unlike traditional email filters that operate on simple keyword matching, an agent built with LangGraph leverages large language models (LLMs) to understand the semantic content of emails, make nuanced decisions, and maintain context across multiple interactions. It can handle tasks such as:
- Scanning your inbox for high-priority messages and summarizing them
- Drafting contextual replies based on conversation history
- Scheduling follow-up emails when certain conditions are met
- Extracting action items and creating to-do entries automatically
- Routing emails to appropriate handlers or departments
LangGraph provides the perfect framework for this because email automation is inherently stateful and multi-step. You need to fetch emails, process them one by one, decide whether to respond, draft a response, potentially request human approval, and then send—all while keeping track of which emails have been processed, what decisions were made, and what the current state of each conversation is.
Why LangGraph for Email Automation?
Traditional frameworks like LangChain are great for single-shot LLM calls, but email automation requires a persistent state machine that can pause, branch, and resume. LangGraph excels here because:
- Explicit State Management: You define a typed state object that flows through nodes. For email agents, this state carries the inbox snapshot, per-email processing status, drafted replies, and approval flags.
- Conditional Routing: LangGraph lets you define edges that branch based on LLM output. For example, after classifying an email's urgency, you can route to "immediate_reply," "defer_to_human," or "archive" nodes.
- Human-in-the-Loop: You can insert interrupt points where the agent pauses and waits for human approval before sending sensitive emails—a critical feature for production email systems.
- Observability & Debugging: LangGraph's graph structure makes it easy to visualize the agent's decision path and debug why a particular email was or wasn't responded to.
- Streaming Support: You can stream intermediate states to a frontend dashboard, showing real-time progress as the agent processes your inbox.
Prerequisites and Setup
Before diving into code, make sure you have the following installed and configured:
pip install langgraph langchain langchain-openai python-dotenv imapclient email-validator
pip install smtplib email-validator # usually included in Python standard libraryYou'll also need:
- An OpenAI API key (or any compatible LLM provider) stored in a
.envfile - IMAP/SMTP credentials for your email provider (or Gmail API credentials if using Gmail)
- Python 3.10+ for best compatibility with LangGraph
Create your .env file:
OPENAI_API_KEY=sk-your-key-here
EMAIL_ADDRESS=your-email@example.com
EMAIL_PASSWORD=your-app-password-or-regular-password
IMAP_SERVER=imap.gmail.com
SMTP_SERVER=smtp.gmail.com
SMTP_PORT=587Step-by-Step Implementation
Step 1: Define the Agent State
The state is the heart of a LangGraph agent. For our email automation agent, we need to track the inbox contents, processing status for each email, generated responses, and any human approval flags. Here's a comprehensive state definition using Python's TypedDict and LangGraph's Annotation:
from typing import TypedDict, List, Dict, Optional, Annotated, Literal
from langgraph.graph.message import add_messages
import operator
class EmailContent(TypedDict):
"""Structure for a single email's parsed content."""
email_id: str
sender: str
subject: str
body: str
received_at: str
thread_id: Optional[str]
class ProcessedEmail(TypedDict):
"""Tracks the agent's processing result for one email."""
email_id: str
urgency: Literal["high", "medium", "low"]
category: Literal["spam", "newsletter", "personal", "work", "urgent", "other"]
action_taken: Literal["replied", "deferred", "archived", "flagged_for_review"]
drafted_reply: Optional[str]
human_approved: Optional[bool]
class EmailAgentState(TypedDict):
"""Complete state flowing through the agent graph."""
# Raw inbox data
new_emails: List[EmailContent]
# Processing queue — emails yet to be processed
processing_queue: List[str] # list of email_ids
# Results for each processed email
processed_emails: Annotated[Dict[str, ProcessedEmail], operator.or_]
# Using operator.or_ as the reducer to merge dicts across parallel branches
# Current email being processed (for node-level context)
current_email_id: Optional[str]
current_email: Optional[EmailContent]
# LLM conversation context
messages: Annotated[List, add_messages]
# Human-in-the-loop flags
pending_approval: bool
approval_granted: Optional[bool]
# Final summary for the user
summary_report: Optional[str]
# Error tracking
errors: List[str]The Annotated type with reducers like operator.or_ and add_messages is critical—it tells LangGraph how to merge state updates when multiple branches write to the same key. For processed_emails, we use operator.or_ (which actually performs a dict union via the | operator in Python 3.9+) so that each processed email's result is accumulated into a single dictionary.
Step 2: Set Up Email Fetching Tools
We'll create two essential tool functions: one for fetching emails via IMAP and another for sending emails via SMTP. These are wrapped as LangChain tools so the LLM can optionally invoke them (though in our graph, we'll call them directly in dedicated nodes for reliability):
import imapclient
import email
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import smtplib
import os
from datetime import datetime, timedelta
from dotenv import load_dotenv
load_dotenv()
def fetch_recent_emails(hours_lookback: int = 24, max_emails: int = 50) -> List[EmailContent]:
"""
Fetch recent emails from the configured IMAP server.
Returns a list of parsed EmailContent dicts.
"""
imap = imapclient.IMAPClient(
os.getenv("IMAP_SERVER"),
ssl=True
)
try:
imap.login(os.getenv("EMAIL_ADDRESS"), os.getenv("EMAIL_PASSWORD"))
imap.select_folder("INBOX")
# Search for emails from the last N hours
since_date = (datetime.now() - timedelta(hours=hours_lookback)).strftime("%d-%b-%Y")
search_criteria = [f"SINCE {since_date}"]
message_ids = imap.search(search_criteria)
# Limit to max_emails, most recent first
message_ids = message_ids[-max_emails:] if len(message_ids) > max_emails else message_ids
emails = []
for msg_id in message_ids:
raw_data = imap.fetch([msg_id], ["RFC822", "INTERNALDATE"])
msg_bytes = raw_data[msg_id][b"RFC822"]
internal_date = raw_data[msg_id][b"INTERNALDATE"]
parsed = email.message_from_bytes(msg_bytes)
# Extract body (handle multipart)
body = ""
if parsed.is_multipart():
for part in parsed.walk():
if part.get_content_type() == "text/plain":
body_bytes = part.get_payload(decode=True)
if body_bytes:
body = body_bytes.decode(errors="replace")
break
else:
body_bytes = parsed.get_payload(decode=True)
if body_bytes:
body = body_bytes.decode(errors="replace")
emails.append(EmailContent(
email_id=str(msg_id),
sender=parsed.get("From", "unknown"),
subject=parsed.get("Subject", "No Subject"),
body=body[:2000], # Truncate very long emails for LLM context
received_at=str(internal_date),
thread_id=parsed.get("Message-ID") or None
))
return emails
finally:
imap.logout()
def send_email(to_address: str, subject: str, body: str, in_reply_to: Optional[str] = None) -> bool:
"""
Send an email via SMTP. Returns True if successful.
"""
msg = MIMEMultipart()
msg["From"] = os.getenv("EMAIL_ADDRESS")
msg["To"] = to_address
msg["Subject"] = subject
if in_reply_to:
msg["In-Reply-To"] = in_reply_to
msg["References"] = in_reply_to
msg.attach(MIMEText(body, "plain"))
try:
with smtplib.SMTP(os.getenv("SMTP_SERVER"), int(os.getenv("SMTP_PORT"))) as server:
server.starttls()
server.login(os.getenv("EMAIL_ADDRESS"), os.getenv("EMAIL_PASSWORD"))
server.send_message(msg)
return True
except Exception as e:
print(f"Failed to send email: {e}")
return FalseStep 3: Create the LLM-Powered Classification Node
This node uses an LLM to analyze a single email and determine its urgency, category, and recommended action. The output is structured JSON that we parse and store in the state:
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
import json
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
CLASSIFICATION_PROMPT = """You are an email triage assistant. Analyze the following email and provide a JSON response with these fields:
- urgency: "high", "medium", or "low"
- category: "spam", "newsletter", "personal", "work", "urgent", or "other"
- should_reply: true or false
- reply_draft: a suggested reply (only if should_reply is true, otherwise empty string)
- reason: brief explanation for your classification
High urgency indicators: deadline mentions, direct asks from managers/clients, security alerts, time-sensitive opportunities.
Low urgency: newsletters, automated notifications, casual personal emails.
Respond ONLY with valid JSON, no other text."""
def classify_email_node(state: EmailAgentState) -> EmailAgentState:
"""
Node: Classify the current email using LLM.
Expects state.current_email to be populated.
"""
if not state.get("current_email"):
return state
email_data = state["current_email"]
human_msg = HumanMessage(content=f"""
Sender: {email_data['sender']}
Subject: {email_data['subject']}
Body:
{email_data['body']}
""")
messages = [SystemMessage(content=CLASSIFICATION_PROMPT), human_msg]
try:
response = llm.invoke(messages)
classification = json.loads(response.content)
except (json.JSONDecodeError, KeyError) as e:
# Fallback: conservative classification on parse failure
classification = {
"urgency": "medium",
"category": "other",
"should_reply": False,
"reply_draft": "",
"reason": f"LLM parse error: {str(e)}"
}
processed = ProcessedEmail(
email_id=email_data["email_id"],
urgency=classification.get("urgency", "medium"),
category=classification.get("category", "other"),
action_taken="flagged_for_review", # Will be updated after routing
drafted_reply=classification.get("reply_draft"),
human_approved=None
)
# Update state
new_processed = state.get("processed_emails", {})
new_processed[email_data["email_id"]] = processed
return {
**state,
"processed_emails": new_processed,
"pending_approval": classification.get("urgency") == "high",
"messages": state.get("messages", []) + messages + [response]
}Step 4: Build the Conditional Routing Logic
After classification, we need to route the email to the appropriate handler. LangGraph uses conditional edges for this—a function that inspects the state and returns the name of the next node:
def route_after_classification(state: EmailAgentState) -> Literal["draft_reply", "flag_for_review", "archive", "handle_error"]:
"""
Conditional routing function.
Inspects the processed result for the current email and decides next step.
"""
email_id = state.get("current_email_id")
if not email_id:
return "handle_error"
processed = state.get("processed_emails", {}).get(email_id)
if not processed:
return "handle_error"
urgency = processed.get("urgency", "medium")
category = processed.get("category", "other")
# Routing logic
if category == "spam":
return "archive"
elif urgency == "high":
return "flag_for_review" # Requires human approval
elif urgency == "medium" and category in ("work", "personal", "urgent"):
return "draft_reply"
else:
return "archive"Step 5: Implement the Action Nodes
Now we build the nodes that actually perform actions—drafting replies, flagging for review, archiving, and sending. Each node modifies the state and may trigger side effects (like actually sending an email):
def draft_reply_node(state: EmailAgentState) -> EmailAgentState:
"""
Node: Refine the reply draft using the LLM with conversation context.
"""
email_data = state.get("current_email")
if not email_data:
return state
email_id = email_data["email_id"]
processed = state.get("processed_emails", {}).get(email_id, {})
# Use the LLM to polish the draft
refine_prompt = f"""
You are drafting a professional email reply. Here is the original email:
Sender: {email_data['sender']}
Subject: {email_data['subject']}
Body: {email_data['body']}
Draft a concise, friendly, and appropriate reply.
The initial suggested reply was: "{processed.get('drafted_reply', '')}"
Improve it and return ONLY the final reply text, no commentary.
"""
refined = llm.invoke([HumanMessage(content=refine_prompt)])
# Update the processed entry
email_id = email_data["email_id"]
updated = dict(processed)
updated["drafted_reply"] = refined.content
updated["action_taken"] = "replied"
new_processed = dict(state.get("processed_emails", {}))
new_processed[email_id] = updated
# Actually send the email
send_success = send_email(
to_address=email_data["sender"],
subject=f"Re: {email_data['subject']}",
body=refined.content,
in_reply_to=email_data.get("thread_id")
)
errors = state.get("errors", [])
if not send_success:
errors.append(f"Failed to send reply to {email_data['sender']}")
return {
**state,
"processed_emails": new_processed,
"errors": errors
}
def flag_for_review_node(state: EmailAgentState) -> EmailAgentState:
"""
Node: Mark email as requiring human approval before any action.
Sets pending_approval=True and pauses execution.
"""
email_data = state.get("current_email")
if not email_data:
return state
email_id = email_data["email_id"]
processed = state.get("processed_emails", {}).get(email_id, {})
updated = dict(processed)
updated["action_taken"] = "flagged_for_review"
updated["human_approved"] = None
new_processed = dict(state.get("processed_emails", {}))
new_processed[email_id] = updated
return {
**state,
"processed_emails": new_processed,
"pending_approval": True
}
def archive_node(state: EmailAgentState) -> EmailAgentState:
"""
Node: Mark email as archived/low-priority. No action needed.
"""
email_data = state.get("current_email")
if not email_data:
return state
email_id = email_data["email_id"]
processed = state.get("processed_emails", {}).get(email_id, {})
updated = dict(processed)
updated["action_taken"] = "archived"
updated["drafted_reply"] = None
new_processed = dict(state.get("processed_emails", {}))
new_processed[email_id] = updated
return {
**state,
"processed_emails": new_processed,
"pending_approval": False
}
def handle_error_node(state: EmailAgentState) -> EmailAgentState:
"""
Node: Log errors and skip the problematic email.
"""
errors = state.get("errors", [])
errors.append(f"Error processing email {state.get('current_email_id', 'unknown')}")
return {
**state,
"errors": errors,
"pending_approval": False
}Step 6: Implement the Human-in-the-Loop Interrupt
For high-urgency emails, we want the agent to pause and wait for human approval before sending. LangGraph supports this via interrupts. We define a special node that triggers the interrupt, and a separate node that resumes after approval:
def human_approval_node(state: EmailAgentState) -> EmailAgentState:
"""
Node: This is where the graph INTERRUPTS for human input.
In LangGraph, you call graph.interrupt() inside a node to pause execution.
The graph resumes when you call graph.update_state() with the approval result.
"""
email_data = state.get("current_email")
if not email_data:
return state
# Build a summary for the human reviewer
email_id = email_data["email_id"]
processed = state.get("processed_emails", {}).get(email_id, {})
summary = f"""
=== HIGH URGENCY EMAIL REQUIRES YOUR APPROVAL ===
From: {email_data['sender']}
Subject: {email_data['subject']}
Body Preview: {email_data['body'][:300]}
Suggested Reply: {processed.get('drafted_reply', 'No draft')}
Reason for urgency: {processed.get('urgency', 'unknown')}
================================================
"""
# This is where we'd trigger the interrupt in production
# For the tutorial, we simulate approval
print(summary)
return {
**state,
"summary_report": summary
}
def resume_after_approval_node(state: EmailAgentState) -> EmailAgentState:
"""
Node: Execute the approved action after human sign-off.
"""
email_id = state.get("current_email_id")
if not email_id:
return state
processed = state.get("processed_emails", {}).get(email_id, {})
approval = state.get("approval_granted", False)
updated = dict(processed)
if approval:
# Send the drafted reply
email_data = state.get("current_email")
if email_data and updated.get("drafted_reply"):
send_email(
to_address=email_data["sender"],
subject=f"Re: {email_data['subject']}",
body=updated["drafted_reply"],
in_reply_to=email_data.get("thread_id")
)
updated["action_taken"] = "replied"
updated["human_approved"] = True
else:
updated["action_taken"] = "archived"
updated["human_approved"] = False
new_processed = dict(state.get("processed_emails", {}))
new_processed[email_id] = updated
return {
**state,
"processed_emails": new_processed,
"pending_approval": False,
"approval_granted": None
}Step 7: Assemble the LangGraph Graph
Now we wire everything together into a complete graph. This is where LangGraph shines—you define nodes and edges declaratively, and the framework handles execution, state passing, and branching:
from langgraph.graph import StateGraph, START, END
def build_email_agent_graph() -> StateGraph:
"""
Construct and compile the complete email automation agent graph.
"""
# Initialize the graph with our state type
workflow = StateGraph(EmailAgentState)
# Add all nodes
workflow.add_node("fetch_emails", fetch_emails_node)
workflow.add_node("classify_email", classify_email_node)
workflow.add_node("draft_reply", draft_reply_node)
workflow.add_node("flag_for_review", flag_for_review_node)
workflow.add_node("archive", archive_node)
workflow.add_node("handle_error", handle_error_node)
workflow.add_node("human_approval", human_approval_node)
workflow.add_node("resume_after_approval", resume_after_approval_node)
workflow.add_node("generate_summary", generate_summary_node)
# Define edges
# Start by fetching emails
workflow.add_edge(START, "fetch_emails")
# After fetching, classify the first email (or use a loop pattern)
workflow.add_edge("fetch_emails", "classify_email")
# Conditional routing after classification
workflow.add_conditional_edges(
"classify_email",
route_after_classification,
{
"draft_reply": "draft_reply",
"flag_for_review": "flag_for_review",
"archive": "archive",
"handle_error": "handle_error"
}
)
# After flagging for review, go to human approval
workflow.add_edge("flag_for_review", "human_approval")
# After human approval (interrupt point), resume
workflow.add_edge("human_approval", "resume_after_approval")
# After drafting reply, archive, or resuming — check if more emails remain
workflow.add_edge("draft_reply", "check_remaining")
workflow.add_edge("archive", "check_remaining")
workflow.add_edge("resume_after_approval", "check_remaining")
# Conditional: either loop back to classify next email or generate summary
workflow.add_conditional_edges(
"check_remaining",
check_remaining_emails,
{
"process_next": "classify_email",
"done": "generate_summary"
}
)
# Final summary then end
workflow.add_edge("generate_summary", END)
workflow.add_edge("handle_error", "check_remaining")
# Compile the graph
return workflow.compile()Step 8: Implement the Fetch and Loop Control Nodes
We need two more nodes: one to initially fetch emails and populate the queue, and another to check whether there are remaining emails to process:
def fetch_emails_node(state: EmailAgentState) -> EmailAgentState:
"""
Node: Fetch recent emails from inbox and populate the processing queue.
This is the entry point of the agent.
"""
print("Fetching recent emails...")
try:
emails = fetch_recent_emails(hours_lookback=24, max_emails=50)
except Exception as e:
return {
**state,
"errors": [f"Email fetch failed: {str(e)}"],
"new_emails": [],
"processing_queue": []
}
# Create processing queue from fetched email IDs
queue = [e["email_id"] for e in emails]
return {
**state,
"new_emails": emails,
"processing_queue": queue,
"processed_emails": {},
"errors": [],
"pending_approval": False
}
def check_remaining_emails(state: EmailAgentState) -> Literal["process_next", "done"]:
"""
Conditional function: Check if there are unprocessed emails.
Pops the next email_id from the queue and sets it as current.
"""
queue = state.get("processing_queue", [])
if not queue:
return "done"
# Pop the next email
next_id = queue[0]
remaining = queue[1:]
# Find the full email content
emails = state.get("new_emails", [])
current_email = next((e for e in emails if e["email_id"] == next_id), None)
# Update state in place for the next classification
state["processing_queue"] = remaining
state["current_email_id"] = next_id
state["current_email"] = current_email
return "process_next"
def generate_summary_node(state: EmailAgentState) -> EmailAgentState:
"""
Node: Generate a final summary report of all processed emails.
"""
processed = state.get("processed_emails", {})
errors = state.get("errors", [])
summary_parts = []
summary_parts.append(f"=== EMAIL PROCESSING SUMMARY ===")
summary_parts.append(f"Total emails processed: {len(processed)}")
high_urgency = sum(1 for p in processed.values() if p.get("urgency") == "high")
replied = sum(1 for p in processed.values() if p.get("action_taken") == "replied")
archived = sum(1 for p in processed.values() if p.get("action_taken") == "archived")
flagged = sum(1 for p in processed.values() if p.get("action_taken") == "flagged_for_review")
summary_parts.append(f"High urgency: {high_urgency}")
summary_parts.append(f"Replied: {replied}")
summary_parts.append(f"Archived: {archived}")
summary_parts.append(f"Flagged for review: {flagged}")
if errors:
summary_parts.append(f"\nErrors encountered: {len(errors)}")
for err in errors:
summary_parts.append(f" - {err}")
summary = "\n".join(summary_parts)
print(summary)
return {
**state,
"summary_report": summary
}Step 9: Add Interrupt Handling for Production
In a production system, you'd use LangGraph's built-in interrupt mechanism. Here's how to structure the interrupt inside the human_approval_node for actual use with LangGraph's interrupt() function:
from langgraph.types import interrupt
def human_approval_node_with_interrupt(state: EmailAgentState) -> EmailAgentState:
"""
Production version: Uses LangGraph's interrupt() to pause execution.
The graph will wait until resume with updated state.
"""
email_data = state.get("current_email")
if not email_data:
return state
email_id = email_data["email_id"]
processed = state.get("processed_emails", {}).get(email_id, {})
# Build approval request
approval_request = {
"email_id": email_id,
"sender": email_data["sender"],
"subject": email_data["subject"],
"body_preview": email_data["body"][:500],
"suggested_reply": processed.get("drafted_reply", ""),
"urgency": processed.get("urgency", "unknown"),
"category": processed.get("category", "unknown")
}
# This triggers the interrupt — graph execution pauses here
# The value returned by interrupt() is what the human provides when resuming
human_decision = interrupt({
"message": "High urgency email requires your approval",
"email_details": approval_request
})
# human_decision could be: {"approved": True, "custom_reply": "..."}
# or {"approved": False}
approved = human_decision.get("approved", False)
custom_reply = human_decision.get("custom_reply", None)
updated = dict(processed)
if custom_reply:
updated["drafted_reply"] = custom_reply
new_processed = dict(state.get("processed_emails", {}))
new_processed[email_id] = updated
return {
**state,
"processed_emails": new_processed,
"approval_granted": approved,
"pending_approval": False
}To resume a graph after an interrupt, you'd call:
# On the human reviewer's side:
config = {"configurable": {"thread_id": "user-session-123"}}
# Resume with approval decision
graph.update_state(
config,
{"approval_granted": True},
as_node="human_approval"
)
# Then continue execution
result = graph.invoke(None, config)Step 10: Run the Complete Agent
Here's the full execution script that ties everything together:
def run_email_agent(test_mode: bool = False):
"""
Execute the complete email automation agent.
Args:
test_mode: If True, uses mock emails instead of real IMAP fetch.
"""
# Build the graph
graph = build_email_agent_graph()
# Initial state
initial_state: EmailAgentState = {
"new_emails": [],
"processing_queue": [],
"processed_emails": {},
"current_email_id": None,
"current_email": None,
"messages": [],
"pending_approval": False,
"approval_granted": None,
"summary_report": None,
"errors": []
}
if test_mode:
# Inject mock emails for testing
initial_state["new_emails"] = [
EmailContent(
email_id="test-001",
sender="boss@company.com",
subject="URGENT: Client presentation tomorrow",
body="Hi, I need the final slides for the client presentation tomorrow at 9am. Please confirm you have them ready.",
received_at="2025-01-15 08:30:00",
thread_id=None
),
EmailContent(
email_id="test-002",
sender="newsletter@techweekly.com",
subject="This week in AI: Latest breakthroughs",
body="Here's your weekly roundup of AI news...",
received_at="2025-01-15 09:00:00",
thread_id=None
),
EmailContent(
email_id="test-003",
sender="friend@personal.com",
subject="Lunch this weekend?",
body="Hey! Are you free for lunch this Saturday?",
received_at="2025-01-15 10:00:00",
thread_id=None
)
]
initial_state["processing_queue"] = ["test-001", "test-002", "test-003"]
# Invoke the graph
config = {"configurable": {"thread_id": "email-agent-session-001"}}
try:
final_state = graph.invoke(initial_state, config)
print("\n" + "="*50)
print("AGENT RUN COMPLETE")
print("="*50)
print(final_state.get("summary_report", "No summary generated"))
# Print per-email details
for email_id, result in final_state.get("processed_emails", {}).items():
print(f"\n--- Email {email_id} ---")
print(f" Urgency: {result.get('urgency')}")
print(f" Category: {result.get('category')}")
print(f" Action: {result.get('action_taken')}")
if result.get("drafted_reply"):
print(f" Reply Draft: {result['drafted_reply'][:100]}...")
return final_state
except Exception as e:
print(f"Agent execution failed: {e}")
raise
if __name__ == "__main__":
run_email_agent(test_mode=True)Step 11: Adding Streaming and Observability
For production monitoring, LangGraph supports streaming intermediate states. Here's how to stream the agent's progress node-by-node:
def run_with_streaming():
"""
Run the agent with streaming to observe each step.
"""
graph = build_email_agent_graph()
initial_state: EmailAgentState = {
"new_emails": [],
"processing_queue": [],
"processed_emails": {},
"current_email_id": None,
"current_email": None,
"messages": [],
"pending_approval": False,
"approval_granted": None,
"summary_report": None,
"errors": []
}
# Add mock emails for demo
initial_state["new_emails"] = [
EmailContent(
email_id="demo-001",
sender="client@bigcorp.com",
subject="Contract renewal - action needed",
body="Our contract is up for renewal next month. Please send over the updated terms by EOD Friday.",
received_at="2025-01-15 14:00:00",
thread_id=None
)
]
initial_state["processing_queue"] = ["demo-001"]
config = {"configurable": {"thread_id": "streaming-demo"}}
# Stream each step
for event in graph.stream(initial_state, config, stream_mode="values"):
node_name = list(event.keys())[0]