API Reference

This document describes the public API for simple-email-gw.

Note: This package provides both async and sync APIs. Use async clients (IMAPClient, SMTPClient) for async applications, or sync wrapper clients (SyncIMAPClient, SyncSMTPClient) for simpler synchronous code. See Sync Client API Reference for sync client documentation.

Choosing Async or Sync

Async clients (recommended for async applications):

  • FastAPI, Quart, asyncio-based applications

  • Better performance in async contexts

  • No thread overhead

Sync clients (for simpler synchronous code):

  • Scripts, CLI tools, synchronous applications

  • Simpler syntax (no async/await)

  • Context manager support (with statement)

Installation

Requirements

  • Python 3.10 or higher

  • An email account (IMAP/SMTP access)

Install from PyPI

pip install simple-email-gw

Install with uv

uv add simple-email-gw

Run with uvx

uvx --from simple-email-gw email-gw-mcp-server

Dependencies

Package

Purpose

fastmcp

MCP server framework

aioimaplib

Async IMAP client

aiosmtplib

Async SMTP client

pydantic

Data validation

pydantic-settings

Settings management

Clients

IMAPClient

Async IMAP client for reading and managing emails.

from simple_email_gw import IMAPClient, EmailAccount

account = EmailAccount(
  name="work",
  imap_host="imap.gmail.com",
  smtp_host="smtp.gmail.com",
  username="user@gmail.com",
  password="app-password"
)

async with IMAPClient(account) as client:
  folders = await client.list_folders()
  messages = await client.search(folder="INBOX", criteria="UNSEEN")
  msg = await client.fetch_message(messages[0])

Methods

Method

Description

connect()

Establish IMAP connection

disconnect()

Close IMAP connection

list_folders()

List all folders/mailboxes

select_folder(folder)

Select a folder and return message count

search(folder, criteria, limit)

Search for messages

fetch_message(message_id, folder)

Fetch a single message

move_message(message_id, source, dest)

Move message between folders

delete_message(message_id, folder, expunge)

Delete a message

mark_message(message_id, folder, flag, action)

Add/remove flags

download_attachment(message_id, folder, filename, output_dir)

Download attachment

SMTPClient

Async SMTP client for sending emails.

from simple_email_gw import SMTPClient, EmailAccount

account = EmailAccount(
  name="work",
  imap_host="imap.gmail.com",
  smtp_host="smtp.gmail.com",
  username="user@gmail.com",
  password="app-password"
)

smtp = SMTPClient(account)
result = await smtp.send_email(
  to=["recipient@example.com"],
  subject="Hello",
  body="World!"
)

Methods

Method

Description

send_email(to, subject, body, cc, bcc, html_body, attachments)

Send new email

reply_email(to, subject, body, in_reply_to, references, html_body)

Reply to thread

forward_email(to, subject, original_from, original_date, original_body)

Forward email

Configuration

EmailAccount

Pydantic model for account configuration.

from simple_email_gw import EmailAccount

account = EmailAccount(
  name="work",
  imap_host="imap.gmail.com",
  imap_port=993,
  smtp_host="smtp.gmail.com",
  smtp_port=587,
  username="user@gmail.com",
  password="app-password",
  auth_method="password",
  use_ssl=True
)

Fields

Field

Type

Default

Description

name

str

“default”

Account name

imap_host

str

required

IMAP server hostname

imap_port

int

993

IMAP port

smtp_host

str

required

SMTP server hostname

smtp_port

int

587

SMTP port

username

str

required

Email address/username

password

SecretStr

None

Password or app password

oauth2_token

SecretStr

None

OAuth2 access token

auth_method

str

“password”

“password” or “oauth2”

use_ssl

bool

True

Use SSL/TLS

Multiple Accounts

Configure multiple accounts using JSON:

export EMAIL_ACCOUNTS_JSON='[
  {
    "name": "work",
    "imap_host": "imap.work.com",
    "smtp_host": "smtp.work.com",
    "username": "work@company.com",
    "password": "work-password"
  },
  {
    "name": "personal",
    "imap_host": "imap.gmail.com",
    "smtp_host": "smtp.gmail.com",
    "username": "personal@gmail.com",
    "password": "app-password"
  }
]'

Connection Pool

Manage connections with automatic pooling.

from simple_email_gw import get_pool

pool = await get_pool()

# Get accounts from environment
accounts = await pool.get_accounts()

# Get client for specific account
client = await pool.get_imap_client("work")

Safety Utilities

RateLimiter

Token bucket rate limiter.

from simple_email_gw import RateLimiter

limiter = RateLimiter(rate=60, window=60)

if await limiter.acquire("account_name"):
  # Request allowed
  pass

Sanitization

Functions to sanitize user input.

from simple_email_gw import (
  sanitize_subject,
  sanitize_message_id,
  sanitize_references,
  sanitize_header_value,
  sanitize_folder_name,
  sanitize_filename,
  sanitize_message_id_numeric,
)

# Sanitize subject line (removes CRLF)
safe = sanitize_subject("Hello\r\nBcc: attacker@evil.com")
# Returns: "Hello Bcc: attacker@evil.com"

# Validate Message-ID format
safe_id = sanitize_message_id("<msg123@example.com>")

# Sanitize folder name (prevents CRLF injection)
safe_folder = sanitize_folder_name("INBOX")

# Sanitize filename (prevents CRLF injection)
safe_filename = sanitize_filename("document.pdf")

Audit Logging

Log security events.

from simple_email_gw import (
  log_event,
  log_email_sent,
  log_auth_attempt,
  log_rate_limited,
  log_attachment_download,
)

log_email_sent(
  account="work",
  recipients=["user@example.com"],
  subject="Hello",
  has_attachments=False
)

log_auth_attempt(
  account="work",
  success=True,
  method="password"
)

Errors

SecurityError

Raised when a security constraint is violated.

from simple_email_gw import SecurityError

try:
  await client.download_attachment(...)
except SecurityError as e:
  # Handle security violation
  pass

WhitelistError

Raised when a recipient is not in the whitelist.

from simple_email_gw import WhitelistError

try:
  await smtp.send_email(to=["untrusted@evil.com"], ...)
except WhitelistError as e:
  # Handle whitelist violation
  pass

RateLimitError

Raised when rate limit is exceeded.

from simple_email_gw import RateLimitError

try:
  client = await pool.get_imap_client("work")
except RateLimitError:
  # Wait and retry
  pass

MCP Server

For MCP server tools and resources, see MCP Tools Reference.

Next Steps