Asyncio SMTP client for sending emails asynchronously in Python applications
npx @tessl/cli install tessl/pypi-aiosmtplib@4.0.0An asyncio SMTP client for Python that provides asynchronous email sending capabilities. Built on top of Python's asyncio framework, aiosmtplib enables non-blocking email operations in asynchronous applications while supporting authentication, TLS/SSL encryption, and comprehensive SMTP features.
pip install aiosmtplibimport aiosmtplib
from aiosmtplib import SMTP, sendImport specific exceptions, types, and utilities:
from aiosmtplib import (
SMTPResponse,
SMTPStatus,
SMTPException,
SMTPConnectError,
SMTPAuthenticationError,
# Other exceptions as needed
)Import utility functions from specific modules:
from aiosmtplib.auth import auth_plain_encode, auth_login_encode, auth_crammd5_verify
from aiosmtplib.email import extract_recipients, extract_sender, flatten_message, parse_address, quote_address
from aiosmtplib.esmtp import parse_esmtp_extensionsimport asyncio
import aiosmtplib
from email.message import EmailMessage
async def send_basic_email():
# Simple high-level API usage
message = EmailMessage()
message["From"] = "sender@example.com"
message["To"] = "recipient@example.com"
message["Subject"] = "Test Email"
message.set_content("Hello from aiosmtplib!")
# Send using the high-level send() function
await aiosmtplib.send(
message,
hostname="smtp.gmail.com",
port=587,
start_tls=True,
username="your-email@gmail.com",
password="your-password"
)
async def send_with_client():
# Using the SMTP client class for persistent connections
smtp = aiosmtplib.SMTP(hostname="localhost", port=1025)
await smtp.connect()
try:
await smtp.sendmail(
"sender@example.com",
["recipient@example.com"],
"Subject: Test\n\nHello World!"
)
finally:
smtp.close()
# Run the async function
asyncio.run(send_basic_email())aiosmtplib provides multiple layers of abstraction for different use cases:
send() function for simple one-off email sending with automatic connection managementSMTP class for persistent connections, advanced features, and fine-grained controlThe library handles connection management, TLS/SSL encryption, authentication protocols (CRAM-MD5, PLAIN, LOGIN), and provides full compatibility with Python's email module for message construction.
Simple function for sending emails with automatic connection management and cleanup. Ideal for one-off email sending scenarios where you don't need persistent connections.
async def send(
message: Union[EmailMessage, Message, str, bytes],
/,
*,
sender: Optional[str] = None,
recipients: Optional[Union[str, Sequence[str]]] = None,
mail_options: Optional[Sequence[str]] = None,
rcpt_options: Optional[Sequence[str]] = None,
hostname: Optional[str] = "localhost",
port: Optional[int] = None,
username: Optional[Union[str, bytes]] = None,
password: Optional[Union[str, bytes]] = None,
local_hostname: Optional[str] = None,
source_address: Optional[tuple[str, int]] = None,
timeout: Optional[float] = 60,
use_tls: bool = False,
start_tls: Optional[bool] = None,
validate_certs: bool = True,
client_cert: Optional[str] = None,
client_key: Optional[str] = None,
tls_context: Optional[ssl.SSLContext] = None,
cert_bundle: Optional[str] = None,
socket_path: Optional[SocketPathType] = None,
sock: Optional[socket.socket] = None,
) -> tuple[dict[str, SMTPResponse], str]: ...Full-featured async SMTP client class providing persistent connections, context manager support, and comprehensive SMTP command implementations. Perfect for applications requiring multiple email operations or fine-grained control.
class SMTP:
def __init__(
self,
*,
hostname: Optional[str] = None,
port: Optional[int] = None,
username: Optional[Union[str, bytes]] = None,
password: Optional[Union[str, bytes]] = None,
local_hostname: Optional[str] = None,
source_address: Optional[tuple[str, int]] = None,
timeout: Optional[float] = 60,
use_tls: bool = False,
start_tls: Optional[bool] = None,
validate_certs: bool = True,
client_cert: Optional[str] = None,
client_key: Optional[str] = None,
tls_context: Optional[ssl.SSLContext] = None,
cert_bundle: Optional[str] = None,
socket_path: Optional[SocketPathType] = None,
sock: Optional[socket.socket] = None,
) -> None: ...
# Connection management
async def connect(...) -> SMTPResponse: ...
def close(self) -> None: ...
# Properties
@property
def is_connected(self) -> bool: ...
@property
def supported_auth_methods(self) -> list[str]: ...
# High-level email sending
async def sendmail(...) -> tuple[dict[str, SMTPResponse], str]: ...
async def send_message(...) -> tuple[dict[str, SMTPResponse], str]: ...
# Authentication
async def login(username: Union[str, bytes], password: Union[str, bytes]) -> SMTPResponse: ...
async def auth_plain(...) -> SMTPResponse: ...
async def auth_login(...) -> SMTPResponse: ...
async def auth_crammd5(...) -> SMTPResponse: ...
# SMTP/ESMTP commands
async def helo(...) -> SMTPResponse: ...
async def ehlo(...) -> SMTPResponse: ...
async def starttls(...) -> SMTPResponse: ...
async def mail(...) -> SMTPResponse: ...
async def rcpt(...) -> SMTPResponse: ...
async def data(...) -> SMTPResponse: ...
async def rset() -> SMTPResponse: ...
async def quit() -> SMTPResponse: ...
async def noop() -> SMTPResponse: ...
async def vrfy(...) -> SMTPResponse: ...
async def expn(...) -> SMTPResponse: ...
async def help() -> str: ...
# Utility methods
def supports_extension(extension: str) -> bool: ...
async def execute_command(*args: bytes) -> SMTPResponse: ...
def get_transport_info(key: str) -> Any: ...Comprehensive exception hierarchy for handling different types of SMTP errors, from connection issues to authentication failures and server responses.
class SMTPException(Exception): ...
class SMTPConnectError(SMTPException, ConnectionError): ...
class SMTPTimeoutError(SMTPException, TimeoutError): ...
class SMTPAuthenticationError(SMTPResponseException): ...
class SMTPRecipientRefused(SMTPResponseException): ...
class SMTPSenderRefused(SMTPResponseException): ...Helper functions for authentication, email message processing, and ESMTP extension parsing. Useful for custom implementations and debugging.
def auth_plain_encode(username: str, password: str) -> bytes: ...
def extract_recipients(message: Union[EmailMessage, Message]) -> list[str]: ...
def parse_esmtp_extensions(message: str) -> tuple[dict[str, str], list[str]]: ...class SMTPResponse(NamedTuple):
"""Server response containing status code and message."""
code: int
message: str
class SMTPStatus(IntEnum):
"""SMTP status codes for response handling."""
invalid_response = -1
system_status_ok = 211
help_message = 214
ready = 220
closing = 221
auth_successful = 235
completed = 250
will_forward = 251
cannot_vrfy = 252
auth_continue = 334
start_input = 354
domain_unavailable = 421
mailbox_unavailable = 450
error_processing = 451
insufficient_storage = 452
tls_not_available = 454
unrecognized_command = 500
unrecognized_parameters = 501
command_not_implemented = 502
bad_command_sequence = 503
parameter_not_implemented = 504
domain_does_not_accept_mail = 521
access_denied = 530
auth_failed = 535
mailbox_does_not_exist = 550
user_not_local = 551
storage_exceeded = 552
mailbox_name_invalid = 553
transaction_failed = 554
syntax_error = 555
SocketPathType = Union[str, bytes, os.PathLike[str]]SMTP_PORT: int = 25
SMTP_TLS_PORT: int = 465
SMTP_STARTTLS_PORT: int = 587
DEFAULT_TIMEOUT: float = 60