Enhanced MCP Server

""" Intelligent Email Agent Automatically composes and sends emails from natural language requests """ import re from datetime import datetime from typing import Dict, Any, Optional, List import structlog from ..adapters.gmail_adapter import GmailAdapter from ..utils.llm_manager import LLMManager logger = structlog.get_logger(__name__) class EmailAgent: """ AI-powered email composition agent Understands natural language like: - "Send email to john@example.com thanking him for the interview" - "Write follow-up email to recruiter@company.com about my application" - "Compose professional email to manager@company.com requesting time off" """ def __init__(self, gmail_adapter: GmailAdapter, llm_manager: LLMManager): self.gmail = gmail_adapter self.llm = llm_manager logger.info("Email Agent initialized") async def compose_and_send( self, request: str, to: Optional[str] = None, subject: Optional[str] = None, additional_context: Optional[str] = None ) -> Dict[str, Any]: """ Compose and send email from natural language request Args: request: Natural language email request to: Recipient email (optional, can be extracted from request) subject: Email subject (optional, will be generated if not provided) additional_context: Additional context for email composition Returns: Result with email details and send status """ logger.info(f"Processing email request: {request}") try: # Step 1: Extract email details using LLM email_details = await self._extract_email_details( request, to, subject, additional_context ) if not email_details.get("success"): return { "success": False, "error": "Could not understand email request", "suggestion": "Please specify recipient, purpose, and key points" } # Step 2: Validate recipient email recipient = email_details.get("to") if not recipient or not self._is_valid_email(recipient): return { "success": False, "error": f"Invalid or missing recipient email: {recipient}", "suggestion": "Please provide a valid email address" } # Step 3: Compose email body using LLM email_body = await self._compose_email_body( purpose=email_details.get("purpose"), key_points=email_details.get("key_points", []), tone=email_details.get("tone", "professional"), recipient_name=email_details.get("recipient_name"), additional_context=additional_context ) if not email_body: return { "success": False, "error": "Failed to compose email body" } # Step 4: Generate subject if not provided email_subject = email_details.get("subject") if not email_subject: email_subject = await self._generate_subject( email_details.get("purpose"), email_body[:200] ) # Step 5: Send email result = self.gmail.send_email( to=recipient, subject=email_subject, body=email_body, cc=email_details.get("cc"), html=False ) if result.get("success"): logger.info( f"Email sent successfully", to=recipient, subject=email_subject ) return { "success": True, "message": f"✅ Email sent to {recipient}", "message_id": result.get("message_id"), "to": recipient, "subject": email_subject, "body": email_body, "preview": email_body[:200] + "..." if len(email_body) > 200 else email_body } else: return { "success": False, "error": result.get("error", "Failed to send email") } except Exception as e: logger.error(f"Email composition failed: {e}") return { "success": False, "error": str(e) } async def draft_email( self, request: str, to: Optional[str] = None, subject: Optional[str] = None, additional_context: Optional[str] = None ) -> Dict[str, Any]: """ Draft an email without sending (preview only) Args: request: Natural language email request to: Recipient email (optional) subject: Email subject (optional) additional_context: Additional context Returns: Email draft for review """ logger.info(f"Drafting email: {request}") try: # Extract email details email_details = await self._extract_email_details( request, to, subject, additional_context ) if not email_details.get("success"): return { "success": False, "error": "Could not understand email request" } # Compose email body email_body = await self._compose_email_body( purpose=email_details.get("purpose"), key_points=email_details.get("key_points", []), tone=email_details.get("tone", "professional"), recipient_name=email_details.get("recipient_name"), additional_context=additional_context ) # Generate subject if needed email_subject = email_details.get("subject") if not email_subject: email_subject = await self._generate_subject( email_details.get("purpose"), email_body[:200] ) return { "success": True, "draft": { "to": email_details.get("to"), "subject": email_subject, "body": email_body, "cc": email_details.get("cc"), "tone": email_details.get("tone"), "purpose": email_details.get("purpose") }, "message": "✅ Email draft ready for review" } except Exception as e: logger.error(f"Email drafting failed: {e}") return { "success": False, "error": str(e) } async def _extract_email_details( self, request: str, to: Optional[str] = None, subject: Optional[str] = None, additional_context: Optional[str] = None ) -> Dict[str, Any]: """ Use LLM to extract email details from natural language Args: request: Natural language request to: Explicit recipient (optional) subject: Explicit subject (optional) additional_context: Additional context Returns: Extracted email details """ prompt = f""" Extract email composition details from this request: "{request}" Additional context: {additional_context or "None"} Return a JSON object with: - to: Recipient email address (extract from request or use provided) - recipient_name: Recipient's name (if mentioned) - subject: Email subject (if clear from context, otherwise null) - purpose: Brief description of email purpose (e.g., "thank interviewer", "follow up on application", "request meeting") - key_points: Array of key points to include in email - tone: Email tone (professional, casual, formal, friendly) - cc: CC recipients if mentioned (otherwise null) Provided recipient: {to or "not specified"} Provided subject: {subject or "not specified"} Examples: Request: "Send email to john@example.com thanking him for the interview yesterday" Output: {{ "to": "john@example.com", "recipient_name": "John", "subject": null, "purpose": "thank interviewer", "key_points": ["thank for time", "appreciate opportunity", "enjoyed conversation"], "tone": "professional", "cc": null }} Request: "Write follow-up email to recruiter@company.com about my software engineer application" Output: {{ "to": "recruiter@company.com", "recipient_name": null, "subject": null, "purpose": "follow up on application", "key_points": ["checking application status", "reaffirm interest", "available for interview"], "tone": "professional", "cc": null }} Request: "Compose email to manager@company.com requesting time off next week" Output: {{ "to": "manager@company.com", "recipient_name": null, "subject": "Time Off Request", "purpose": "request time off", "key_points": ["request time off next week", "provide coverage plan", "available for questions"], "tone": "professional", "cc": null }} Now extract from: "{request}" Return ONLY the JSON object, nothing else. """ try: response = await self.llm.generate(prompt) # Parse JSON from response import json content = response.content.strip() # Handle markdown code blocks if "```json" in content: content = content.split("```json")[1].split("```")[0].strip() elif "```" in content: content = content.split("```")[1].split("```")[0].strip() details = json.loads(content) details["success"] = True # Override with explicit values if provided if to: details["to"] = to if subject: details["subject"] = subject return details except Exception as e: logger.error(f"Failed to extract email details: {e}") return {"success": False} async def _compose_email_body( self, purpose: str, key_points: List[str], tone: str = "professional", recipient_name: Optional[str] = None, additional_context: Optional[str] = None ) -> str: """ Compose email body using LLM Args: purpose: Email purpose key_points: Key points to include tone: Email tone recipient_name: Recipient's name additional_context: Additional context Returns: Composed email body """ greeting = f"Dear {recipient_name}," if recipient_name else "Hello," key_points_text = "\n".join([f"- {point}" for point in key_points]) prompt = f""" Compose a {tone} email with the following details: Purpose: {purpose} Key points to include: {key_points_text} Additional context: {additional_context or "None"} Guidelines: - Start with appropriate greeting: {greeting} - Keep it concise (3-4 paragraphs max) - Use {tone} tone - Include all key points naturally - End with appropriate closing - Sign off professionally Return ONLY the email body, no subject line, no additional text. """ try: response = await self.llm.generate(prompt) email_body = response.content.strip() # Remove any markdown formatting if "```" in email_body: email_body = email_body.split("```")[1].split("```")[0].strip() return email_body except Exception as e: logger.error(f"Failed to compose email body: {e}") return "" async def _generate_subject(self, purpose: str, body_preview: str) -> str: """ Generate email subject line Args: purpose: Email purpose body_preview: Preview of email body Returns: Generated subject line """ prompt = f""" Generate a concise, professional email subject line for: Purpose: {purpose} Email preview: {body_preview} Return ONLY the subject line (5-10 words), no quotes, no explanation. """ try: response = await self.llm.generate(prompt) subject = response.content.strip() # Remove quotes if present subject = subject.strip('"').strip("'") # Limit length if len(subject) > 100: subject = subject[:97] + "..." return subject except Exception as e: logger.error(f"Failed to generate subject: {e}") return "Follow-up" def _is_valid_email(self, email: str) -> bool: """ Validate email address format Args: email: Email address to validate Returns: True if valid, False otherwise """ pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return bool(re.match(pattern, email))