""" Email Agent Module. Handles email-related tasks using Microsoft Graph API. """ import logging import json from typing import Dict, Any, List, Optional from modules.workflow.agentBase import AgentBase logger = logging.getLogger(__name__) class AgentEmail(AgentBase): """Agent for handling email-related tasks.""" def __init__(self): """Initialize the email agent.""" super().__init__() self.name = "email" self.label = "Email Agent" self.description = "Handles email composition and sending using Microsoft Graph API" self.capabilities = [ "email_composition", "email_draft_creation", "email_template_generation" ] self.serviceBase = None def setDependencies(self, serviceBase=None): """Set external dependencies for the agent.""" self.serviceBase = serviceBase async def processTask(self, task: Dict[str, Any]) -> Dict[str, Any]: """ Process an email-related task. Args: task: Task object containing: - prompt: Instructions for the agent - inputDocuments: List of documents to process - outputSpecifications: List of required output documents - context: Additional context including workflow info Returns: Dictionary containing: - feedback: Text response explaining what was done - documents: List of created documents """ try: # Extract task information prompt = task.get("prompt", "") inputDocuments = task.get("inputDocuments", []) outputSpecs = task.get("outputSpecifications", []) # Check AI service if not self.service.base: return { "feedback": "The Email agent requires an AI service to function.", "documents": [] } # Check if Microsoft connector is available if not hasattr(self.service, 'msft'): return { "feedback": "Microsoft connector not available. Please ensure Microsoft integration is properly configured.", "documents": [] } # Get Microsoft token token_data = self.service.msft.getMsftToken() if not token_data: # Create authentication trigger document auth_doc = self._createFrontendAuthTriggerDocument() return { "feedback": "Microsoft authentication required. Please authenticate to continue.", "documents": [auth_doc] } # Extract document data from input documentContents, attachments = self._processInputDocuments(inputDocuments) # Generate email subject and body using AI emailTemplate = await self._generateEmailTemplate(prompt, documentContents) # Create HTML preview of the email htmlPreview = self._createHtmlPreview(emailTemplate) # Attempt to create a draft email using Microsoft Graph API draft_result = self.service.msft.createDraftEmail( emailTemplate["recipient"], emailTemplate["subject"], emailTemplate["htmlBody"], attachments ) # Prepare output documents documents = [] # Process output specifications for spec in outputSpecs: label = spec.get("label", "") description = spec.get("description", "") if label.endswith(".html"): # Create the HTML template file templateDoc = self.formatAgentDocumentOutput( label, emailTemplate["htmlBody"], # Use the actual HTML body, not the preview "text/html" ) documents.append(templateDoc) elif label.endswith(".json"): # Create JSON template if requested templateJson = json.dumps(emailTemplate, indent=2) templateDoc = self.formatAgentDocumentOutput( label, templateJson, "application/json" ) documents.append(templateDoc) else: # Default to preview for other cases previewDoc = self.formatAgentDocumentOutput( label, htmlPreview, "text/html" ) documents.append(previewDoc) # Prepare feedback message if draft_result: feedback = f"Email draft created successfully for {emailTemplate.get('recipient')}. The subject is: '{emailTemplate['subject']}'" if attachments: feedback += f" with {len(attachments)} attachment(s)" feedback += ". You can open and edit it in your Outlook draft folder." else: feedback = "Email template created but could not save as draft. HTML preview and template are available as documents." return { "feedback": feedback, "documents": documents } except Exception as e: logger.error(f"Error in email agent: {str(e)}") return { "feedback": f"Error processing email task: {str(e)}", "documents": [] } def _createFrontendAuthTriggerDocument(self) -> Dict[str, Any]: """Create a document that triggers Microsoft authentication in the frontend.""" return { "name": "microsoft_auth", "ext": "html", "mimeType": "text/html", "data": """
Please click the button below to authenticate with Microsoft:
This email is regarding your request: {prompt}
" } except Exception as e: logger.warning(f"Error generating email template: {str(e)}") return { "recipient": "recipient@example.com", "subject": "Information Regarding Your Request", "plainBody": f"This email is regarding your request: {prompt}", "htmlBody": f"This email is regarding your request: {prompt}
" } def _createHtmlPreview(self, emailTemplate: Dict[str, Any]) -> str: """ Create an HTML preview of the email template. Args: emailTemplate: Email template dictionary Returns: HTML string for preview """ html = f"""No content
')}