varunk3249/gmail-inbox-agent
1
1import gradio as gr2import os3from multi_tool_agent.gmail_agent_logic import (4 get_gmail_service,5 search_emails,6 summarize_email_with_gemini,7 generate_reply_with_gemini,8 send_reply,9 get_total_unread_count,10 get_emails_received_today_count,11 list_recent_emails,12 get_auth_url,13 exchange_code_for_credentials,14 store_user_credentials,15 create_session_id,16 is_user_authenticated,17 create_oauth_flow18)19import google.generativeai as genai20from dotenv import load_dotenv21import json22 23load_dotenv()24 25# --- Deployment: Write credentials from environment variables to files ---26if 'GMAIL_CREDENTIALS_JSON' in os.environ:27 with open('credentials.json', 'w') as f:28 f.write(os.environ.get('GMAIL_CREDENTIALS_JSON'))29 print("Created credentials.json from environment variable.")30 31if 'GMAIL_TOKEN_JSON' in os.environ:32 with open('token.json', 'w') as f:33 f.write(os.environ.get('GMAIL_TOKEN_JSON'))34 print("Created token.json from environment variable.")35# --- End Deployment ---36 37try:38 gemini_api_key = os.environ.get("GOOGLE_API_KEY")39 if not gemini_api_key:40 raise ValueError("GOOGLE_API_KEY not found in environment variables. Please set it in the .env file and restart.")41 genai.configure(api_key=gemini_api_key)42 gemini_model = genai.GenerativeModel('gemini-2.5-flash-lite')43 print("Gemini model initialized successfully for Gradio app.")44except Exception as e:45 raise Exception(f"FATAL: Error initializing Gemini model: {e}")46 47# --- LLM Prompt for Intent Recognition ---48CONTROLLER_PROMPT_TEMPLATE = """49You are the controller for a Gmail assistant. Analyze the user's message and determine the primary intent and necessary parameters based on the conversation history.50 51Available intents and their required parameters:52- LIST_RECENT: requires optional 'count' (integer, default 5) to list the most recent emails in the inbox.53- SEARCH: requires 'query' (e.g., "from:a@b.com subject:hello")54- SUMMARIZE_BY_ID: requires 'email_id'55- SUMMARIZE_LAST: requires context indicating a specific email (e.g., from a previous search or mention). Check context['last_email_details']['id'].56- GENERATE_REPLY: requires 'reply_instructions' (what the user wants to say) and context from a previously summarized email (context['last_email_details'] required).57- SEND_REPLY: requires confirmation (e.g., "yes", "send it") and context from a previously generated reply draft (context['last_reply_draft'] and context['last_email_details'] required).58- GET_UNREAD_COUNT: No parameters required.59- GET_TODAY_EMAIL_COUNT: No parameters required.60- GREETING/OTHER: if the intent is unclear, a simple greeting, or doesn't match the capabilities.61 62Conversation History:63{history_string}64 65Current User message: "{user_message}"66 67Current Context (JSON):68{context_json}69 70Based ONLY on the **Current User message** and the **Current Context**, determine the single most likely intent and extract the parameters.71 72Output your decision STRICTLY as a JSON object with 'intent' (string) and 'parameters' (dictionary) keys. If parameters are not applicable or derivable, use an empty dictionary {{}}.73Example for "list my last 3 emails": {{"intent": "LIST_RECENT", "parameters": {{"count": 3}}}}74Example for "search for emails from test@test.com": {{"intent": "SEARCH", "parameters": {{"query": "from:test@test.com"}}}}75Example for "summarize email with id 123": {{"intent": "SUMMARIZE_BY_ID", "parameters": {{"email_id": "123"}}}}76Example for "draft a reply saying thanks": {{"intent": "GENERATE_REPLY", "parameters": {{"reply_instructions": "saying thanks"}}}}77Example for "yes send it": {{"intent": "SEND_REPLY", "parameters": {{}}}}78Example for "how many unread emails do I have": {{"intent": "GET_UNREAD_COUNT", "parameters": {{}}}}79Example for "how many emails today": {{"intent": "GET_TODAY_EMAIL_COUNT", "parameters": {{}}}}80Example for "hello there": {{"intent": "GREETING/OTHER", "parameters": {{}}}}81 82JSON Response:83"""84 85def authenticate_user(auth_code):86 """Handle user authentication with OAuth code."""87 if not auth_code or not auth_code.strip():88 return None, "Please enter the authorization code."89 90 try:91 flow = create_oauth_flow()92 auth_url, _ = flow.authorization_url(prompt='consent')93 94 # Exchange code for credentials95 credentials = exchange_code_for_credentials(auth_code.strip(), flow)96 if credentials:97 # Create session and store credentials98 session_id = create_session_id()99 if store_user_credentials(session_id, credentials):100 return session_id, "Authentication successful! You can now use Gmail features."101 else:102 return None, "Failed to store credentials. Please try again."103 else:104 return None, "Invalid authorization code. Please try again."105 except Exception as e:106 return None, f"Authentication error: {e}"107 108# --- Chatbot Logic ---109def handle_chat(message, history, session_state):110 """111 Processes user message using an LLM controller, interacts with Gmail/Gemini tools.112 """113 # Get session info114 session_id = session_state.get("session_id") if session_state else None115 conversation_context = session_state.get("conversation_context", {116 "last_email_summary": None,117 "last_email_details": {},118 "last_reply_draft": None,119 }) if session_state else {120 "last_email_summary": None,121 "last_email_details": {},122 "last_reply_draft": None,123 }124 125 # Check authentication126 if not session_id or not is_user_authenticated(session_id):127 return "🔐 Please authenticate with Gmail first using the Authentication tab above."128 129 # Basic checks130 if not gemini_model:131 return "Error: Gemini model is not available. Check API key and configuration."132 133 # --- 1. Call LLM Controller ---134 history_string = ""135 if history:136 for h in history:137 history_string += f"User: {h[0]}\nAssistant: {h[1]}\n"138 context_json = json.dumps(conversation_context, indent=2)139 prompt = CONTROLLER_PROMPT_TEMPLATE.format(140 history_string=history_string,141 user_message=message,142 context_json=context_json143 )144 145 try:146 print(f"--- Sending Controller Prompt ---\n{prompt}\n------------------------------")147 controller_response = gemini_model.generate_content(prompt)148 print(f"--- Controller Response ---\n{controller_response.text}\n--------------------------- ")149 150 cleaned_response_text = controller_response.text.strip().replace('```json', '').replace('```', '')151 decision = json.loads(cleaned_response_text)152 intent = decision.get("intent")153 parameters = decision.get("parameters", {})154 155 except json.JSONDecodeError as e:156 print(f"Error decoding controller JSON: {e}\nResponse was: {controller_response.text}")157 return "Sorry, I had trouble understanding that request (JSON Decode Error)."158 except Exception as e:159 print(f"Error during controller LLM call: {e}")160 return f"Sorry, an error occurred while processing your request: {e}"161 162 # --- 2. Execute Action based on Intent ---163 try:164 if intent == "LIST_RECENT":165 count = parameters.get("count", 5) # Default to 5 if not specified166 try:167 count = int(count)168 except ValueError:169 count = 5170 171 list_result = list_recent_emails(user_id='me', max_results=count, session_id=session_id)172 if list_result["status"] == "success" and list_result["emails"]:173 email_strings = []174 for email in list_result["emails"]:175 email_str = (176 f"Subject: {email.get('subject', 'N/A')}\n\n" # Double newline177 f"From: {email.get('from', 'N/A')}\n\n" # Double newline178 f"Date: {email.get('date', 'N/A')}"179 )180 email_strings.append(email_str)181 response_text = f"Here are your last {len(email_strings)} emails:\n\n" + "\n\n---\n\n".join(email_strings)182 conversation_context["last_email_details"] = list_result["emails"][0] # Store first found183 conversation_context["last_reply_draft"] = None # Clear any old draft184 elif list_result["status"] == "success":185 response_text = "No emails found in your inbox."186 else:187 response_text = f"Error listing recent emails: {list_result.get('error_message', 'Unknown error')}"188 189 elif intent == "SEARCH":190 query = parameters.get("query")191 if not query:192 response_text = "My controller understood you want to search, but didn't find search criteria. Please specify (e.g., 'from:...' or 'subject:...')."193 else:194 search_result = search_emails(query=query, user_id='me', session_id=session_id)195 if search_result["status"] == "success" and search_result["emails"]:196 # Format emails197 email_strings = []198 for email in search_result["emails"]:199 email_str = (200 f"Subject: {email.get('subject', 'N/A')}\n\n" # Double newline201 f"From: {email.get('from', 'N/A')}\n\n" # Double newline202 f"Date: {email.get('date', 'N/A')}"203 )204 email_strings.append(email_str)205 206 response_text = "Found emails:\n\n" + "\n\n---\n\n".join(email_strings)207 208 # Store the first result's ID for potential follow-up209 conversation_context["last_email_details"] = search_result["emails"][0] # Store first found210 conversation_context["last_reply_draft"] = None # Clear any old draft211 elif search_result["status"] == "success":212 response_text = "No emails found matching your query."213 else:214 response_text = f"Error searching emails: {search_result.get('error_message', 'Unknown error')}"215 216 elif intent == "SUMMARIZE_BY_ID":217 email_id = parameters.get("email_id")218 if email_id:219 summary_result = summarize_email_with_gemini(user_id='me', email_id=email_id, session_id=session_id)220 if summary_result["status"] == "success":221 response_text = f"Summary:\n{summary_result['summary']}"222 conversation_context["last_email_summary"] = summary_result['summary']223 conversation_context["last_email_details"] = summary_result # Store all details224 conversation_context["last_reply_draft"] = None # Clear any old draft225 else:226 response_text = f"Error summarizing email {email_id}: {summary_result.get('error_message', 'Unknown error')}"227 else:228 response_text = "My controller understood you want to summarize by ID, but didn't find an ID. Please provide it."229 230 elif intent == "SUMMARIZE_LAST":231 email_id = conversation_context["last_email_details"].get("id")232 if email_id:233 summary_result = summarize_email_with_gemini(user_id='me', email_id=email_id, session_id=session_id)234 if summary_result["status"] == "success":235 response_text = f"Summary of the last mentioned email (ID: {email_id}):\n{summary_result['summary']}"236 conversation_context["last_email_summary"] = summary_result['summary']237 conversation_context["last_email_details"] = summary_result # Store all details238 conversation_context["last_reply_draft"] = None # Clear any old draft239 else:240 response_text = f"Error summarizing email {email_id}: {summary_result.get('error_message', 'Unknown error')}"241 else:242 response_text = "I don't have a 'last email' in context to summarize. Please search for or specify an email first."243 244 elif intent == "GENERATE_REPLY":245 instructions = parameters.get("reply_instructions", "")246 details = conversation_context.get("last_email_details", {})247 original_body = details.get("original_body")248 249 if original_body:250 # Combine original body with user instructions for the prompt251 generation_prompt_body = f"User wants reply to address: '{instructions}'\n\nOriginal Email Body:\n{original_body}"252 253 reply_result = generate_reply_with_gemini(254 original_subject=details.get("subject", "No Subject"),255 original_body=generation_prompt_body256 )257 if reply_result["status"] == "success":258 response_text = f"Draft Reply:\n------\n{reply_result['reply_body']}\n------\n\nWould you like me to send this reply?"259 conversation_context["last_reply_draft"] = reply_result['reply_body'] # Store draft260 else:261 response_text = f"Error generating reply draft: {reply_result.get('error_message', 'Unknown error')}"262 else:263 response_text = "I need the context of an email (specifically its body) to generate a reply. Please summarize an email first."264 265 elif intent == "SEND_REPLY":266 details = conversation_context.get("last_email_details", {})267 draft = conversation_context.get("last_reply_draft")268 269 if (draft and details.get("sender_email") and details.get("subject") and270 details.get("thread_id") and details.get("original_message_id")):271 272 send_result = send_reply(273 user_id='me',274 to=details["sender_email"],275 sender='me',276 subject=details["subject"],277 reply_body=draft,278 thread_id=details["thread_id"],279 original_message_id=details["original_message_id"],280 references=details.get("references", ""),281 session_id=session_id282 )283 if send_result["status"] == "success":284 response_text = f"Reply sent successfully! Message ID: {send_result['message_id']}"285 conversation_context["last_reply_draft"] = None # Clear draft after sending286 # Optionally clear last_email_details too?287 else:288 response_text = f"Error sending reply: {send_result.get('error_message', 'Unknown error')}"289 elif not draft:290 response_text = "There is no reply draft stored in context to send. Please generate one first."291 else:292 response_text = "I'm missing some details from the original email context (like sender, thread ID, or message ID) needed to send the reply. Please summarize the relevant email again."293 294 elif intent == "GET_UNREAD_COUNT":295 unread_result = get_total_unread_count(user_id='me', session_id=session_id)296 if unread_result["status"] == "success":297 response_text = f"You have {unread_result['unread_count']} unread emails in your inbox."298 else:299 response_text = f"Error getting unread count: {unread_result.get('error_message', 'Unknown error')}"300 301 elif intent == "GET_TODAY_EMAIL_COUNT":302 today_count_result = get_emails_received_today_count(user_id='me', session_id=session_id)303 if today_count_result["status"] == "success":304 response_text = f"You received approximately {today_count_result['today_count']} emails in the last 24 hours."305 else:306 response_text = f"Error counting today's emails: {today_count_result.get('error_message', 'Unknown error')}"307 308 elif intent == "GREETING/OTHER":309 response_text = "Hello! How can I help you with your Gmail today?"310 311 else: # Handles cases where intent is missing or unrecognized by the Python code312 response_text = f"Sorry, I received an unexpected intent ('{intent}') from the controller. I don't know how to handle that."313 314 except Exception as e:315 print(f"Error executing action for intent {intent}: {e}") # Log unexpected errors316 response_text = f"An unexpected error occurred while executing the action: {e}"317 318 # Update session state319 if session_state:320 session_state["conversation_context"] = conversation_context321 322 return response_text323 324if __name__ == "__main__":325 with gr.Blocks(326 theme=gr.themes.Soft(327 font=[gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"],328 ),329 title="Gmail AI Agent"330 ) as app:331 # Session state332 session_state = gr.State({})333 334 gr.Markdown(335 """336<div style="text-align: center;">337 <h1>📧 Gmail AI Agent</h1>338 <p>Your personal assistant for managing your Gmail. Each user can authenticate with their own Gmail account.</p>339</div>340"""341 )342 343 with gr.Tabs():344 with gr.TabItem("🔐 Authentication"):345 gr.Markdown(346 """347### Step 1: Get Authorization Code348Click the button below to get an authorization URL, then follow these steps:349 3501. **Click "Get Authorization URL"** below3512. **Copy the URL** that appears3523. **Open the URL** in your browser3534. **Sign in** to your Gmail account3545. **Allow permissions** for the app3556. **Copy the authorization code** from the browser3567. **Paste the code** in the text box below and click "Authenticate"357"""358 )359 360 get_url_btn = gr.Button("🔗 Get Authorization URL", variant="primary")361 auth_url_display = gr.Textbox(362 label="Authorization URL",363 placeholder="Click 'Get Authorization URL' to generate the URL",364 interactive=False,365 lines=3366 )367 368 auth_code_input = gr.Textbox(369 label="Authorization Code",370 placeholder="Paste the authorization code here",371 lines=2372 )373 374 auth_btn = gr.Button("🔑 Authenticate", variant="primary")375 auth_status = gr.Textbox(376 label="Authentication Status",377 interactive=False,378 lines=2379 )380 381 def get_url():382 try:383 auth_url, _ = get_auth_url()384 return auth_url385 except Exception as e:386 return f"Error generating URL: {e}"387 388 def handle_auth(auth_code, session_state):389 session_id, message = authenticate_user(auth_code)390 if session_id:391 session_state["session_id"] = session_id392 session_state["conversation_context"] = {393 "last_email_summary": None,394 "last_email_details": {},395 "last_reply_draft": None,396 }397 return message, session_state, "" # Clear auth code on success398 else:399 return message, session_state, auth_code # Keep auth code on failure400 401 get_url_btn.click(402 fn=get_url,403 outputs=auth_url_display404 )405 406 auth_btn.click(407 fn=handle_auth,408 inputs=[auth_code_input, session_state],409 outputs=[auth_status, session_state, auth_code_input]410 )411 412 with gr.TabItem("💬 Chat"):413 gr.Markdown(414 """415**Here's what I can do once you're authenticated:**416- **List emails**: e.g., 'Show my last 5 emails'417- **Search emails**: e.g., 'Find emails from boss@company.com about the project report'418- **Summarize emails**: e.g., 'Summarize the last email we discussed?'419- **Draft & Send replies**: e.g., 'Draft a reply saying I will look into it', then 'Ok send it'420- **Get counts**: e.g., 'How many unread emails do I have?'421"""422 )423 chatbot = gr.Chatbot(424 [],425 elem_id="chatbot",426 placeholder="Authenticate first, then start chatting about your Gmail!",427 height=600,428 type="tuples",429 )430 431 def chat_fn(message, history, session_state):432 response = handle_chat(message, history, session_state)433 return response434 435 gr.ChatInterface(436 fn=chat_fn,437 chatbot=chatbot,438 additional_inputs=[session_state],439 examples=[440 ["How many unread emails do I have?"],441 ["Show my last 5 emails"],442 ["Find emails from my manager"],443 ["How many emails did I get today?"],444 ],445 title=None,446 description=None,447 type="tuples",448 )449 450 app.launch()