IMAP email in Python

Python services receive email through IMAP connections defined in the Dashboard - messages can be fetched, marked as seen and deleted, and a connection can poll its mailbox on a schedule, invoking your service with each message it finds.

The same code works with classic IMAP servers and with Microsoft 365 connections that use OAuth2 - the connection holds the credentials and your Python code stays the same, with no knowledge of Microsoft 365 or OAuth2 required.

Create a connection

To create a connection, go to Connections > Outgoing > E-mail > IMAP in the Dashboard, click Create a new IMAP connection and fill in the form:

1. Name: My Connection
2. User / e-mail: the mailbox to read from
3. Click Toggle options in the Generic IMAP row and fill in Host, e.g. imap.example.com
4. Click OK

Receive email

The connection's get method yields each message that matches the connection's get criteria, e.g. UNSEEN:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class MyService(Service):

    def handle(self):
        conn = self.email.imap.get('My Connection').conn

        for msg_id, msg in conn.get():

            # Access the message
            self.logger.info(msg.data)

Mark messages as seen

Call .mark_seen() on a message object and the server marks it as seen:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class MyService(Service):

    def handle(self):
        conn = self.email.imap.get('My Connection').conn

        for msg_id, msg in conn.get():

            # Access the message
            self.logger.info(msg.data)

            # To mark the message seen
            msg.mark_seen()

Delete messages

Call .delete() on a message object and the server marks it as deleted:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class MyService(Service):

    def handle(self):
        conn = self.email.imap.get('My Connection').conn

        for msg_id, msg in conn.get():

            # Access the message
            self.logger.info(msg.data)

            # To delete the message
            msg.delete()

Read email on a schedule

Instead of polling a mailbox in your own code, an IMAP connection can do it for you. Expand the Scheduler section when you create or edit a connection in the Dashboard and fill in these fields:

• Run every - how often to poll the mailbox, e.g. every 5 minutes
• Start date - when to start polling
• Service - the service to invoke for each message received
• Invoke with - what the service receives on input, either Message or Each attachment

Saving the connection auto-creates a scheduler job for it. On each run, the job fetches all the messages matching the connection's get criteria, e.g. UNSEEN, and invokes your service with each of them, depending on the invoke-with mode.

After a message is processed, it is acknowledged automatically - if your service raises no exception, the message is marked as seen. If the service raises an exception, the message is left untouched, so it still matches the UNSEEN criteria and your service receives it anew on the next run. Messages are never lost when processing fails and there is no need to call msg.mark_seen() yourself.

Invoke with messages

In the Message mode, which is the default one, your service receives the message object on input - the same kind of object that conn.get() yields - so it can read the message's contents:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class ProcessNewEmail(Service):

    def handle(self):

        # The IMAP message that we were invoked with
        msg = self.request.payload

        # Access the message
        self.logger.info('Subject: %s', msg.data.subject)
        self.logger.info('From: %s', msg.data.sent_from)
        self.logger.info('Body: %s', msg.data.body['plain'])

Invoke with attachments

In the Each attachment mode, the message's body is ignored and your service is invoked once for each attachment that the message contains. A message without attachments is acknowledged without invoking anything.

The input is an IMAPAttachment object with these attributes:

• filename - the attachment's file name
• content_type - its MIME type, e.g. application/pdf
• size - the size of the attachment's data, in bytes
• data - the attachment's contents, as bytes
• content_id - the attachment's Content-ID header or None if there was none
• message - the message object that the attachment was extracted from
• msg_uid - the UID of that message
• subject - the subject of that message
• sent_from - who sent that message

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class ProcessNewAttachment(Service):

    def handle(self):

        # The IMAP attachment that we were invoked with
        attachment = self.request.payload

        # Access the attachment
        self.logger.info('File name: %s', attachment.filename)
        self.logger.info('Content type: %s', attachment.content_type)
        self.logger.info('Size: %s', attachment.size)
        self.logger.info('From message: %s', attachment.subject)

        # The contents is available as bytes
        data = attachment.data

If any of the per-attachment invocations raises an exception, the whole message is left untouched and all of its attachments are received anew on the next run.

See also

Learn more