Asyncio SMTP client for sending emails asynchronously in Python applications
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
An 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 = 60Install with Tessl CLI
npx tessl i tessl/pypi-aiosmtplib