"""
LangGraph Email + Calendar Agent Workflow (Refactored & Fixed)
---------------------------------------------------------------
Fetch unread Gmail messages externally,
then process each through the LangGraph agent to decide between
'reply' and 'calendar_event' actions.
Enhancements:
✅ Fixed router output (no more "unknown channel" warnings)
✅ Improved Calendar ReAct agent (no generic responses)
✅ Smarter HR-style classification prompt
✅ Cleaner logging and structure
"""
import os
import imaplib
import email
import smtplib
from email.header import decode_header
from email.message import EmailMessage
from typing import Literal, Dict, List
from datetime import datetime
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
# LangChain / LangGraph imports
from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.types import Command
from langgraph.prebuilt import create_react_agent
from langchain_google_community import CalendarToolkit
# ---------------- CONFIG ----------------
load_dotenv()
SCOPES = ["https://www.googleapis.com/auth/calendar"]
EMAIL_ACCOUNT = os.getenv("EMAIL_USER")
EMAIL_PASSWORD = os.getenv("EMAIL_PASS")
IMAP_SERVER = "imap.gmail.com"
SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 465
# ---------------- GOOGLE CALENDAR SETUP ----------------
flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
credentials = flow.run_local_server(port=0)
calendar_api_resource = build("calendar", "v3", credentials=credentials)
calendar_toolkit = CalendarToolkit(api_resource=calendar_api_resource)
# ---------------- ROUTER SCHEMA ----------------
class RouterSchema(BaseModel):
reasoning: str = Field(description="Reasoning behind classification.")
classification: Literal["reply", "Calendar"] = Field(
description="Whether the email needs a reply or calendar action."
)
# ---------------- STATE SCHEMA ----------------
class State(MessagesState):
email_input: Dict
job_details: Dict | None = None
classification_decision: Literal["reply", "Calendar"] | None = None
calendar_output: Dict | None = None
send: bool = False
class StateInput(TypedDict):
email_input: Dict
# ---------------- LLM SETUP ----------------
llm = init_chat_model("openai:o4-mini")
llm_router = llm.with_structured_output(RouterSchema)
# ---------------- FETCH EMAILS ----------------
def fetch_unread_emails() -> List[Dict]:
"""Fetch unread Gmail messages externally."""
mails = []
mail = imaplib.IMAP4_SSL(IMAP_SERVER)
mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
mail.select("inbox")
status, messages = mail.search(None, "(UNSEEN)")
if status != "OK":
print("❌ No new emails found.")
return []
for num in messages[0].split():
_, data = mail.fetch(num, "(RFC822)")
msg = email.message_from_bytes(data[0][1])
subject, encoding = decode_header(msg.get("Subject"))[0]
if isinstance(subject, bytes):
subject = subject.decode(encoding or "utf-8")
sender = msg.get("From")
body = ""
if msg.is_multipart():
for part in msg.walk():
ctype = part.get_content_type()
disp = str(part.get("Content-Disposition"))
if ctype == "text/plain" and "attachment" not in disp:
try:
body = part.get_payload(decode=True).decode("utf-8", errors="ignore")
except Exception:
body = str(part.get_payload())
break
else:
try:
body = msg.get_payload(decode=True).decode("utf-8", errors="ignore")
except Exception:
body = str(msg.get_payload())
mails.append({
"email_recipient": sender,
"subject": subject,
"body": body,
"job_details": None
})
mail.logout()
print(f"✅ Fetched {len(mails)} unread emails.")
return mails
# ---------------- STATE UPDATER ----------------
def update_state(state: State) -> Dict:
"""Initialize or update state with fetched email."""
email_item = state.get("email_input", {})
print("🧭 State updated with new email and job details placeholder.")
return {"email_input": email_item, "job_details": None}
# ---------------- ROUTER ----------------
def router(state: State) -> Command[Literal["send_email", "calendar_event", "__end__"]]:
"""Route email to either reply or calendar path."""
email_item = state.get("email_input", {})
if not email_item:
print("[Router] No email input, ending workflow.")
return Command(goto="__end__", update={"messages": [{"role": "system", "content": "No unread emails"}]})
email_recipient = email_item["email_recipient"]
subject = email_item["subject"]
body = email_item["body"]
triage_instructions = """
You are an intelligent email triage agent. Your task is to classify each email into one of two categories:
1. CALENDAR — Emails that require scheduling, rescheduling, or confirming meetings/events.
2. RESPOND — Emails that require a direct written reply, but no calendar action.
CALENDAR Emails:
These are messages where the user needs to take an action in Google Calendar or similar scheduling tools.
They typically involve:
- Interview requests or confirmations
- Meeting scheduling, rescheduling, or cancellations
- Event invitations (e.g., conference calls, client demos, HR interviews)
- Requests to find a suitable time for a meeting
- Availability checks (e.g., "Are you free on Tuesday?")
- Calendar invites that need acceptance or confirmation
- Time or date coordination for appointments (doctor, personal, HR, etc.)
- “Let’s meet”, “Schedule a call”, or “Book a slot” style emails
Do NOT classify as CALENDAR if:
- The email only mentions dates in passing but doesn’t require a meeting.
- The action can be completed via a written reply only.
---
✉️ RESPOND Emails
These are messages that require a thoughtful or informative written response.
They usually involve:
- Direct questions that need clarification or explanation
- Project updates where acknowledgment or feedback is expected
- HR or client communication that doesn’t include scheduling (e.g., document requests, approvals)
- Status checks (“Can you send the report?”)
- Information sharing (“Please confirm receipt”, “Can you review this?”)
- Personal responses (“Thanks for your help”, “Congratulations”, etc.)
- Apologies, follow-ups, or issue resolutions that don’t involve time coordination
- Any general correspondence that requires replying but not scheduling
---
our output must be a single classification string: Either "calendar" or "reply"
"""
system_prompt = """
< Role >
Your role is to triage incoming emails based upon instructs and background information below.
</ Role >
< Instructions >
Categorize each email into one of three categories:
1. CALENDAR - Emails that require scheduling or calendar actions
2. RESPOND - Emails that need a direct response
Classify the below email into one of these categories.
</ Instructions >
< Rules >
{triage_instructions}
</ Rules >
""".format(triage_instructions = triage_instructions)
user_prompt = """
Please determine how to handle the below email thread:
From: {email_recipient}
Subject: {subject}
Body: {body}
""".format(email_recipient = email_recipient, subject = subject, body = body[:2000])
result = llm_router.invoke([
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
])
classification = result.classification.lower()
print(f"[Router] Classification result: {classification}")
if classification == "reply":
print("📧 Classified as 'reply'")
return Command(
goto="send_email",
update={
"classification_decision": result.classification,
# "messages": [{"role": "user", "content": f"Respond to: {subject}"}],
"email_input": email_item,
},
)
elif classification == "calendar":
print("📅 Classified as 'Calendar'")
return Command(
goto="calendar_event",
update={
"classification_decision": result.classification,
# "messages": [{"role": "user", "content": f"Handle calendar event from: {subject}"}],
"email_input": email_item,
},
)
else:
print(f"[Router] Unknown classification '{classification}', ending workflow.")
return Command(goto="__end__", update={"messages": [{"role": "system", "content": "Unknown classification"}]})
# ---------------- UTIL ----------------
def clean_header(value: str) -> str:
if not value:
return ""
return value.replace("\r", "").replace("\n", "").strip()
# ---------------- EMAIL NODE ----------------
def send_email(state: State) -> Dict:
"""Generate a professional HR response."""
email_item = state.get("email_input", {})
body = email_item.get("body", "")
subject = email_item.get("subject", "")
recipient = email_item.get("email_recipient", "")
response = llm.invoke([
{"role": "system", "content": "You are a professional HR assistant replying to candidate queries. Be polite, formal, and helpful."},
{"role": "user", "content": f"Email body:\n{body}"}
])
reply_text = response.content
msg = EmailMessage()
msg["From"] = EMAIL_ACCOUNT
msg["To"] = recipient
msg["Subject"] = clean_header(f"Re: {subject}")
msg.set_content(reply_text)
with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server:
server.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
server.send_message(msg)
print(f"✅ Sent HR reply to {recipient}")
return {"send": True}
# ---------------- CALENDAR NODE ----------------
calendar_agent = create_react_agent(
model=llm,
tools=calendar_toolkit.get_tools(),
name="calendar_agent",
prompt = f"""
You are an expert Google Calendar ReAct agent built for HR scheduling. You have access to tools including:
- GetCalendarsInfo()
- CalendarSearchEvents(calendar_id, time_min, time_max, query)
- CalendarCreateEvent(summary, start_datetime, end_datetime, timezone, attendees, description, reminders, conference=False, calendar_id=None)
- CalendarUpdateEvent(event_id, fields...)
- (Other calendar tools as provided)
OBJECTIVE
- Autonomously create or update interview/meeting events based on the provided email subject/body.
- Resolve conflicts automatically and choose reasonable defaults when information is missing.
- Return a compact, valid JSON result (see Output Schema) plus a 1–3 line human summary to be used in confirmation emails.
- Do NOT ask the user for follow-up — infer defaults and proceed.
CRITICAL RULES
- Always call GetCalendarsInfo first to identify available calendars and select the best calendar_id (prefer "primary" or a calendar owned by the organizer).
- Normalize all times to Asia/Kolkata and return timestamps in "YYYY-MM-DD HH:MM:SS" format.
- Never output internal chain-of-thought or detailed reasoning. You may perform step-by-step reasoning **internally** but DO NOT expose it in the output. Only return the structured result and concise summary.
- If the email explicitly requests a video call (e.g., "Google Meet", "Zoom", "video call"), set conference=True; otherwise default to conference=False.
- Defaults:
- duration = 1 hour (unless a duration is explicitly stated)
- timezone = "Asia/Kolkata"
- calendar = primary (unless GetCalendarsInfo suggests better one)
- reminders = default reminders
- Business hours preference: Mon–Fri 09:00–17:30 IST unless email indicates otherwise.
- Conflict search window: ±15 minutes for immediate conflict checks; when searching for alternatives, look up to 14 days ahead, advancing in 30-minute increments within business hours. If email marks "urgent", expand search beyond business hours.
INTERNAL PROCESS (follow this plan internally; DO NOT print these steps):
1. Parse the email subject/body for: candidate/person name, role/title, organizer, explicit date/time windows, timezone hints, duration, attendee email addresses, and whether a video conference is requested.
2. If explicit slot(s) exist, normalize them to Asia/Kolkata. If none, propose the default: next business day at 10:00 AM IST.
3. Call GetCalendarsInfo() to list calendars and choose calendar_id.
4. For a chosen start_time, call CalendarSearchEvents(calendar_id=<id>, time_min=start_time, time_max=end_time) to detect conflicts.
5. If conflict found, search forward for the nearest free slot within 14 days (increment 30 min) during business hours. If still unavailable and the email is urgent, expand to evenings/weekends.
6. With a free slot determined, call CalendarCreateEvent(...) with:
- summary (use pattern: "Interview: )
- start_datetime, end_datetime (end = start + duration)
- timezone = "Asia/Kolkata"
- attendees = list of parsed emails (include organizer if provided)
- description = body (strip signatures)
- reminders = default
- conference = True only if video-call indicated
7. If updating an existing event is more appropriate (email references existing invite or event ID), use CalendarUpdateEvent.
8. After event creation/update, produce the structured JSON result and a short human_summary suitable for an email confirmation. Include alternative_slots (if conflict resolution chose alternatives) so the email agent can optionally mention them.
TOOL USAGE EXAMPLES (do these exact conceptual calls; adapt to your tool signatures):
- GetCalendarsInfo()
- CalendarSearchEvents(calendar_id="<id>", time_min="2025-10-10 09:00:00", time_max="2025-10-10 10:00:00")
- CalendarCreateEvent(summary="Interview: Jane Doe — Backend Engineer", start_datetime="2025-10-10 10:00:00", end_datetime="2025-10-10 11:00:00", timezone="Asia/Kolkata", attendees=["jane@example.com","recruiter@example.com"], description="...", reminders=True, conference=False)
OUTPUT SCHEMA (return this EXACT JSON object only; must be valid JSON):
{{
"action": "create" | "update" | "no_action", # create if a new event was made
"calendar_id": "<selected_calendar_id>",
"event": {{
"summary": "<event title>",
"start_datetime": "YYYY-MM-DD HH:MM:SS",
"end_datetime": "YYYY-MM-DD HH:MM:SS",
"timezone": "Asia/Kolkata",
"attendees": ["a@b.com"],
"description": "<detailed description>",
"reminders": true,
"conference": false
}},
"conflict_handled": true | false,
"alternative_slots": ["YYYY-MM-DD HH:MM:SS", "..."], # optional; empty list if not applicable
"human_summary": "<2-3 line summary for candidate/email>",
"notify": true # set true if the calling system should send confirmation emails
}}
INPUT AVAILABLE (the agent will receive):
- subject: string
- body: string
- from_email: string
- current_time: "{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
EDGE CASES & SMALL RULES
- If the email references an existing event by event id or link, prefer updating it (action="update") rather than creating a duplicate.
- Remove signatures and quoted text before using the body for event description.
- If attendees cannot be determined (no emails found), still create the event on the chosen calendar and set notify=True so the email agent can send an invite to the sender.
- Keep the human_summary short, actionable, and polite (example: "Interview scheduled — Interview with Jane Doe (Backend Engineer) on 2025-10-10 10:00–11:00 IST. An invite has been sent.").
FINAL NOTE
- Perform detailed internal reasoning and tool calls, but only emit the structured JSON and the human_summary above. Do NOT output internal chain-of-thought or step-by-step deliberation.
"""
)
def extract_datetime_from_text(text: str) -> str | None:
"""Extract natural-language date/time and convert to datetime string."""
from dateparser import parse
dt = parse(text, settings={'TIMEZONE': 'Asia/Kolkata', 'RETURN_AS_TIMEZONE_AWARE': False})
if not dt:
return None
return dt.strftime("%Y-%m-%d %H:%M:%S")
def calendar_event(state: State) -> Dict:
"""Handle scheduling events using the Calendar agent."""
email_item = state.get("email_input", {})
subject = email_item.get("subject", "")
body = email_item.get("body", "")
sender = email_item.get("email_recipient", "")
response = calendar_agent.invoke(
{"role": "user", "content": f"Schedule or manage event:\nFrom: {sender}\nSubject: {subject}\nBody:\n{body}"}
)
print("📅 Calendar agent executed:", response)
return {"calendar_output": str(response)}
# ---------------- BUILD GRAPH ----------------
graph = StateGraph(State)
graph.add_node("update_state", update_state)
graph.add_node("router", router)
graph.add_node("send_email", send_email)
graph.add_node("calendar_event", calendar_event)
graph.add_edge(START, "update_state")
graph.add_edge("update_state", "router")
graph.add_conditional_edges(
"router",
{
"send_email": send_email,
"calendar_event": calendar_event,
}
)
graph.add_edge("send_email", END)
graph.add_edge("calendar_event", END)
agent_executor = graph.compile()
# ---------------- MAIN ----------------
def main():
"""Fetch unread emails and process sequentially."""
all_emails = fetch_unread_emails()
if not all_emails:
print("✅ No unread emails to process.")
return
for i, email_item in enumerate(all_emails, start=1):
print(f"\n📨 Processing email {i}/{len(all_emails)}: {email_item['subject']}")
output = agent_executor.invoke({"email_input": email_item})
print(f"✅ Completed: {email_item['subject']}\nResult →", output)
if __name__ == "__main__":
print("🚀 Starting LangGraph Agent Workflow (HR MODE)...")
main()
Task router with path (‘__pregel_pull’, ‘router’) ….