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
The send() function provides a simple, one-function interface for sending emails asynchronously. It handles connection management, authentication, and cleanup automatically, making it ideal for applications that need to send emails without managing persistent connections.
Send emails using the high-level send() function with automatic connection management.
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]:
"""
Send an email message with automatic connection management.
Parameters:
- message: Email content as EmailMessage, Message, str, or bytes
- sender: From email address (optional if message has From header)
- recipients: Recipient email addresses (optional if message has To/Cc/Bcc headers)
- mail_options: SMTP MAIL command options
- rcpt_options: SMTP RCPT command options
- hostname: SMTP server hostname or IP
- port: SMTP server port (defaults based on TLS settings)
- username: Authentication username
- password: Authentication password
- local_hostname: Local hostname for HELO/EHLO
- source_address: Source address tuple (host, port)
- timeout: Connection timeout in seconds
- use_tls: Use direct TLS/SSL connection
- start_tls: Use STARTTLS upgrade (None=auto, True=required, False=disabled)
- validate_certs: Validate server certificates
- client_cert: Path to client certificate file
- client_key: Path to client private key file
- tls_context: Custom SSL context
- cert_bundle: Path to CA certificate bundle
- socket_path: Unix domain socket path
- sock: Existing connected socket
Returns:
Tuple of (response_dict, data_response) where response_dict maps
recipient addresses to their SMTP responses
Raises:
- ValueError: If required arguments are missing
- SMTPConnectError: If connection fails
- SMTPAuthenticationError: If authentication fails
- SMTPException: For other SMTP-related errors
"""import asyncio
from email.message import EmailMessage
import aiosmtplib
async def send_basic_email():
message = EmailMessage()
message["From"] = "sender@example.com"
message["To"] = "recipient@example.com"
message["Subject"] = "Hello from aiosmtplib"
message.set_content("This is the email body.")
response = await aiosmtplib.send(
message,
hostname="smtp.gmail.com",
port=587,
start_tls=True,
username="your-email@gmail.com",
password="your-password"
)
print(f"Email sent: {response}")
asyncio.run(send_basic_email())import asyncio
import aiosmtplib
async def send_raw_email():
raw_message = """From: sender@example.com
To: recipient@example.com
Subject: Raw Message
This is a raw email message.
"""
response = await aiosmtplib.send(
raw_message,
sender="sender@example.com",
recipients=["recipient@example.com"],
hostname="localhost",
port=1025
)
print(f"Raw email sent: {response}")
asyncio.run(send_raw_email())import asyncio
from email.message import EmailMessage
import aiosmtplib
async def send_to_multiple():
message = EmailMessage()
message["From"] = "sender@example.com"
message["To"] = "recipient1@example.com, recipient2@example.com"
message["Cc"] = "cc@example.com"
message["Subject"] = "Email to Multiple Recipients"
message.set_content("This email goes to multiple people.")
response = await aiosmtplib.send(
message,
hostname="smtp.example.com",
port=587,
start_tls=True,
username="sender@example.com",
password="password"
)
# Check individual recipient responses
for recipient, smtp_response in response[0].items():
print(f"{recipient}: {smtp_response}")
asyncio.run(send_to_multiple())import asyncio
import ssl
from email.message import EmailMessage
import aiosmtplib
async def send_with_custom_tls():
message = EmailMessage()
message["From"] = "sender@example.com"
message["To"] = "recipient@example.com"
message["Subject"] = "Secure Email"
message.set_content("This email uses custom TLS settings.")
# Custom SSL context
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
response = await aiosmtplib.send(
message,
hostname="smtp.example.com",
port=465,
use_tls=True, # Direct TLS connection
tls_context=context,
username="sender@example.com",
password="password"
)
print(f"Secure email sent: {response}")
asyncio.run(send_with_custom_tls())import asyncio
from email.message import EmailMessage
import aiosmtplib
async def send_with_error_handling():
message = EmailMessage()
message["From"] = "sender@example.com"
message["To"] = "recipient@example.com"
message["Subject"] = "Test Email"
message.set_content("Testing error handling.")
try:
response = await aiosmtplib.send(
message,
hostname="smtp.example.com",
port=587,
start_tls=True,
username="sender@example.com",
password="wrong-password",
timeout=30
)
print(f"Email sent successfully: {response}")
except aiosmtplib.SMTPAuthenticationError as e:
print(f"Authentication failed: {e}")
except aiosmtplib.SMTPConnectError as e:
print(f"Connection failed: {e}")
except aiosmtplib.SMTPTimeoutError as e:
print(f"Operation timed out: {e}")
except aiosmtplib.SMTPException as e:
print(f"SMTP error: {e}")
asyncio.run(send_with_error_handling())The send() function automatically selects default ports based on encryption settings:
start_tls=True)use_tls=True)use_tls=True: Direct TLS/SSL connection (usually port 465)start_tls=True: Upgrade to TLS after initial connection (usually port 587)start_tls=None: Auto-detect STARTTLS support (default)start_tls=False: Disable TLS encryptionAuthentication is performed automatically if username and password are provided. The function supports all authentication methods available in the underlying SMTP client (CRAM-MD5, PLAIN, LOGIN).
The send() function returns a tuple containing:
SMTPResponse objectsresponse_dict, data_response = await aiosmtplib.send(message, ...)
# Check individual recipient responses
for recipient, smtp_response in response_dict.items():
print(f"{recipient}: {smtp_response.code} {smtp_response.message}")
print(f"Final response: {data_response}")Install with Tessl CLI
npx tessl i tessl/pypi-aiosmtplib