"""
Microsoft Outlook Email Operations Module
"""
import base64
import re
import logging
from typing import Dict, Any, List, Optional
from datetime import datetime, UTC
import json
import uuid
import requests
from modules.workflows.methods.methodBase import MethodBase, action
from modules.datamodels.datamodelWorkflow import ActionResult, ActionDocument
from modules.datamodels.datamodelAi import AiCallOptions, OperationType, Priority, AiCallRequest
from modules.datamodels.datamodelChat import ChatDocument
from modules.datamodels.datamodelUam import ConnectionStatus
logger = logging.getLogger(__name__)
class MethodOutlook(MethodBase):
"""Outlook method implementation for email operations"""
def __init__(self, services):
"""Initialize the Outlook method"""
super().__init__(services)
self.name = "outlook"
self.description = "Handle Microsoft Outlook email operations"
def _format_timestamp_for_filename(self) -> str:
"""Format current timestamp as YYYYMMDD-hhmmss for filenames."""
return datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
def _getMicrosoftConnection(self, connectionReference: str) -> Optional[Dict[str, Any]]:
"""
Helper function to get Microsoft connection details.
"""
try:
logger.debug(f"Getting Microsoft connection for reference: {connectionReference}")
# Get the connection from the service
userConnection = self.services.workflow.getUserConnectionFromConnectionReference(connectionReference)
if not userConnection:
logger.error(f"Connection not found: {connectionReference}")
return None
logger.debug(f"Found connection: {userConnection.id}, status: {userConnection.status.value}, authority: {userConnection.authority.value}")
# Get a fresh token for this connection
token = self.services.utils.getFreshConnectionToken(userConnection.id)
if not token:
logger.error(f"Fresh token not found for connection: {userConnection.id}")
logger.debug(f"Connection details: {userConnection}")
return None
logger.debug(f"Fresh token retrieved for connection {userConnection.id}")
# Check if connection is active
if userConnection.status.value != "active":
logger.error(f"Connection is not active: {userConnection.id}, status: {userConnection.status.value}")
return None
return {
"id": userConnection.id,
"accessToken": token.tokenAccess,
"refreshToken": token.tokenRefresh,
"scopes": ["Mail.ReadWrite", "Mail.Send", "Mail.ReadWrite.Shared", "User.Read"] # Valid Microsoft Graph API scopes
}
except Exception as e:
logger.error(f"Error getting Microsoft connection: {str(e)}")
return None
async def _checkPermissions(self, connection: Dict[str, Any]) -> bool:
"""
Check if the current connection has the necessary permissions for Outlook operations.
"""
try:
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Test permissions by trying to access the user's mail folder
test_url = f"{graph_url}/me/mailFolders"
response = requests.get(test_url, headers=headers)
if response.status_code == 200:
return True
elif response.status_code == 403:
logger.error("Permission denied - connection lacks necessary mail permissions")
logger.error("Required scopes: Mail.ReadWrite, Mail.Send, Mail.ReadWrite.Shared")
return False
else:
logger.warning(f"Permission check returned status {response.status_code}")
return False
except Exception as e:
logger.error(f"Error checking permissions: {str(e)}")
return False
def _sanitizeSearchQuery(self, query: str) -> str:
"""
Sanitize and validate search query for Microsoft Graph API
Microsoft Graph API has specific requirements for search queries:
- Escape special characters properly
- Handle search operators correctly
- Ensure query format is valid
"""
if not query:
return ""
# Clean the query
clean_query = query.strip()
# Handle folder specifications first
if clean_query.lower().startswith('folder:'):
folder_name = clean_query[7:].strip()
if folder_name:
# Return the folder specification as-is
return clean_query
# Remove any double quotes that might cause issues
clean_query = clean_query.replace('"', '')
# Handle common search operators
if any(op in clean_query.lower() for op in ['from:', 'to:', 'subject:', 'received:', 'hasattachment:']):
# This is an advanced search query, return as-is
return clean_query
# For basic text search, ensure it's safe for contains() filter
# Remove any characters that might break the OData filter syntax
# Remove or escape characters that could break OData filter syntax
safe_query = re.sub(r'[\\\'"]', '', clean_query)
return safe_query
def _buildSearchParameters(self, query: str, folder: str, limit: int) -> Dict[str, Any]:
"""
Build search parameters for Microsoft Graph API
This method handles the complexity of building search parameters
while avoiding conflicts between $search and $filter parameters.
"""
params = {
"$top": limit
}
if not query or not query.strip():
# No query specified, just get emails from folder
if folder and folder.lower() != "all":
params["$filter"] = f"parentFolderId eq '{folder}'"
# Add orderby for basic queries
params["$orderby"] = "receivedDateTime desc"
return params
clean_query = self._sanitizeSearchQuery(query)
# Check if this is a folder specification (e.g., "folder:Drafts", "folder:Inbox")
if clean_query.lower().startswith('folder:'):
folder_name = clean_query[7:].strip() # Remove "folder:" prefix
if folder_name:
# This is a folder specification, not a text search
# Just filter by folder and return
params["$filter"] = f"parentFolderId eq '{folder_name}'"
params["$orderby"] = "receivedDateTime desc"
return params
# Check if this is a complex search query with multiple operators
if any(op in clean_query.lower() for op in ['from:', 'to:', 'subject:', 'received:', 'hasattachment:']):
# This is an advanced search query, use $search
# Microsoft Graph API supports complex search syntax
params["$search"] = f'"{clean_query}"'
# Note: When using $search, we cannot combine it with $orderby or $filter for folder
# We'll need to filter results after the API call
# Folder filtering will be done after the API call
else:
# Use $filter for basic text search, but keep it simple to avoid "InefficientFilter" error
# Microsoft Graph API has limitations on complex filters
if len(clean_query) > 50:
# If query is too long, truncate it to avoid complex filter issues
clean_query = clean_query[:50]
# Use only subject search to keep filter simple
params["$filter"] = f"contains(subject,'{clean_query}')"
# Add folder filter if specified
if folder and folder.lower() != "all":
params["$filter"] = f"{params['$filter']} and parentFolderId eq '{folder}'"
# Add orderby for basic queries
params["$orderby"] = "receivedDateTime desc"
return params
def _buildGraphFilter(self, filter_text: str) -> Dict[str, str]:
"""
Build proper Microsoft Graph API filter parameters based on filter text
Args:
filter_text (str): The filter text to process
Returns:
Dict[str, str]: Dictionary with either $filter or $search parameter
"""
if not filter_text:
return {}
filter_text = filter_text.strip()
# Handle folder specifications (e.g., "folder:Drafts", "folder:Inbox")
if filter_text.lower().startswith('folder:'):
folder_name = filter_text[7:].strip() # Remove "folder:" prefix
if folder_name:
# This is a folder specification, return empty to let the main method handle it
return {}
# Handle search queries (from:, to:, subject:, etc.) - check this FIRST
if any(filter_text.startswith(prefix) for prefix in ['from:', 'to:', 'subject:', 'received:', 'hasattachment:']):
return {"$search": f'"{filter_text}"'}
# Handle email address filters (only if it's NOT a search query)
if '@' in filter_text and '.' in filter_text and ' ' not in filter_text and not filter_text.startswith('from:'):
return {"$filter": f"from/fromAddress/address eq '{filter_text}'"}
# Handle text content - search in subject
return {"$filter": f"contains(subject,'{filter_text}')"}
def _getFolderId(self, folder_name: str, connection: Dict[str, Any]) -> Optional[str]:
"""
Get the folder ID for a given folder name
This is needed for proper filtering when using advanced search queries
"""
try:
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Get mail folders
api_url = f"{graph_url}/me/mailFolders"
response = requests.get(api_url, headers=headers)
if response.status_code == 200:
folders_data = response.json()
all_folders = folders_data.get("value", [])
# Try exact match first
for folder in all_folders:
if folder.get("displayName", "").lower() == folder_name.lower():
return folder.get("id")
# Try common variations for Drafts folder
if folder_name.lower() == "drafts":
draft_variations = ["drafts", "draft", "entwürfe", "entwurf", "brouillons", "brouillon"]
for folder in all_folders:
folder_display_name = folder.get("displayName", "").lower()
if any(variation in folder_display_name for variation in draft_variations):
return folder.get("id")
# Try common variations for other folders
if folder_name.lower() == "sent items":
sent_variations = ["sent items", "sent", "gesendete elemente", "éléments envoyés"]
for folder in all_folders:
folder_display_name = folder.get("displayName", "").lower()
if any(variation in folder_display_name for variation in sent_variations):
return folder.get("id")
logger.warning(f"Folder '{folder_name}' not found. Available folders: {[f.get('displayName', 'Unknown') for f in all_folders]}")
return None
else:
logger.warning(f"Could not retrieve folders: {response.status_code}")
return None
except Exception as e:
logger.warning(f"Error getting folder ID for '{folder_name}': {str(e)}")
return None
@action
async def readEmails(self, parameters: Dict[str, Any]) -> ActionResult:
"""
Read emails from Microsoft Outlook mailbox
USE FOR: Reading emails from Outlook, checking mailbox contents, retrieving email data
DO NOT USE FOR: Sending emails, composing emails, web research, document generation
INPUT REQUIREMENTS: Requires connectionReference (Microsoft connection)
OUTPUT FORMAT: JSON with email data and metadata
DEPENDENCIES: Requires Microsoft connection, requires internet access
WORKFLOW POSITION: Use for email analysis, before composing responses
Parameters:
connectionReference (str): Reference to the Microsoft connection (must be a connection label from AVAILABLE_CONNECTIONS_INDEX list)
folder (str, optional): Email folder to read from (default: "Inbox")
limit (int, optional): Maximum number of emails to read (default: 10)
filter (str, optional): Filter criteria for emails. Supports: Email address (e.g., "user@domain.com") - filters by sender, Search queries (e.g., "from:user@domain.com", "subject:meeting"), Text content (e.g., "project update") - searches in subject
expectedDocumentFormats (list, optional): Expected document formats with extension, mimeType, description
"""
try:
connectionReference = parameters.get("connectionReference")
folder = parameters.get("folder", "Inbox")
limit = parameters.get("limit", 10)
filter = parameters.get("filter")
expectedDocumentFormats = parameters.get("expectedDocumentFormats", [])
if not connectionReference:
return ActionResult.isFailure(error="Connection reference is required")
# Validate filter parameter if provided
if filter:
# Remove any potentially dangerous characters that could break the filter
filter = filter.strip()
if len(filter) > 100:
logger.warning(f"Filter too long ({len(filter)} chars), truncating to 100 characters")
filter = filter[:100]
# Get Microsoft connection
connection = self._getMicrosoftConnection(connectionReference)
if not connection:
return ActionResult.isFailure(error="No valid Microsoft connection found for the provided connection reference")
# Read emails using Microsoft Graph API
try:
# Microsoft Graph API endpoint for messages
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Build the API request
api_url = f"{graph_url}/me/mailFolders/{folder}/messages"
params = {
"$top": limit,
"$orderby": "receivedDateTime desc"
}
if filter:
# Build proper Graph API filter parameters
filter_params = self._buildGraphFilter(filter)
params.update(filter_params)
# If using $search, remove $orderby as they can't be combined
if "$search" in params:
params.pop("$orderby", None)
# Filter applied
# Make the API call
response = requests.get(api_url, headers=headers, params=params)
if response.status_code != 200:
logger.error(f"Graph API error: {response.status_code} - {response.text}")
logger.error(f"Request URL: {response.url}")
logger.error(f"Request headers: {headers}")
logger.error(f"Request params: {params}")
response.raise_for_status()
emails_data = response.json()
email_data = {
"emails": emails_data.get("value", []),
"count": len(emails_data.get("value", [])),
"folder": folder,
"filter": filter,
"apiResponse": emails_data
}
except ImportError:
logger.error("requests module not available")
return ActionResult.isFailure(error="requests module not available")
except requests.exceptions.HTTPError as e:
if e.response.status_code == 400:
logger.error(f"Bad Request (400) - Invalid filter or parameter: {e.response.text}")
return ActionResult.isFailure(error=f"Invalid filter syntax. Please check your filter parameter. Error: {e.response.text}")
elif e.response.status_code == 401:
logger.error("Unauthorized (401) - Access token may be expired or invalid")
return ActionResult.isFailure(error="Authentication failed. Please check your connection and try again.")
elif e.response.status_code == 403:
logger.error("Forbidden (403) - Insufficient permissions to access emails")
return ActionResult.isFailure(error="Insufficient permissions to read emails from this folder.")
else:
logger.error(f"HTTP Error {e.response.status_code}: {e.response.text}")
return ActionResult.isFailure(error=f"HTTP Error {e.response.status_code}: {e.response.text}")
except Exception as e:
logger.error(f"Error reading emails from Microsoft Graph API: {str(e)}")
return ActionResult.isFailure(error=f"Failed to read emails: {str(e)}")
# Determine output format based on expected formats
output_extension = ".json" # Default
output_mime_type = "application/json" # Default
if expectedDocumentFormats and len(expectedDocumentFormats) > 0:
# Use the first expected format
expected_format = expectedDocumentFormats[0]
output_extension = expected_format.get("extension", ".json")
output_mime_type = expected_format.get("mimeType", "application/json")
logger.info(f"Using expected format: {output_extension} ({output_mime_type})")
else:
logger.info("No expected format specified, using default .json format")
# Create result data as JSON string
result_data = {
"connectionReference": connectionReference,
"folder": folder,
"limit": limit,
"filter": filter,
"emails": email_data,
"connection": {
"id": connection["id"],
"authority": "microsoft",
"reference": connectionReference
},
"timestamp": self.services.utils.getUtcTimestamp()
}
return ActionResult.isSuccess(
documents=[ActionDocument(
documentName=f"outlook_emails_{self._format_timestamp_for_filename()}.json",
documentData=json.dumps(result_data, indent=2),
mimeType="application/json"
)]
)
except Exception as e:
logger.error(f"Error reading emails: {str(e)}")
return ActionResult.isFailure(
error=str(e)
)
@action
async def searchEmails(self, parameters: Dict[str, Any]) -> ActionResult:
"""
Search emails in Microsoft Outlook mailbox
USE FOR: Finding specific emails, searching mailbox contents, filtering email data
DO NOT USE FOR: Sending emails, composing emails, web research, document generation
INPUT REQUIREMENTS: Requires connectionReference (Microsoft connection) and query (search terms)
OUTPUT FORMAT: JSON with search results and email data
DEPENDENCIES: Requires Microsoft connection, requires internet access
WORKFLOW POSITION: Use for finding specific emails, before reading or responding
Parameters:
connectionReference (str): Reference to the Microsoft connection (must be a connection label from AVAILABLE_CONNECTIONS_INDEX list)
query (str): Search query
folder (str, optional): Folder to search in (default: "All")
limit (int, optional): Maximum number of results (default: 20)
expectedDocumentFormats (list, optional): Expected document formats with extension, mimeType, description
"""
try:
connectionReference = parameters.get("connectionReference")
query = parameters.get("query")
folder = parameters.get("folder", "All")
limit = parameters.get("limit", 20)
expectedDocumentFormats = parameters.get("expectedDocumentFormats", [])
# Validate parameters
if not connectionReference:
return ActionResult.isFailure(error="Connection reference is required")
if not query or not query.strip():
return ActionResult.isFailure(error="Search query is required and cannot be empty")
# Check if this is a folder specification query
if query.strip().lower().startswith('folder:'):
folder_name = query.strip()[7:].strip() # Remove "folder:" prefix
if not folder_name:
return ActionResult.isFailure(error="Invalid folder specification. Use format 'folder:FolderName'")
logger.info(f"Search query is a folder specification: {folder_name}")
# Validate limit
try:
limit = int(limit)
if limit <= 0 or limit > 1000: # Microsoft Graph API has limits
limit = 20
logger.warning(f"Limit {limit} is out of range, using default value 20")
except (ValueError, TypeError):
limit = 20
logger.warning(f"Invalid limit value, using default value 20")
# Get Microsoft connection
connection = self._getMicrosoftConnection(connectionReference)
if not connection:
return ActionResult.isFailure(error="No valid Microsoft connection found for the provided connection reference")
# Search emails using Microsoft Graph API
try:
# Microsoft Graph API endpoint for searching messages
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Build the search API request
api_url = f"{graph_url}/me/messages"
params = self._buildSearchParameters(query, folder, limit)
# Log search parameters for debugging
logger.debug(f"Search query: '{query}'")
logger.debug(f"Search folder: '{folder}'")
logger.debug(f"Search parameters: {params}")
logger.debug(f"API URL: {api_url}")
# Make the API call
response = requests.get(api_url, headers=headers, params=params)
# Log response details for debugging
if response.status_code != 200:
# Log detailed error information
try:
error_data = response.json()
logger.error(f"Microsoft Graph API error: {response.status_code} - {error_data}")
except:
logger.error(f"Microsoft Graph API error: {response.status_code} - {response.text}")
# Check for specific error types and provide helpful messages
if response.status_code == 400:
logger.error("Bad Request (400) - Check search query format and parameters")
logger.error(f"Search query: '{query}'")
logger.error(f"Search parameters: {params}")
logger.error(f"API URL: {api_url}")
elif response.status_code == 401:
logger.error("Unauthorized (401) - Check access token and permissions")
elif response.status_code == 403:
logger.error("Forbidden (403) - Check API permissions and scopes")
elif response.status_code == 429:
logger.error("Too Many Requests (429) - Rate limit exceeded")
raise Exception(f"Microsoft Graph API returned {response.status_code}: {response.text}")
response.raise_for_status()
search_data = response.json()
emails = search_data.get("value", [])
# Apply folder filtering if needed and we used $search
if folder and folder.lower() != "all" and "$search" in params:
# Get the actual folder ID for proper filtering
folder_id = self._getFolderId(folder, connection)
if folder_id:
# Filter results by folder ID
filtered_emails = []
for email in emails:
if email.get("parentFolderId") == folder_id:
filtered_emails.append(email)
emails = filtered_emails
logger.debug(f"Applied folder filtering: {len(filtered_emails)} emails found in folder {folder}")
else:
# Fallback: try to filter by folder name (less reliable)
filtered_emails = []
for email in emails:
# Check if email has folder information
if hasattr(email, 'parentFolderId') and email.get('parentFolderId'):
if email.get('parentFolderId') == folder:
filtered_emails.append(email)
else:
# If no folder info, include the email (less strict filtering)
filtered_emails.append(email)
emails = filtered_emails
logger.debug(f"Applied fallback folder filtering: {len(filtered_emails)} emails found in folder {folder}")
# Special handling for folder specification queries
if query.strip().lower().startswith('folder:'):
folder_name = query.strip()[7:].strip()
folder_id = self._getFolderId(folder_name, connection)
if folder_id:
# Filter results to only include emails from the specified folder
filtered_emails = []
for email in emails:
if email.get("parentFolderId") == folder_id:
filtered_emails.append(email)
emails = filtered_emails
logger.debug(f"Applied folder specification filtering: {len(filtered_emails)} emails found in folder {folder_name}")
else:
logger.warning(f"Could not find folder ID for folder specification: {folder_name}")
search_result = {
"query": query,
"results": emails,
"count": len(emails),
"folder": folder,
"limit": limit,
"apiResponse": search_data,
"searchParams": params
}
except ImportError:
logger.error("requests module not available")
return ActionResult.isFailure(error="requests module not available")
except Exception as e:
logger.error(f"Error searching emails via Microsoft Graph API: {str(e)}")
return ActionResult.isFailure(error=f"Failed to search emails: {str(e)}")
# Determine output format based on expected formats
output_extension = ".json" # Default
output_mime_type = "application/json" # Default
if expectedDocumentFormats and len(expectedDocumentFormats) > 0:
# Use the first expected format
expected_format = expectedDocumentFormats[0]
output_extension = expected_format.get("extension", ".json")
output_mime_type = expected_format.get("mimeType", "application/json")
logger.info(f"Using expected format: {output_extension} ({output_mime_type})")
else:
logger.info("No expected format specified, using default .json format")
result_data = {
"connectionReference": connectionReference,
"query": query,
"folder": folder,
"limit": limit,
"searchResults": search_result,
"connection": {
"id": connection["id"],
"authority": "microsoft",
"reference": connectionReference
},
"timestamp": self.services.utils.getUtcTimestamp()
}
return ActionResult(
success=True,
documents=[ActionDocument(
documentName=f"outlook_email_search_{self._format_timestamp_for_filename()}.json",
documentData=json.dumps(result_data, indent=2),
mimeType="application/json"
)]
)
except Exception as e:
logger.error(f"Error searching emails: {str(e)}")
return ActionResult.isFailure(error=str(e))
async def listDrafts(self, parameters: Dict[str, Any]) -> ActionResult:
"""
List email drafts in Outlook
Parameters:
connectionReference (str): Reference to the Microsoft connection (must be a connection label from AVAILABLE_CONNECTIONS_INDEX list)
folder (str, optional): Folder to search for drafts (default: "Drafts")
limit (int, optional): Maximum number of drafts to list (default: 20)
expectedDocumentFormats (list, optional): Expected document formats with extension, mimeType, description
"""
try:
connectionReference = parameters.get("connectionReference")
folder = parameters.get("folder", "Drafts")
limit = parameters.get("limit", 20)
expectedDocumentFormats = parameters.get("expectedDocumentFormats", [])
if not connectionReference:
return ActionResult.isFailure(error="Connection reference is required")
# Get Microsoft connection
connection = self._getMicrosoftConnection(connectionReference)
if not connection:
return ActionResult.isFailure(error="No valid Microsoft connection found for the provided connection reference")
# List drafts using Microsoft Graph API
try:
# Microsoft Graph API endpoint for messages
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Get the folder ID for the specified folder
folder_id = self._getFolderId(folder, connection)
if folder_id:
# List messages in the specific folder
api_url = f"{graph_url}/me/mailFolders/{folder_id}/messages"
else:
# Fallback: list all messages (might include drafts)
api_url = f"{graph_url}/me/messages"
logger.warning(f"Could not find folder '{folder}', listing all messages")
params = {
"$top": limit,
"$orderby": "lastModifiedDateTime desc",
"$select": "id,subject,from,toRecipients,ccRecipients,bccRecipients,receivedDateTime,lastModifiedDateTime,parentFolderId,isDraft"
}
# Make the API call
response = requests.get(api_url, headers=headers, params=params)
response.raise_for_status()
messages_data = response.json()
messages = messages_data.get("value", [])
# Filter for drafts if we're looking at all messages
if not folder_id:
drafts = [msg for msg in messages if msg.get("isDraft", False)]
messages = drafts
drafts_result = {
"folder": folder,
"folderId": folder_id,
"drafts": messages,
"count": len(messages),
"limit": limit,
"apiResponse": messages_data
}
except ImportError:
logger.error("requests module not available")
return ActionResult.isFailure(error="requests module not available")
except Exception as e:
logger.error(f"Error listing drafts via Microsoft Graph API: {str(e)}")
return ActionResult.isFailure(error=f"Failed to list drafts: {str(e)}")
# Determine output format based on expected formats
output_extension = ".json" # Default
output_mime_type = "application/json" # Default
if expectedDocumentFormats and len(expectedDocumentFormats) > 0:
# Use the first expected format
expected_format = expectedDocumentFormats[0]
output_extension = expected_format.get("extension", ".json")
output_mime_type = expected_format.get("mimeType", "application/json")
logger.info(f"Using expected format: {output_extension} ({output_mime_type})")
else:
logger.info("No expected format specified, using default .json format")
result_data = {
"connectionReference": connectionReference,
"folder": folder,
"limit": limit,
"draftsResult": drafts_result,
"connection": {
"id": connection["id"],
"authority": "microsoft",
"reference": connectionReference
},
"timestamp": self.services.utils.getUtcTimestamp()
}
return ActionResult(
success=True,
documents=[ActionDocument(
documentName=f"outlook_drafts_list_{self._format_timestamp_for_filename()}.json",
documentData=json.dumps(result_data, indent=2),
mimeType="application/json"
)]
)
except Exception as e:
logger.error(f"Error listing drafts: {str(e)}")
return ActionResult.isFailure(error=str(e))
async def findDrafts(self, parameters: Dict[str, Any]) -> ActionResult:
"""
Find email drafts across all folders in Outlook
Parameters:
connectionReference (str): Reference to the Microsoft connection (must be a connection label from AVAILABLE_CONNECTIONS_INDEX list)
limit (int, optional): Maximum number of drafts to find (default: 50)
expectedDocumentFormats (list, optional): Expected document formats with extension, mimeType, description
"""
try:
connectionReference = parameters.get("connectionReference")
limit = parameters.get("limit", 50)
expectedDocumentFormats = parameters.get("expectedDocumentFormats", [])
if not connectionReference:
return ActionResult.isFailure(error="Connection reference is required")
# Get Microsoft connection
connection = self._getMicrosoftConnection(connectionReference)
if not connection:
return ActionResult.isFailure(error="No valid Microsoft connection found for the provided connection reference")
# Find drafts using Microsoft Graph API
try:
# Microsoft Graph API endpoint for messages
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Get all messages and filter for drafts
api_url = f"{graph_url}/me/messages"
params = {
"$top": limit,
"$select": "id,subject,from,toRecipients,ccRecipients,bccRecipients,receivedDateTime,lastModifiedDateTime,parentFolderId,isDraft,webLink",
"$filter": "isDraft eq true"
}
# Make the API call
response = requests.get(api_url, headers=headers, params=params)
response.raise_for_status()
messages_data = response.json()
drafts = messages_data.get("value", [])
# Get folder information for each draft
for draft in drafts:
if "parentFolderId" in draft:
folder_info = self._getFolderNameById(draft["parentFolderId"], connection)
draft["folderName"] = folder_info
drafts_result = {
"totalDrafts": len(drafts),
"drafts": drafts,
"limit": limit,
"apiResponse": messages_data
}
except ImportError:
logger.error("requests module not available")
return ActionResult.isFailure(error="requests module not available")
except Exception as e:
logger.error(f"Error finding drafts via Microsoft Graph API: {str(e)}")
return ActionResult.isFailure(error=f"Failed to find drafts: {str(e)}")
# Determine output format based on expected formats
output_extension = ".json" # Default
output_mime_type = "application/json" # Default
if expectedDocumentFormats and len(expectedDocumentFormats) > 0:
# Use the first expected format
expected_format = expectedDocumentFormats[0]
output_extension = expected_format.get("extension", ".json")
output_mime_type = expected_format.get("mimeType", "application/json")
logger.info(f"Using expected format: {output_extension} ({output_mime_type})")
else:
logger.info("No expected format specified, using default .json format")
result_data = {
"connectionReference": connectionReference,
"limit": limit,
"draftsResult": drafts_result,
"connection": {
"id": connection["id"],
"authority": "microsoft",
"reference": connectionReference
},
"timestamp": self.services.utils.getUtcTimestamp()
}
return ActionResult(
success=True,
documents=[ActionDocument(
documentName=f"outlook_drafts_found_{self._format_timestamp_for_filename()}.json",
documentData=json.dumps(result_data, indent=2),
mimeType="application/json"
)]
)
except Exception as e:
logger.error(f"Error finding drafts: {str(e)}")
return ActionResult.isFailure(error=str(e))
def _getFolderNameById(self, folder_id: str, connection: Dict[str, Any]) -> str:
"""
Get folder name by folder ID
This is a helper method to identify which folder a draft is in
"""
try:
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Get folder information
api_url = f"{graph_url}/me/mailFolders/{folder_id}"
response = requests.get(api_url, headers=headers)
if response.status_code == 200:
folder_data = response.json()
return folder_data.get("displayName", f"Unknown Folder ({folder_id})")
else:
return f"Unknown Folder ({folder_id})"
except Exception as e:
logger.warning(f"Error getting folder name for ID '{folder_id}': {str(e)}")
return f"Unknown Folder ({folder_id})"
async def checkDraftsFolder(self, parameters: Dict[str, Any]) -> ActionResult:
"""
Check the contents of the Drafts folder directly
Parameters:
connectionReference (str): Reference to the Microsoft connection (must be a connection label from AVAILABLE_CONNECTIONS_INDEX list)
limit (int, optional): Maximum number of drafts to check (default: 20)
expectedDocumentFormats (list, optional): Expected document formats with extension, mimeType, description
"""
try:
connectionReference = parameters.get("connectionReference")
limit = parameters.get("limit", 20)
expectedDocumentFormats = parameters.get("expectedDocumentFormats", [])
if not connectionReference:
return ActionResult.isFailure(error="Connection reference is required")
# Get Microsoft connection
connection = self._getMicrosoftConnection(connectionReference)
if not connection:
return ActionResult.isFailure(error="No valid Microsoft connection found for the provided connection reference")
# Check Drafts folder directly
try:
# Microsoft Graph API endpoint for messages
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Get the Drafts folder ID
drafts_folder_id = self._getFolderId("Drafts", connection)
if not drafts_folder_id:
return ActionResult.isFailure(error="Could not find Drafts folder")
# Get messages directly from Drafts folder
api_url = f"{graph_url}/me/mailFolders/{drafts_folder_id}/messages"
params = {
"$top": limit,
"$select": "id,subject,from,toRecipients,ccRecipients,bccRecipients,receivedDateTime,lastModifiedDateTime,isDraft,webLink",
"$orderby": "lastModifiedDateTime desc"
}
# Make the API call
response = requests.get(api_url, headers=headers, params=params)
response.raise_for_status()
messages_data = response.json()
drafts = messages_data.get("value", [])
drafts_result = {
"draftsFolderId": drafts_folder_id,
"totalDrafts": len(drafts),
"drafts": drafts,
"limit": limit,
"apiResponse": messages_data,
"apiUrl": api_url
}
except ImportError:
logger.error("requests module not available")
return ActionResult.isFailure(error="requests module not available")
except Exception as e:
logger.error(f"Error checking Drafts folder via Microsoft Graph API: {str(e)}")
return ActionResult.isFailure(error=f"Failed to check Drafts folder: {str(e)}")
# Determine output format based on expected formats
output_extension = ".json" # Default
output_mime_type = "application/json" # Default
if expectedDocumentFormats and len(expectedDocumentFormats) > 0:
# Use the first expected format
expected_format = expectedDocumentFormats[0]
output_extension = expected_format.get("extension", ".json")
output_mime_type = expected_format.get("mimeType", "application/json")
logger.info(f"Using expected format: {output_extension} ({output_mime_type})")
else:
logger.info("No expected format specified, using default .json format")
result_data = {
"connectionReference": connectionReference,
"limit": limit,
"draftsResult": drafts_result,
"connection": {
"id": connection["id"],
"authority": "microsoft",
"reference": connectionReference
},
"timestamp": self.services.utils.getUtcTimestamp()
}
return ActionResult(
success=True,
documents=[ActionDocument(
documentName=f"outlook_drafts_folder_check_{self._format_timestamp_for_filename()}.json",
documentData=json.dumps(result_data, indent=2),
mimeType="application/json"
)]
)
except Exception as e:
logger.error(f"Error checking Drafts folder: {str(e)}")
return ActionResult.isFailure(error=str(e))
@action
async def composeAndSendEmailDirect(self, parameters: Dict[str, Any]) -> ActionResult:
"""
Compose and send email directly with subject, body, and attachments
USE FOR: When you have all email details ready and want to send directly
DO NOT USE FOR: When you need AI to generate email content from context
INPUT REQUIREMENTS: Requires connectionReference, to, subject, body parameters
OUTPUT FORMAT: Email draft created successfully with confirmation
DEPENDENCIES: Requires Microsoft connection and recipient details
WORKFLOW POSITION: Use when you have complete email information ready to send
Parameters:
connectionReference (str): REQUIRED - Reference to the Microsoft connection (must be a connection label from AVAILABLE_CONNECTIONS_INDEX list)
to (List[str]): REQUIRED - Email recipient addresses
subject (str): REQUIRED - Email subject line
body (str): REQUIRED - Email body content
cc (List[str], optional): CC recipients
bcc (List[str], optional): BCC recipients
attachmentDocumentList (List[str], optional): Document references to include as email attachments
"""
try:
connectionReference = parameters.get("connectionReference")
to = parameters.get("to")
subject = parameters.get("subject")
body = parameters.get("body")
cc = parameters.get("cc", [])
bcc = parameters.get("bcc", [])
attachmentDocumentList = parameters.get("attachmentDocumentList", [])
if not connectionReference or not to or not subject or not body:
return ActionResult.isFailure(error="connectionReference, to, subject, and body are required")
# Convert single values to lists for all recipient parameters
if isinstance(to, str):
to = [to]
if isinstance(cc, str):
cc = [cc]
if isinstance(bcc, str):
bcc = [bcc]
if isinstance(attachmentDocumentList, str):
attachmentDocumentList = [attachmentDocumentList]
# Get Microsoft connection
connection = self._getMicrosoftConnection(connectionReference)
if not connection:
return ActionResult.isFailure(error="No valid Microsoft connection found")
# Check permissions
permissions_ok = await self._checkPermissions(connection)
if not permissions_ok:
return ActionResult.isFailure(error="Connection lacks necessary permissions for Outlook operations")
# Create and send the email message
try:
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Clean and format body content
cleaned_body = body.strip()
# Check if body is already HTML
if cleaned_body.startswith('') or cleaned_body.startswith('') or '
' in cleaned_body:
html_body = cleaned_body
else:
# Convert plain text to proper HTML formatting
html_body = cleaned_body.replace('\n', '
')
html_body = f"{html_body}"
# Build the email message
message = {
"subject": subject,
"body": {
"contentType": "HTML",
"content": html_body
},
"toRecipients": [{"emailAddress": {"address": email}} for email in to],
"ccRecipients": [{"emailAddress": {"address": email}} for email in cc] if cc else [],
"bccRecipients": [{"emailAddress": {"address": email}} for email in bcc] if bcc else []
}
# Add attachments if provided
if attachmentDocumentList:
message["attachments"] = []
for attachment_ref in attachmentDocumentList:
# Get attachment document from service center
attachment_docs = self.services.workflow.getChatDocumentsFromDocumentList([attachment_ref])
if attachment_docs:
for doc in attachment_docs:
file_id = getattr(doc, 'fileId', None)
if file_id:
try:
file_content = self.services.workflow.getFileData(file_id)
if file_content:
if isinstance(file_content, bytes):
content_bytes = file_content
else:
content_bytes = str(file_content).encode('utf-8')
base64_content = base64.b64encode(content_bytes).decode('utf-8')
attachment = {
"@odata.type": "#microsoft.graph.fileAttachment",
"name": doc.fileName,
"contentType": doc.mimeType or "application/octet-stream",
"contentBytes": base64_content
}
message["attachments"].append(attachment)
except Exception as e:
logger.error(f"Error reading attachment file {doc.fileName}: {str(e)}")
# Create the draft message
drafts_folder_id = self._getFolderId("Drafts", connection)
if drafts_folder_id:
api_url = f"{graph_url}/me/mailFolders/{drafts_folder_id}/messages"
else:
api_url = f"{graph_url}/me/messages"
logger.warning("Could not find Drafts folder, creating draft in default location")
response = requests.post(api_url, headers=headers, json=message)
if response.status_code in [200, 201]:
draft_data = response.json()
draft_id = draft_data.get("id", "Unknown")
result_data = {
"status": "success",
"message": "Email draft created successfully",
"draftId": draft_id,
"folder": "Drafts (Entwürfe)",
"mailbox": connection.get('userEmail', 'Unknown'),
"subject": subject,
"recipients": to,
"cc": cc,
"bcc": bcc,
"attachments": len(attachmentDocumentList),
"timestamp": self.services.utils.getUtcTimestamp()
}
return ActionResult(
success=True,
documents=[ActionDocument(
documentName=f"email_draft_created_{self._format_timestamp_for_filename()}.json",
documentData=json.dumps(result_data, indent=2),
mimeType="application/json"
)]
)
else:
logger.error(f"Failed to create draft. Status: {response.status_code}, Response: {response.text}")
return ActionResult.isFailure(error=f"Failed to create email draft: {response.status_code} - {response.text}")
except Exception as e:
logger.error(f"Error creating email via Microsoft Graph API: {str(e)}")
return ActionResult.isFailure(error=f"Failed to create email: {str(e)}")
except Exception as e:
logger.error(f"Error in composeAndSendEmailDirect: {str(e)}")
return ActionResult.isFailure(error=str(e))
@action
async def composeAndSendEmailWithContext(self, parameters: Dict[str, Any]) -> ActionResult:
"""
Compose and send email using AI to generate subject and body from context and documents
USE FOR: When you have context and documents but need AI to compose the email content
DO NOT USE FOR: When you already have complete email details ready
INPUT REQUIREMENTS: Requires connectionReference, to, context, and optional documentList
OUTPUT FORMAT: Email draft created successfully with AI-generated content
DEPENDENCIES: Requires Microsoft connection, AI service, and context/documents
WORKFLOW POSITION: Use when you need AI to generate email content from available information
Parameters:
connectionReference (str): REQUIRED - Reference to the Microsoft connection (must be a connection label from AVAILABLE_CONNECTIONS_INDEX list)
to (List[str]): REQUIRED - Email recipient addresses
context (str): REQUIRED - Context information for email composition
documentList (List[str], optional): Document references to include as context and attachments
cc (List[str], optional): CC recipients
bcc (List[str], optional): BCC recipients
emailStyle (str, optional): Email style (formal, casual, business) - default: "business"
maxLength (int, optional): Maximum length for generated content - default: 1000
"""
try:
connectionReference = parameters.get("connectionReference")
to = parameters.get("to")
context = parameters.get("context")
documentList = parameters.get("documentList", [])
cc = parameters.get("cc", [])
bcc = parameters.get("bcc", [])
emailStyle = parameters.get("emailStyle", "business")
maxLength = parameters.get("maxLength", 1000)
if not connectionReference or not to or not context:
return ActionResult.isFailure(error="connectionReference, to, and context are required")
# Convert single values to lists for all recipient parameters
if isinstance(to, str):
to = [to]
if isinstance(cc, str):
cc = [cc]
if isinstance(bcc, str):
bcc = [bcc]
if isinstance(documentList, str):
documentList = [documentList]
# Get Microsoft connection
connection = self._getMicrosoftConnection(connectionReference)
if not connection:
return ActionResult.isFailure(error="No valid Microsoft connection found")
# Check permissions
permissions_ok = await self._checkPermissions(connection)
if not permissions_ok:
return ActionResult.isFailure(error="Connection lacks necessary permissions for Outlook operations")
# Prepare documents for AI processing
chatDocuments = []
if documentList:
chatDocuments = self.services.workflow.getChatDocumentsFromDocumentList(documentList)
# Create AI prompt for email composition
ai_prompt = f"""
Compose a professional email based on the following context and requirements:
CONTEXT:
{context}
RECIPIENT: {to}
EMAIL STYLE: {emailStyle}
MAX LENGTH: {maxLength} characters
Please generate:
1. A clear, professional subject line
2. A well-structured email body that addresses the context appropriately
3. Use the {emailStyle} tone throughout
Return your response in the following JSON format:
{{
"subject": "Your generated subject line here",
"body": "Your generated email body here (can include HTML formatting like
for line breaks)"
}}
Make sure the email is:
- Professional and appropriate for the context
- Clear and concise
- Well-structured with proper greeting and closing
- Relevant to the provided context
"""
# Call AI service to generate email content
try:
ai_response = await self.services.ai.callAi(
prompt=ai_prompt,
documents=chatDocuments,
options=AiCallOptions(
operationType="email_composition",
priority="normal",
compressPrompt=False,
compressContext=True,
processDocumentsIndividually=True,
processingMode="detailed",
resultFormat="json",
maxCost=0.50,
maxProcessingTime=30
)
)
# Parse AI response
try:
ai_content = ai_response
# Extract JSON from AI response
if "```json" in ai_content:
json_start = ai_content.find("```json") + 7
json_end = ai_content.find("```", json_start)
json_content = ai_content[json_start:json_end].strip()
elif "{" in ai_content and "}" in ai_content:
json_start = ai_content.find("{")
json_end = ai_content.rfind("}") + 1
json_content = ai_content[json_start:json_end]
else:
json_content = ai_content
email_data = json.loads(json_content)
subject = email_data.get("subject", "")
body = email_data.get("body", "")
if not subject or not body:
return ActionResult.isFailure(error="AI did not generate valid subject and body")
except json.JSONDecodeError as e:
logger.error(f"Failed to parse AI response as JSON: {str(e)}")
logger.error(f"AI response content: {ai_response}")
return ActionResult.isFailure(error="AI response was not valid JSON format")
except Exception as e:
logger.error(f"Error calling AI service: {str(e)}")
return ActionResult.isFailure(error=f"Failed to generate email content: {str(e)}")
# Now create the email with AI-generated content
try:
graph_url = "https://graph.microsoft.com/v1.0"
headers = {
"Authorization": f"Bearer {connection['accessToken']}",
"Content-Type": "application/json"
}
# Clean and format body content
cleaned_body = body.strip()
# Check if body is already HTML
if cleaned_body.startswith('') or cleaned_body.startswith('') or '
' in cleaned_body:
html_body = cleaned_body
else:
# Convert plain text to proper HTML formatting
html_body = cleaned_body.replace('\n', '
')
html_body = f"{html_body}"
# Build the email message
message = {
"subject": subject,
"body": {
"contentType": "HTML",
"content": html_body
},
"toRecipients": [{"emailAddress": {"address": email}} for email in to],
"ccRecipients": [{"emailAddress": {"address": email}} for email in cc] if cc else [],
"bccRecipients": [{"emailAddress": {"address": email}} for email in bcc] if bcc else []
}
# Add documents as attachments if provided
if documentList:
message["attachments"] = []
for attachment_ref in documentList:
# Get attachment document from service center
attachment_docs = self.services.workflow.getChatDocumentsFromDocumentList([attachment_ref])
if attachment_docs:
for doc in attachment_docs:
file_id = getattr(doc, 'fileId', None)
if file_id:
try:
file_content = self.services.workflow.getFileData(file_id)
if file_content:
if isinstance(file_content, bytes):
content_bytes = file_content
else:
content_bytes = str(file_content).encode('utf-8')
base64_content = base64.b64encode(content_bytes).decode('utf-8')
attachment = {
"@odata.type": "#microsoft.graph.fileAttachment",
"name": doc.fileName,
"contentType": doc.mimeType or "application/octet-stream",
"contentBytes": base64_content
}
message["attachments"].append(attachment)
except Exception as e:
logger.error(f"Error reading attachment file {doc.fileName}: {str(e)}")
# Create the draft message
drafts_folder_id = self._getFolderId("Drafts", connection)
if drafts_folder_id:
api_url = f"{graph_url}/me/mailFolders/{drafts_folder_id}/messages"
else:
api_url = f"{graph_url}/me/messages"
logger.warning("Could not find Drafts folder, creating draft in default location")
response = requests.post(api_url, headers=headers, json=message)
if response.status_code in [200, 201]:
draft_data = response.json()
draft_id = draft_data.get("id", "Unknown")
result_data = {
"status": "success",
"message": "Email draft created successfully with AI-generated content",
"draftId": draft_id,
"folder": "Drafts (Entwürfe)",
"mailbox": connection.get('userEmail', 'Unknown'),
"subject": subject,
"body": body,
"recipients": to,
"cc": cc,
"bcc": bcc,
"attachments": len(documentList),
"aiGenerated": True,
"context": context,
"emailStyle": emailStyle,
"timestamp": self.services.utils.getUtcTimestamp()
}
return ActionResult(
success=True,
documents=[ActionDocument(
documentName=f"ai_generated_email_draft_{self._format_timestamp_for_filename()}.json",
documentData=json.dumps(result_data, indent=2),
mimeType="application/json"
)]
)
else:
logger.error(f"Failed to create draft. Status: {response.status_code}, Response: {response.text}")
return ActionResult.isFailure(error=f"Failed to create email draft: {response.status_code} - {response.text}")
except Exception as e:
logger.error(f"Error creating email via Microsoft Graph API: {str(e)}")
return ActionResult.isFailure(error=f"Failed to create email: {str(e)}")
except Exception as e:
logger.error(f"Error in composeAndSendEmailWithContext: {str(e)}")
return ActionResult.isFailure(error=str(e))
async def checkPermissions(self, parameters: Dict[str, Any]) -> ActionResult:
"""
Check if the current Microsoft connection has the necessary permissions for Outlook operations.
Parameters:
connectionReference (str): Reference to the Microsoft connection (must be a connection label from AVAILABLE_CONNECTIONS_INDEX list) to check
"""
try:
connectionReference = parameters.get("connectionReference")
if not connectionReference:
return ActionResult.isFailure(error="Connection reference is required")
# Get Microsoft connection
connection = self._getMicrosoftConnection(connectionReference)
if not connection:
return ActionResult.isFailure(error="Failed to get Microsoft connection")
# Check permissions
permissions_ok = await self._checkPermissions(connection)
if permissions_ok:
result_data = {
"permissions": "✅ All necessary permissions are available",
"scopes": connection.get("scopes", []),
"connectionId": connection.get("id"),
"status": "ready"
}
return ActionResult(
success=True,
documents=[ActionDocument(
documentName=f"outlook_permissions_check_{self._format_timestamp_for_filename()}.json",
documentData=json.dumps(result_data, indent=2),
mimeType="application/json"
)]
)
else:
result_data = {
"permissions": "❌ Missing necessary permissions",
"requiredScopes": ["Mail.ReadWrite", "Mail.Send", "Mail.ReadWrite.Shared", "User.Read"],
"currentScopes": connection.get("scopes", []),
"connectionId": connection.get("id"),
"status": "needs_reauthentication",
"message": "Please re-authenticate your Microsoft connection to get updated permissions."
}
return ActionResult(
success=False,
documents=[ActionDocument(
documentName=f"outlook_permissions_check_{self._format_timestamp_for_filename()}.json",
documentData=json.dumps(result_data, indent=2),
mimeType="application/json"
)],
error="Connection lacks necessary permissions for Outlook operations"
)
except Exception as e:
logger.error(f"Error checking permissions: {str(e)}")
return ActionResult.isFailure(error=str(e))