mindroot.coreplugins.email package

Submodules

mindroot.coreplugins.email.email_provider module

class mindroot.coreplugins.email.email_provider.EmailProvider(config: Dict)[source]

Bases: object

async check_emails(folder: str = 'INBOX', criteria: Dict = None, batch_size: int = None, max_messages: int = None, start_id: str = None) → Dict[source]

Check for emails using IMAP

async mark_as_processed(message_ids: List[str], folder: str = 'INBOX') → Dict[source]

Mark messages as processed using IMAP

async send_email(to: str, subject: str, body: str, reply_to_message: Dict = None, headers: Dict = None) → Dict[source]

Send an email using SMTP

mindroot.coreplugins.email.imap_handler module

class mindroot.coreplugins.email.imap_handler.IMAPHandler(config: Dict)[source]

Bases: object

build_search_criteria(criteria: Dict = None, last_uid: str = None) → str[source]

Build IMAP search criteria string

async check_emails(folder: str = 'INBOX', criteria: Dict = None, batch_size: int = None, max_messages: int = None, start_id: str = None) → Dict[source]

Check for emails in specified folder with pagination using UIDs

connect() → IMAP4[source]

Establish IMAP connection

decode_text(text_bytes: bytes) → str[source]

Safely decode text bytes to string, trying multiple encodings

get_email_body(email_message) → str[source]

Extract email body with better encoding handling

async mark_as_processed(message_ids: List[str], folder: str = 'INBOX') → Dict[source]

Mark messages as processed/read using UIDs

mindroot.coreplugins.email.mod module

async mindroot.coreplugins.email.mod.check_emails(folder: str = 'INBOX', criteria: Dict = None, context=None) → Dict[source]

Service to check for new emails

async mindroot.coreplugins.email.mod.init_email_provider(config: Dict = None, context=None) → bool[source]

Initialize the email provider with config

async mindroot.coreplugins.email.mod.send_email(to: str, subject: str, body: str, context=None) → Dict[source]

Service to send an email

Parameters:

  • to – Recipient email address

  • subject – Email subject

  • body – Email body (can be plain text or HTML - auto-detected)

  • context – Optional context

Returns:

Dict with success status and error info

mindroot.coreplugins.email.services module

async mindroot.coreplugins.email.services.check_emails(config: Dict, folder: str = 'INBOX', criteria: Dict = None, batch_size: int = None, max_messages: int = None, start_id: str = None) → Dict[source]

Service to check for new emails with pagination support

async mindroot.coreplugins.email.services.get_provider() → EmailProvider[source]

Get the provider instance, initializing with defaults if needed

async mindroot.coreplugins.email.services.init_provider(config: Dict) → None[source]

Initialize the email provider with config

async mindroot.coreplugins.email.services.mark_as_processed(config: Dict, message_ids: List[str], folder: str = 'INBOX') → Dict[source]

Service to mark emails as processed

async mindroot.coreplugins.email.services.send_email_helper(config: Dict, to: str, subject: str, body: str, reply_to_message: Dict = None, headers: Dict = None) → Dict[source]

Helper function to send an email (not a service)

mindroot.coreplugins.email.smtp_handler module

class mindroot.coreplugins.email.smtp_handler.SMTPHandler(config: Dict)[source]

Bases: object

async connect() → SMTP[source]

Establish SMTP connection

async send_email(to: str, subject: str, body: str, reply_to_message: Dict = None, headers: Dict = None) → Dict[source]

Send an email

mindroot.coreplugins.email.test_batch module

mindroot.coreplugins.email.test_email module

mindroot.coreplugins.email.test_email_service module

Test script to verify email service functionality. Run this to test if email sending works with your configuration.

async mindroot.coreplugins.email.test_email_service.test_email_service()[source]

Test the email service with current environment configuration

Module contents