Skip to content

email_channel

email_channel

EmailChannel — SMTP/IMAP email adapter (stdlib only, zero extra deps).

Classes

EmailChannel 

EmailChannel(
    smtp_host: str = "",
    smtp_port: int = 587,
    *,
    imap_host: str = "",
    imap_port: int = 993,
    username: str = "",
    password: str = "",
    use_tls: bool = True,
    bus: Optional[EventBus] = None,
)

Bases: BaseChannel

Email channel adapter using stdlib smtplib and imaplib.

PARAMETER DESCRIPTION
smtp_host

SMTP server hostname.

TYPE: str DEFAULT: ''

smtp_port

SMTP server port (default 587 for STARTTLS).

TYPE: int DEFAULT: 587

imap_host

IMAP server hostname for incoming messages.

TYPE: str DEFAULT: ''

imap_port

IMAP server port (default 993 for SSL).

TYPE: int DEFAULT: 993

username

Email username. Falls back to EMAIL_USERNAME env var.

TYPE: str DEFAULT: ''

password

Email password. Falls back to EMAIL_PASSWORD env var.

TYPE: str DEFAULT: ''

use_tls

Whether to use TLS (default True).

TYPE: bool DEFAULT: True

bus

Optional event bus for publishing channel events.

TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/channels/email_channel.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(
    self,
    smtp_host: str = "",
    smtp_port: int = 587,
    *,
    imap_host: str = "",
    imap_port: int = 993,
    username: str = "",
    password: str = "",
    use_tls: bool = True,
    bus: Optional[EventBus] = None,
) -> None:
    self._smtp_host = smtp_host
    self._smtp_port = smtp_port
    self._imap_host = imap_host
    self._imap_port = imap_port
    self._username = username or os.environ.get("EMAIL_USERNAME", "")
    self._password = password or os.environ.get("EMAIL_PASSWORD", "")
    self._use_tls = use_tls
    self._bus = bus
    self._handlers: List[ChannelHandler] = []
    self._status = ChannelStatus.DISCONNECTED
    self._listener_thread: Optional[threading.Thread] = None
    self._stop_event = threading.Event()
Methods:
connect 
connect() -> None

Start IMAP polling for incoming messages (if configured).

Source code in src/diapason/channels/email_channel.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def connect(self) -> None:
    """Start IMAP polling for incoming messages (if configured)."""
    if not self._smtp_host or not self._username:
        logger.warning("Email channel not fully configured")
        self._status = ChannelStatus.ERROR
        return

    self._stop_event.clear()
    self._status = ChannelStatus.CONNECTING

    if self._imap_host:
        self._listener_thread = threading.Thread(
            target=self._imap_poll_loop,
            daemon=True,
        )
        self._listener_thread.start()

    self._status = ChannelStatus.CONNECTED
    logger.info("Email channel connected (SMTP: %s)", self._smtp_host)
disconnect 
disconnect() -> None

Stop the IMAP listener thread.

Source code in src/diapason/channels/email_channel.py 
 97
 98
 99
100
101
102
103
def disconnect(self) -> None:
    """Stop the IMAP listener thread."""
    self._stop_event.set()
    if self._listener_thread is not None:
        self._listener_thread.join(timeout=5.0)
        self._listener_thread = None
    self._status = ChannelStatus.DISCONNECTED
send 
send(
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool

Send an email via SMTP. channel is the recipient address.

Source code in src/diapason/channels/email_channel.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def send(
    self,
    channel: str,
    content: str,
    *,
    conversation_id: str = "",
    metadata: Dict[str, Any] | None = None,
) -> bool:
    """Send an email via SMTP. ``channel`` is the recipient address."""
    if not self._smtp_host or not self._username:
        logger.warning("Cannot send: email not configured")
        return False

    try:
        msg = MIMEText(content)
        msg["From"] = self._username
        msg["To"] = channel
        msg["Subject"] = (metadata or {}).get(
            "subject",
            "Message from Diapason",
        )
        if conversation_id:
            msg["In-Reply-To"] = conversation_id

        if self._use_tls:
            with smtplib.SMTP(self._smtp_host, self._smtp_port) as server:
                server.starttls()
                server.login(self._username, self._password)
                server.send_message(msg)
        else:
            with smtplib.SMTP(self._smtp_host, self._smtp_port) as server:
                if self._password:
                    server.login(self._username, self._password)
                server.send_message(msg)

        self._publish_sent(channel, content, conversation_id)
        return True
    except Exception:
        logger.debug("Email send failed", exc_info=True)
        return False
status 
status() -> ChannelStatus

Return the current connection status.

Source code in src/diapason/channels/email_channel.py 
148
149
150
def status(self) -> ChannelStatus:
    """Return the current connection status."""
    return self._status
list_channels 
list_channels() -> List[str]

Return available channel identifiers.

Source code in src/diapason/channels/email_channel.py 
152
153
154
def list_channels(self) -> List[str]:
    """Return available channel identifiers."""
    return ["email"]
on_message 
on_message(handler: ChannelHandler) -> None

Register a callback for incoming email messages.

Source code in src/diapason/channels/email_channel.py 
156
157
158
def on_message(self, handler: ChannelHandler) -> None:
    """Register a callback for incoming email messages."""
    self._handlers.append(handler)