How to Send Emails in FastAPI (2026 Guide)

Most "how to send email in FastAPI" tutorials show smtplib or a sync library call inside an async route. That defeats the whole point of FastAPI being async. You end up blocking your event loop on every email send.

This guide covers the full picture: picking a provider, sending emails asynchronously with httpx, validating inputs with Pydantic, offloading sends with BackgroundTasks, building HTML templates with Jinja2, handling Stripe webhooks, and shipping to production. For broader Python coverage, see our Python email guide, or check out the Django and Flask guides. All code examples use Python type hints and FastAPI's dependency injection.

Pick an Email Provider

You have three solid options. Each code example below lets you switch between them.

Sequenzy is built for SaaS. Transactional emails, marketing campaigns, automated sequences, subscriber management, all from one API. If you're building a SaaS product, this is the simplest path because you won't need to glue together three different tools later. Native Stripe integration and built-in retries.
Resend is a developer-friendly transactional email API. Clean DX, good docs, solid deliverability. They have one-off broadcast campaigns but no automations or sequences.
SendGrid is the enterprise standard. Feature-rich, sometimes complex. Good if you need high volume and don't mind a bigger API surface.

FastAPI email choice	Best for	Why	Avoid
`httpx.AsyncClient`	All provider API calls in async routes	Does not block the event loop	Sync provider SDK calls inside routes
`BackgroundTasks`	Lightweight post-response sends	Simple built-in offload for small apps	Long-running or retry-heavy campaign work
External worker	Campaigns, retries, and webhook fanout	Durable processing and better observability	Treating BackgroundTasks as a queue
Pydantic settings	Provider keys and sender config	Startup validation catches missing secrets	Raw `os.environ` scattered through code
Provider automation	Lifecycle and billing sequences	Keeps timing/state out of FastAPI	Sending duplicate lifecycle emails from both systems

Install

Terminal

pip install fastapi uvicorn httpx pydantic-settings jinja2

Terminal

pip install fastapi uvicorn httpx pydantic-settings jinja2

Terminal

pip install fastapi uvicorn httpx pydantic-settings jinja2

We're using httpx (async HTTP client) for all three providers instead of their Python SDKs. Why? The Resend and SendGrid Python SDKs are synchronous - they'll block your FastAPI event loop. Every email provider is just an HTTP API, and httpx gives you proper async support.

Dependency	Role	Production note
`fastapi`	API routes, DI, webhook endpoints	Keep email side effects outside critical request paths
`httpx`	Async provider HTTP calls	Reuse one client with app lifespan
`pydantic-settings`	Typed config and `.env` loading	Fail fast when keys are missing
`jinja2`	HTML email templates	Render once, then pass HTML to provider
`uvicorn`	Local ASGI server	Use production ASGI deployment settings

Add your API key to .env:

.env

SEQUENZY_API_KEY=sq_your_api_key_here

.env

RESEND_API_KEY=re_your_api_key_here

.env

SENDGRID_API_KEY=SG.your_api_key_here

Configure Settings with Pydantic

Use pydantic-settings to manage configuration. It reads from environment variables and .env files with full type validation:

config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
  sequenzy_api_key: str
  app_url: str = "https://yourapp.com"

  model_config = {"env_file": ".env"}

settings = Settings()

config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
  resend_api_key: str
  from_email: str = "Your App <noreply@yourdomain.com>"
  app_url: str = "https://yourapp.com"

  model_config = {"env_file": ".env"}

settings = Settings()

config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
  sendgrid_api_key: str
  from_email: str = "noreply@yourdomain.com"
  app_url: str = "https://yourapp.com"

  model_config = {"env_file": ".env"}

settings = Settings()

This is better than raw os.environ because you get validation at startup - if SEQUENZY_API_KEY is missing, FastAPI won't even start instead of crashing on the first email send.

Create the Email Client

Create a reusable async email client. Use httpx.AsyncClient with a managed lifecycle so connections are properly pooled and cleaned up:

email_client.py

import httpx
from config import settings

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.sequenzy.com/v1",
          headers={"Authorization": f"Bearer {settings.sequenzy_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized. Call startup() first.")
      return self._client

  async def send(self, to: str, subject: str, body: str) -> dict:
      response = await self.client.post(
          "/transactional/send",
          json={"to": to, "subject": subject, "body": body},
      )
      response.raise_for_status()
      return response.json()

email_client = EmailClient()

email_client.py

import httpx
from config import settings

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.resend.com",
          headers={"Authorization": f"Bearer {settings.resend_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized. Call startup() first.")
      return self._client

  async def send(self, to: str, subject: str, html: str) -> dict:
      response = await self.client.post(
          "/emails",
          json={
              "from": settings.from_email,
              "to": to,
              "subject": subject,
              "html": html,
          },
      )
      response.raise_for_status()
      return response.json()

email_client = EmailClient()

email_client.py

import httpx
from config import settings

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.sendgrid.com/v3",
          headers={"Authorization": f"Bearer {settings.sendgrid_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized. Call startup() first.")
      return self._client

  async def send(self, to: str, subject: str, html: str) -> dict:
      response = await self.client.post(
          "/mail/send",
          json={
              "personalizations": [{"to": [{"email": to}]}],
              "from": {"email": settings.from_email},
              "subject": subject,
              "content": [{"type": "text/html", "value": html}],
          },
      )
      response.raise_for_status()
      return {"sent": True}

email_client = EmailClient()

Key things happening here:

Connection pooling: httpx.AsyncClient reuses TCP connections across requests. Creating a new client per request wastes time on TLS handshakes.
Lifecycle management: The client opens on startup and closes on shutdown. No leaked connections.
Timeout: 30 seconds is generous but prevents hanging forever if the provider is down.

Wire Up the App Lifecycle

Use FastAPI's lifespan context manager to start and stop the email client:

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from email_client import email_client
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    await email_client.startup()
    yield
    await email_client.shutdown()
 
app = FastAPI(lifespan=lifespan)

This ensures the httpx.AsyncClient is properly initialized before your app handles requests and properly closed when the app shuts down.

Send Your First Email

The simplest possible email. Pydantic validates the request body, FastAPI handles the rest.

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from email_client import email_client

@asynccontextmanager
async def lifespan(app: FastAPI):
  await email_client.startup()
  yield
  await email_client.shutdown()

app = FastAPI(lifespan=lifespan)

class SendRequest(BaseModel):
  email: EmailStr
  name: str

@app.post("/api/send")
async def send_welcome(req: SendRequest):
  try:
      result = await email_client.send(
          to=req.email,
          subject=f"Welcome, {req.name}",
          body=f"<h1>Welcome, {req.name}</h1><p>Your account is ready.</p>",
      )
      return {"jobId": result["jobId"]}
  except Exception as e:
      raise HTTPException(status_code=500, detail=str(e))

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from email_client import email_client

@asynccontextmanager
async def lifespan(app: FastAPI):
  await email_client.startup()
  yield
  await email_client.shutdown()

app = FastAPI(lifespan=lifespan)

class SendRequest(BaseModel):
  email: EmailStr
  name: str

@app.post("/api/send")
async def send_welcome(req: SendRequest):
  try:
      result = await email_client.send(
          to=req.email,
          subject=f"Welcome, {req.name}",
          html=f"<h1>Welcome, {req.name}</h1><p>Your account is ready.</p>",
      )
      return {"id": result["id"]}
  except Exception as e:
      raise HTTPException(status_code=500, detail=str(e))

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from email_client import email_client

@asynccontextmanager
async def lifespan(app: FastAPI):
  await email_client.startup()
  yield
  await email_client.shutdown()

app = FastAPI(lifespan=lifespan)

class SendRequest(BaseModel):
  email: EmailStr
  name: str

@app.post("/api/send")
async def send_welcome(req: SendRequest):
  try:
      await email_client.send(
          to=req.email,
          subject=f"Welcome, {req.name}",
          html=f"<h1>Welcome, {req.name}</h1><p>Your account is ready.</p>",
      )
      return {"sent": True}
  except Exception as e:
      raise HTTPException(status_code=500, detail=str(e))

Test it:

uvicorn main:app --reload
curl -X POST http://localhost:8000/api/send \
  -H "Content-Type: application/json" \
  -d '{"email": "user@example.com", "name": "Alice"}'

Pydantic's EmailStr validates the email format before your code even runs. Invalid emails get a 422 response automatically.

Build Email Templates with Jinja2

Raw f-strings get messy fast. Jinja2 gives you inheritance, conditionals, loops, and auto-escaping:

project/
├── templates/
│   └── emails/
│       ├── base.html
│       ├── welcome.html
│       ├── password-reset.html
│       └── receipt.html
├── email_client.py
├── email_renderer.py
├── config.py
└── main.py

Start with a base template:

<!-- templates/emails/base.html -->
<!DOCTYPE html>
<html>
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body style="margin:0;padding:0;background-color:#f6f9fc;font-family:sans-serif;">
  <div style="max-width:480px;margin:40px auto;background:#fff;padding:40px;border-radius:8px;">
    {% block content %}{% endblock %}
    <hr style="border:none;border-top:1px solid #e5e7eb;margin:32px 0 16px;" />
    <p style="font-size:12px;color:#9ca3af;">
      &copy; {{ year }} Your App. All rights reserved.
    </p>
  </div>
</body>
</html>

Then extend it for each email type:

<!-- templates/emails/welcome.html -->
{% extends "base.html" %}
 
{% block content %}
<h1 style="font-size:24px;margin-bottom:16px;">Welcome, {{ name }}</h1>
<p style="font-size:16px;line-height:1.6;color:#374151;">
  Your account is ready. You can log in and start exploring.
</p>
<a href="{{ login_url }}"
   style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;font-size:14px;font-weight:600;margin-top:16px;">
  Go to Dashboard
</a>
{% endblock %}

Create a renderer that loads and renders templates:

# email_renderer.py
from datetime import datetime
from jinja2 import Environment, FileSystemLoader, select_autoescape
 
env = Environment(
    loader=FileSystemLoader("templates/emails"),
    autoescape=select_autoescape(["html"]),
)
 
def render_email(template_name: str, **kwargs) -> str:
    template = env.get_template(template_name)
    return template.render(year=datetime.now().year, **kwargs)

Use it in your routes:

main.py

from email_renderer import render_email

@app.post("/api/send-welcome")
async def send_welcome(req: SendRequest):
  html = render_email("welcome.html", name=req.name, login_url="https://app.yoursite.com")

  result = await email_client.send(
      to=req.email,
      subject=f"Welcome, {req.name}",
      body=html,
  )
  return {"jobId": result["jobId"]}

main.py

from email_renderer import render_email

@app.post("/api/send-welcome")
async def send_welcome(req: SendRequest):
  html = render_email("welcome.html", name=req.name, login_url="https://app.yoursite.com")

  result = await email_client.send(
      to=req.email,
      subject=f"Welcome, {req.name}",
      html=html,
  )
  return {"id": result["id"]}

main.py

from email_renderer import render_email

@app.post("/api/send-welcome")
async def send_welcome(req: SendRequest):
  html = render_email("welcome.html", name=req.name, login_url="https://app.yoursite.com")

  await email_client.send(
      to=req.email,
      subject=f"Welcome, {req.name}",
      html=html,
  )
  return {"sent": True}

Background Tasks

FastAPI's built-in BackgroundTasks lets you return a response immediately and send the email after. The email runs in the same process, after the response is sent:

main.py

from fastapi import BackgroundTasks
from email_renderer import render_email
import logging

logger = logging.getLogger(__name__)

async def send_welcome_email(email: str, name: str):
  try:
      html = render_email("welcome.html", name=name, login_url="https://app.yoursite.com")
      await email_client.send(to=email, subject=f"Welcome, {name}", body=html)
      logger.info(f"Welcome email sent to {email}")
  except Exception:
      logger.exception(f"Failed to send welcome email to {email}")

@app.post("/api/signup")
async def signup(req: SendRequest, background_tasks: BackgroundTasks):
  # Create user in database first...

  background_tasks.add_task(send_welcome_email, req.email, req.name)

  return {"message": "Account created"}

main.py

from fastapi import BackgroundTasks
from email_renderer import render_email
import logging

logger = logging.getLogger(__name__)

async def send_welcome_email(email: str, name: str):
  try:
      html = render_email("welcome.html", name=name, login_url="https://app.yoursite.com")
      await email_client.send(to=email, subject=f"Welcome, {name}", html=html)
      logger.info(f"Welcome email sent to {email}")
  except Exception:
      logger.exception(f"Failed to send welcome email to {email}")

@app.post("/api/signup")
async def signup(req: SendRequest, background_tasks: BackgroundTasks):
  # Create user in database first...

  background_tasks.add_task(send_welcome_email, req.email, req.name)

  return {"message": "Account created"}

main.py

from fastapi import BackgroundTasks
from email_renderer import render_email
import logging

logger = logging.getLogger(__name__)

async def send_welcome_email(email: str, name: str):
  try:
      html = render_email("welcome.html", name=name, login_url="https://app.yoursite.com")
      await email_client.send(to=email, subject=f"Welcome, {name}", html=html)
      logger.info(f"Welcome email sent to {email}")
  except Exception:
      logger.exception(f"Failed to send welcome email to {email}")

@app.post("/api/signup")
async def signup(req: SendRequest, background_tasks: BackgroundTasks):
  # Create user in database first...

  background_tasks.add_task(send_welcome_email, req.email, req.name)

  return {"message": "Account created"}

The BackgroundTasks parameter is injected by FastAPI automatically. The response returns immediately, and the email sends after. If the email fails, it logs the error but doesn't crash the request.

Dependency Injection for Email

FastAPI's Depends system lets you inject the email client into routes cleanly. This is especially useful if you want to swap the client in tests:

# dependencies.py
from email_client import email_client, EmailClient
 
async def get_email_client() -> EmailClient:
    return email_client

# routes/emails.py
from fastapi import APIRouter, Depends, BackgroundTasks
from pydantic import BaseModel, EmailStr
from dependencies import get_email_client
from email_client import EmailClient
 
router = APIRouter(prefix="/api")
 
class ContactRequest(BaseModel):
    email: EmailStr
    name: str
    message: str
 
@router.post("/contact")
async def contact(
    req: ContactRequest,
    background_tasks: BackgroundTasks,
    email: EmailClient = Depends(get_email_client),
):
    async def send():
        await email.send(
            to="you@yourcompany.com",
            subject=f"Contact from {req.name}",
            body=f"<p><strong>From:</strong> {req.email}</p><p>{req.message}</p>",
        )
 
    background_tasks.add_task(send)
    return {"message": "Message received"}

In tests, you override the dependency:

from fastapi.testclient import TestClient
from unittest.mock import AsyncMock
 
mock_client = AsyncMock()
app.dependency_overrides[get_email_client] = lambda: mock_client
 
client = TestClient(app)
response = client.post("/api/contact", json={...})
mock_client.send.assert_called_once()

Common Email Patterns for SaaS

Here are the emails almost every SaaS app needs, with production-ready implementations.

Password Reset

emails/password_reset.py

from email_client import email_client
from email_renderer import render_email
from config import settings

async def send_password_reset(email: str, reset_token: str):
  reset_url = f"{settings.app_url}/reset-password?token={reset_token}"
  html = render_email("password-reset.html", reset_url=reset_url)

  await email_client.send(
      to=email,
      subject="Reset your password",
      body=html,
  )

emails/password_reset.py

from email_client import email_client
from email_renderer import render_email
from config import settings

async def send_password_reset(email: str, reset_token: str):
  reset_url = f"{settings.app_url}/reset-password?token={reset_token}"
  html = render_email("password-reset.html", reset_url=reset_url)

  await email_client.send(
      to=email,
      subject="Reset your password",
      html=html,
  )

emails/password_reset.py

from email_client import email_client
from email_renderer import render_email
from config import settings

async def send_password_reset(email: str, reset_token: str):
  reset_url = f"{settings.app_url}/reset-password?token={reset_token}"
  html = render_email("password-reset.html", reset_url=reset_url)

  await email_client.send(
      to=email,
      subject="Reset your password",
      html=html,
  )

The template:

<!-- templates/emails/password-reset.html -->
{% extends "base.html" %}
 
{% block content %}
<h2 style="font-size:20px;margin-bottom:16px;">Password Reset</h2>
<p style="font-size:16px;line-height:1.6;color:#374151;">
  Click the button below to reset your password. This link expires in 1 hour.
</p>
<a href="{{ reset_url }}"
   style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;font-weight:600;">
  Reset Password
</a>
<p style="color:#6b7280;font-size:14px;margin-top:24px;">
  If you didn't request this, ignore this email.
</p>
{% endblock %}

Payment Receipt

emails/receipt.py

from email_client import email_client
from email_renderer import render_email

async def send_receipt(email: str, amount: int, plan: str, invoice_url: str):
  formatted = f"${amount / 100:.2f}"
  html = render_email(
      "receipt.html",
      amount=formatted,
      plan=plan,
      invoice_url=invoice_url,
  )

  await email_client.send(
      to=email,
      subject=f"Payment receipt - {formatted}",
      body=html,
  )

emails/receipt.py

from email_client import email_client
from email_renderer import render_email

async def send_receipt(email: str, amount: int, plan: str, invoice_url: str):
  formatted = f"${amount / 100:.2f}"
  html = render_email(
      "receipt.html",
      amount=formatted,
      plan=plan,
      invoice_url=invoice_url,
  )

  await email_client.send(
      to=email,
      subject=f"Payment receipt - {formatted}",
      html=html,
  )

emails/receipt.py

from email_client import email_client
from email_renderer import render_email

async def send_receipt(email: str, amount: int, plan: str, invoice_url: str):
  formatted = f"${amount / 100:.2f}"
  html = render_email(
      "receipt.html",
      amount=formatted,
      plan=plan,
      invoice_url=invoice_url,
  )

  await email_client.send(
      to=email,
      subject=f"Payment receipt - {formatted}",
      html=html,
  )

<!-- templates/emails/receipt.html -->
{% extends "base.html" %}
 
{% block content %}
<h2 style="font-size:20px;margin-bottom:16px;">Payment Received</h2>
<p style="font-size:16px;color:#374151;">Thanks for your payment. Here's your receipt:</p>
<table style="width:100%;border-collapse:collapse;margin:16px 0;">
  <tr>
    <td style="padding:8px;border-bottom:1px solid #e5e7eb;">Plan</td>
    <td style="padding:8px;border-bottom:1px solid #e5e7eb;text-align:right;">{{ plan }}</td>
  </tr>
  <tr>
    <td style="padding:8px;font-weight:600;">Total</td>
    <td style="padding:8px;text-align:right;font-weight:600;">{{ amount }}</td>
  </tr>
</table>
<a href="{{ invoice_url }}" style="color:#f97316;">View full invoice</a>
{% endblock %}

Stripe Webhook Handler

routes/stripe_webhook.py

import hmac
import hashlib
import time
from fastapi import APIRouter, Request, HTTPException, BackgroundTasks
from emails.receipt import send_receipt
from email_client import email_client
from config import settings

router = APIRouter()

def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
  parts = dict(item.split("=", 1) for item in signature.split(","))
  timestamp = parts.get("t", "")
  expected_sig = parts.get("v1", "")

  if not timestamp or not expected_sig:
      return False

  # Check timestamp is within 5 minutes
  if abs(time.time() - int(timestamp)) > 300:
      return False

  signed_payload = f"{timestamp}.{payload.decode()}"
  computed = hmac.new(
      secret.encode(),
      signed_payload.encode(),
      hashlib.sha256,
  ).hexdigest()

  return hmac.compare_digest(computed, expected_sig)

@router.post("/api/webhooks/stripe")
async def stripe_webhook(request: Request, background_tasks: BackgroundTasks):
  body = await request.body()
  signature = request.headers.get("stripe-signature", "")

  if not verify_stripe_signature(body, signature, settings.stripe_webhook_secret):
      raise HTTPException(status_code=400, detail="Invalid signature")

  event = await request.json()

  if event["type"] == "checkout.session.completed":
      session = event["data"]["object"]

      background_tasks.add_task(
          send_receipt,
          email=session["customer_email"],
          amount=session["amount_total"],
          plan=session.get("metadata", {}).get("plan", "Pro"),
          invoice_url=session.get("invoice_url", f"{settings.app_url}/billing"),
      )

  if event["type"] == "invoice.payment_failed":
      invoice = event["data"]["object"]

      async def send_failed_payment():
          await email_client.send(
              to=invoice["customer_email"],
              subject="Payment failed - action needed",
              body=f"""
                  <h2>Payment Failed</h2>
                  <p>We couldn't process your payment. Please update your billing info.</p>
                  <a href="{settings.app_url}/billing"
                     style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;">
                    Update Billing
                  </a>
              """,
          )

      background_tasks.add_task(send_failed_payment)

  return {"received": True}

routes/stripe_webhook.py

import hmac
import hashlib
import time
from fastapi import APIRouter, Request, HTTPException, BackgroundTasks
from emails.receipt import send_receipt
from email_client import email_client
from config import settings

router = APIRouter()

def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
  parts = dict(item.split("=", 1) for item in signature.split(","))
  timestamp = parts.get("t", "")
  expected_sig = parts.get("v1", "")

  if not timestamp or not expected_sig:
      return False

  if abs(time.time() - int(timestamp)) > 300:
      return False

  signed_payload = f"{timestamp}.{payload.decode()}"
  computed = hmac.new(
      secret.encode(),
      signed_payload.encode(),
      hashlib.sha256,
  ).hexdigest()

  return hmac.compare_digest(computed, expected_sig)

@router.post("/api/webhooks/stripe")
async def stripe_webhook(request: Request, background_tasks: BackgroundTasks):
  body = await request.body()
  signature = request.headers.get("stripe-signature", "")

  if not verify_stripe_signature(body, signature, settings.stripe_webhook_secret):
      raise HTTPException(status_code=400, detail="Invalid signature")

  event = await request.json()

  if event["type"] == "checkout.session.completed":
      session = event["data"]["object"]

      background_tasks.add_task(
          send_receipt,
          email=session["customer_email"],
          amount=session["amount_total"],
          plan=session.get("metadata", {}).get("plan", "Pro"),
          invoice_url=session.get("invoice_url", f"{settings.app_url}/billing"),
      )

  if event["type"] == "invoice.payment_failed":
      invoice = event["data"]["object"]

      async def send_failed_payment():
          await email_client.send(
              to=invoice["customer_email"],
              subject="Payment failed - action needed",
              html=f"""
                  <h2>Payment Failed</h2>
                  <p>We couldn't process your payment. Please update your billing info.</p>
                  <a href="{settings.app_url}/billing"
                     style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;">
                    Update Billing
                  </a>
              """,
          )

      background_tasks.add_task(send_failed_payment)

  return {"received": True}

routes/stripe_webhook.py

import hmac
import hashlib
import time
from fastapi import APIRouter, Request, HTTPException, BackgroundTasks
from emails.receipt import send_receipt
from email_client import email_client
from config import settings

router = APIRouter()

def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
  parts = dict(item.split("=", 1) for item in signature.split(","))
  timestamp = parts.get("t", "")
  expected_sig = parts.get("v1", "")

  if not timestamp or not expected_sig:
      return False

  if abs(time.time() - int(timestamp)) > 300:
      return False

  signed_payload = f"{timestamp}.{payload.decode()}"
  computed = hmac.new(
      secret.encode(),
      signed_payload.encode(),
      hashlib.sha256,
  ).hexdigest()

  return hmac.compare_digest(computed, expected_sig)

@router.post("/api/webhooks/stripe")
async def stripe_webhook(request: Request, background_tasks: BackgroundTasks):
  body = await request.body()
  signature = request.headers.get("stripe-signature", "")

  if not verify_stripe_signature(body, signature, settings.stripe_webhook_secret):
      raise HTTPException(status_code=400, detail="Invalid signature")

  event = await request.json()

  if event["type"] == "checkout.session.completed":
      session = event["data"]["object"]

      background_tasks.add_task(
          send_receipt,
          email=session["customer_email"],
          amount=session["amount_total"],
          plan=session.get("metadata", {}).get("plan", "Pro"),
          invoice_url=session.get("invoice_url", f"{settings.app_url}/billing"),
      )

  if event["type"] == "invoice.payment_failed":
      invoice = event["data"]["object"]

      async def send_failed_payment():
          await email_client.send(
              to=invoice["customer_email"],
              subject="Payment failed - action needed",
              html=f"""
                  <h2>Payment Failed</h2>
                  <p>We couldn't process your payment. Please update your billing info.</p>
                  <a href="{settings.app_url}/billing"
                     style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;">
                    Update Billing
                  </a>
              """,
          )

      background_tasks.add_task(send_failed_payment)

  return {"received": True}

Key things: we use request.body() to get the raw bytes for signature verification, then request.json() for the parsed event. The hmac.compare_digest function prevents timing attacks.

Error Handling

Emails fail. Networks time out. Rate limits hit. Here's how to handle it properly.

email_client.py

import httpx
import logging
from config import settings

logger = logging.getLogger(__name__)

class EmailSendError(Exception):
  def __init__(self, message: str, status: int, retryable: bool):
      super().__init__(message)
      self.status = status
      self.retryable = retryable

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.sequenzy.com/v1",
          headers={"Authorization": f"Bearer {settings.sequenzy_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized")
      return self._client

  async def send(self, to: str, subject: str, body: str) -> dict:
      try:
          response = await self.client.post(
              "/transactional/send",
              json={"to": to, "subject": subject, "body": body},
          )
      except httpx.TimeoutException:
          raise EmailSendError("Request timed out", 0, retryable=True)
      except httpx.ConnectError:
          raise EmailSendError("Connection failed", 0, retryable=True)

      if response.status_code == 429:
          raise EmailSendError("Rate limited", 429, retryable=True)
      if response.status_code == 401:
          raise EmailSendError("Invalid API key", 401, retryable=False)
      if response.status_code >= 500:
          raise EmailSendError(f"Server error: {response.text}", response.status_code, retryable=True)
      if not response.is_success:
          raise EmailSendError(f"Send failed: {response.text}", response.status_code, retryable=False)

      logger.info(f"Email sent to {to}: {subject}")
      return response.json()

email_client = EmailClient()

email_client.py

import httpx
import logging
from config import settings

logger = logging.getLogger(__name__)

class EmailSendError(Exception):
  def __init__(self, message: str, status: int, retryable: bool):
      super().__init__(message)
      self.status = status
      self.retryable = retryable

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.resend.com",
          headers={"Authorization": f"Bearer {settings.resend_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized")
      return self._client

  async def send(self, to: str, subject: str, html: str) -> dict:
      try:
          response = await self.client.post(
              "/emails",
              json={
                  "from": settings.from_email,
                  "to": to,
                  "subject": subject,
                  "html": html,
              },
          )
      except httpx.TimeoutException:
          raise EmailSendError("Request timed out", 0, retryable=True)
      except httpx.ConnectError:
          raise EmailSendError("Connection failed", 0, retryable=True)

      if response.status_code == 429:
          raise EmailSendError("Rate limited", 429, retryable=True)
      if response.status_code in (401, 403):
          raise EmailSendError("Invalid API key", response.status_code, retryable=False)
      if response.status_code >= 500:
          raise EmailSendError(f"Server error: {response.text}", response.status_code, retryable=True)
      if not response.is_success:
          raise EmailSendError(f"Send failed: {response.text}", response.status_code, retryable=False)

      logger.info(f"Email sent to {to}: {subject}")
      return response.json()

email_client = EmailClient()

email_client.py

import httpx
import logging
from config import settings

logger = logging.getLogger(__name__)

class EmailSendError(Exception):
  def __init__(self, message: str, status: int, retryable: bool):
      super().__init__(message)
      self.status = status
      self.retryable = retryable

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.sendgrid.com/v3",
          headers={"Authorization": f"Bearer {settings.sendgrid_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized")
      return self._client

  async def send(self, to: str, subject: str, html: str) -> dict:
      try:
          response = await self.client.post(
              "/mail/send",
              json={
                  "personalizations": [{"to": [{"email": to}]}],
                  "from": {"email": settings.from_email},
                  "subject": subject,
                  "content": [{"type": "text/html", "value": html}],
              },
          )
      except httpx.TimeoutException:
          raise EmailSendError("Request timed out", 0, retryable=True)
      except httpx.ConnectError:
          raise EmailSendError("Connection failed", 0, retryable=True)

      if response.status_code == 429:
          raise EmailSendError("Rate limited", 429, retryable=True)
      if response.status_code in (401, 403):
          raise EmailSendError("Invalid API key", response.status_code, retryable=False)
      if response.status_code >= 500:
          raise EmailSendError(f"Server error: {response.text}", response.status_code, retryable=True)
      if not response.is_success:
          raise EmailSendError(f"Send failed: {response.text}", response.status_code, retryable=False)

      logger.info(f"Email sent to {to}: {subject}")
      return {"sent": True}

email_client = EmailClient()

For critical emails (password resets, receipts), add a retry wrapper:

# retry.py
import asyncio
from email_client import EmailSendError
 
async def with_retry(fn, max_retries: int = 3):
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except EmailSendError as e:
            if not e.retryable or attempt == max_retries:
                raise
            delay = 2 ** attempt
            await asyncio.sleep(delay)
 
# Usage
from emails.password_reset import send_password_reset
 
await with_retry(lambda: send_password_reset("user@example.com", token))

Rate Limiting with Middleware

Add basic rate limiting to prevent abuse on public-facing endpoints:

# middleware.py
import time
from collections import defaultdict
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
 
class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, max_per_minute: int = 10):
        super().__init__(app)
        self.max_per_minute = max_per_minute
        self.requests: dict[str, list[float]] = defaultdict(list)
 
    async def dispatch(self, request: Request, call_next):
        # Only rate limit email endpoints
        if not request.url.path.startswith("/api/send"):
            return await call_next(request)
 
        client_ip = request.client.host if request.client else "unknown"
        now = time.time()
        minute_ago = now - 60
 
        # Clean old entries and check limit
        self.requests[client_ip] = [
            t for t in self.requests[client_ip] if t > minute_ago
        ]
 
        if len(self.requests[client_ip]) >= self.max_per_minute:
            raise HTTPException(status_code= 429, detail="Too many requests")
 
        self.requests[client_ip].append(now)
        return await call_next(request)

# main.py
from middleware import RateLimitMiddleware
 
app = FastAPI(lifespan=lifespan)
app.add_middleware(RateLimitMiddleware, max_per_minute=10)

Going to Production

Before you start sending real emails, handle these things.

1. Verify Your Domain

Every email provider requires domain verification. This means adding DNS records (SPF, DKIM, and usually DMARC) to prove you own the domain you're sending from. See our email authentication setup guide for the full process.

Without this, your emails go straight to spam. No exceptions.

2. Use a Dedicated Sending Domain

Send from something like mail.yourapp.com instead of your root domain. If your email reputation takes a hit, your main domain stays clean.

3. Manage Settings Properly

Use pydantic-settings (shown above) for all configuration. It validates at startup, reads from .env files, and gives you type safety. Never use raw os.environ.get() for critical config.

4. Set Up Structured Logging

import logging
import json
 
class JSONFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
        })
 
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logging.basicConfig(handlers=[handler], level=logging.INFO)

5. Use Gunicorn with Uvicorn Workers

For production, run with Gunicorn as the process manager:

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

This gives you multiple worker processes for better throughput and reliability.

6. Health Check Endpoint

@app.get("/health")
async def health():
    return {"status": "ok"}

Beyond Transactional: Marketing Emails and Sequences

At some point, you'll want more than one-off transactional emails. You'll want to:

Send onboarding sequences that guide new users through your product over several days
Run marketing campaigns to announce features or share updates
Automate lifecycle emails based on what users do (or don't do) in your app
Track engagement to see which emails get opened and clicked

Most teams stitch together a transactional provider (Resend, SendGrid) with a separate marketing tool (Mailchimp, ConvertKit). That means two dashboards, two billing systems, and keeping subscriber lists in sync.

Sequenzy handles both from one platform. Same API, same dashboard. You get transactional sends, marketing campaigns, automated sequences, subscriber segments, and native Stripe integration for SaaS-specific automations like trial conversion and churn prevention.

Here's what subscriber management looks like with httpx:

# Add a subscriber when they sign up
await client.post(
    "/subscribers",
    json={
        "email": "user@example.com",
        "firstName": "Jane",
        "tags": ["signed-up"],
        "customAttributes": {"plan": "free", "source": "organic"},
    },
)
 
# Tag them when they upgrade
await client.post(
    "/subscribers/tags",
    json={"email": "user@example.com", "tag": "customer"},
)
 
# Track events to trigger automated sequences
await client.post(
    "/subscribers/events",
    json={
        "email": "user@example.com",
        "event": "onboarding.completed",
        "properties": {"completedSteps": 5},
    },
)

You set up sequences in the Sequenzy dashboard (onboarding drip, trial conversion, churn prevention), and the API triggers them based on what happens in your app.

FAQ

Should I use the Resend/SendGrid Python SDK or `httpx`?

Use httpx. The Resend Python SDK and SendGrid's sendgrid package are both synchronous. In a FastAPI async route, calling a sync SDK blocks the event loop, defeating the purpose of async. With httpx.AsyncClient, your email sends are truly non-blocking.

Can I use `smtplib` instead of an email API?

You can, but you shouldn't. smtplib is synchronous, doesn't handle deliverability (SPF/DKIM/DMARC), doesn't provide bounce handling, and doesn't retry on failure. Email APIs handle all of this for you. The only reason to use smtplib is if you're running your own mail server, which you almost certainly shouldn't be.

What about Celery for background email tasks?

FastAPI's BackgroundTasks is enough for most apps. It runs in the same process, requires no extra infrastructure, and handles async functions natively. Celery adds Redis/RabbitMQ as a dependency and adds operational complexity. Consider Celery only when you need: persistent task queues that survive restarts, task scheduling (cron-like), or when email volume is high enough that you want dedicated worker processes.

How do I test email sending?

Override the email client dependency in tests with a mock. FastAPI's app.dependency_overrides makes this clean. For integration tests, most providers have sandbox/test modes. You can also use Mailpit as a local catch-all SMTP server.

What's the difference between `BackgroundTasks` and `asyncio.create_task`?

BackgroundTasks runs after the response is sent and is tied to the request lifecycle. asyncio.create_task runs immediately in the event loop. For email, BackgroundTasks is the right choice because you want the response to return before the email sends.

How do I handle email bounces?

Set up a webhook endpoint to receive delivery notifications from your email provider. Each provider sends events for bounces, complaints, and deliveries. Parse the webhook, update your subscriber status, and stop sending to bounced addresses.

Can I send attachments?

Yes. With httpx, include the attachment as base64-encoded data in the JSON body (check your provider's API docs for the exact format). For large attachments, consider uploading to S3 first and including a download link in the email instead.

How do I send to multiple recipients?

For transactional emails (like a receipt), send one email per recipient. For marketing campaigns, use your provider's batch API. Never put multiple recipients in the to field - that exposes everyone's email addresses to each other.

Why `pydantic-settings` instead of `python-dotenv`?

pydantic-settings does everything python-dotenv does (reads .env files) plus validates types, sets defaults, and catches missing required variables at startup. With python-dotenv, a missing API key silently returns None and crashes at the first email send. With pydantic-settings, your app won't start if required config is missing.

How do I preview email templates during development?

Render the template and write it to a file, then open in a browser:

html = render_email("welcome.html", name="Alice", login_url="#")
with open("/tmp/preview.html", "w") as f:
    f.write(html)
# Open /tmp/preview.html in your browser

Or create a preview endpoint (only in development):

@app.get("/preview/{template_name}")
async def preview_email(template_name: str):
    html = render_email(f"{template_name}.html", name="Alice", login_url="#")
    return HTMLResponse(html)

Wrapping Up

Here's what we covered:

Async email sending with httpx.AsyncClient for non-blocking HTTP calls
Pydantic settings for validated, type-safe configuration
Connection pooling with managed client lifecycle via lifespan
Jinja2 templates with inheritance for maintainable email HTML
BackgroundTasks for sending emails without blocking responses
Dependency injection for clean, testable email routes
Error handling with custom exceptions and retries
Stripe webhooks with hmac signature verification
Production checklist: domain verification, structured logging, Gunicorn workers

The code in this guide is production-ready. Copy the patterns that fit your app, swap in your provider of choice, and start sending.

Frequently Asked Questions

Should I use FastAPI's async capabilities for email sending?

Yes. Use async email-sending functions with httpx.AsyncClient or an async-compatible SDK. This prevents email sends from blocking your event loop, so other requests can be processed while waiting for the email API response.

How do I send emails in the background in FastAPI?

Use FastAPI's BackgroundTasks for simple fire-and-forget emails. For production workloads, use Celery, ARQ, or SAQ for proper retry logic and monitoring. BackgroundTasks is convenient but doesn't persist jobs if the server restarts.

How do I validate email addresses in FastAPI?

Use Pydantic models with EmailStr from pydantic[email]. Define your request body as a Pydantic model with email: EmailStr and FastAPI automatically validates the format before your handler runs. Return a 422 error for invalid addresses.

How do I handle email API errors in FastAPI?

Catch exceptions from your email SDK and raise HTTPException with appropriate status codes. Use FastAPI's exception handlers for consistent error responses. Log the full error details with structlog or Python's logging for debugging.

Can I use Jinja2 templates for emails in FastAPI?

Yes. Install jinja2 and create an Environment that loads templates from a directory. Render templates with dynamic variables before passing the HTML to your email provider. This separates email content from business logic.

How do I test FastAPI email endpoints?

Use httpx.AsyncClient with FastAPI's TestClient and mock the email SDK with unittest.mock.patch. Verify your endpoint returns the correct response and that the email function was called with expected parameters.

How do I rate limit email endpoints in FastAPI?

Use slowapi which integrates with FastAPI's dependency injection. Decorate your endpoints with rate limit rules like @limiter.limit("5/minute"). Rate limit by IP address for public endpoints and by user ID for authenticated ones.

Should I use FastAPI's dependency injection for the email client?

Yes. Create the email client as a dependency with Depends() so it's initialized once and reused across requests. This also makes testing easier since you can override the dependency with a mock in tests.

How do I send emails with file attachments in FastAPI?

Accept file uploads with UploadFile parameter, read the file content with await file.read(), base64-encode it, and pass it to your email SDK's attachment parameter. Validate file size and type in a dependency to reject oversized or disallowed files.

How do I configure different email providers for dev and production in FastAPI?

Use Pydantic Settings with environment-specific .env files. Create an EmailSettings class that reads the provider and API key from environment variables. In development, use a sandbox API key or a local email capture tool.

Pick an Email Provider

You have three solid options. Each code example below lets you switch between them.

Sequenzy is built for SaaS. Transactional emails, marketing campaigns, automated sequences, subscriber management, all from one API. If you're building a SaaS product, this is the simplest path because you won't need to glue together three different tools later. Native Stripe integration and built-in retries.
Resend is a developer-friendly transactional email API. Clean DX, good docs, solid deliverability. They have one-off broadcast campaigns but no automations or sequences.
SendGrid is the enterprise standard. Feature-rich, sometimes complex. Good if you need high volume and don't mind a bigger API surface.

FastAPI email choice	Best for	Why	Avoid
`httpx.AsyncClient`	All provider API calls in async routes	Does not block the event loop	Sync provider SDK calls inside routes
`BackgroundTasks`	Lightweight post-response sends	Simple built-in offload for small apps	Long-running or retry-heavy campaign work
External worker	Campaigns, retries, and webhook fanout	Durable processing and better observability	Treating BackgroundTasks as a queue
Pydantic settings	Provider keys and sender config	Startup validation catches missing secrets	Raw `os.environ` scattered through code
Provider automation	Lifecycle and billing sequences	Keeps timing/state out of FastAPI	Sending duplicate lifecycle emails from both systems

Install

Terminal

pip install fastapi uvicorn httpx pydantic-settings jinja2

Terminal

pip install fastapi uvicorn httpx pydantic-settings jinja2

Terminal

pip install fastapi uvicorn httpx pydantic-settings jinja2

Dependency	Role	Production note
`fastapi`	API routes, DI, webhook endpoints	Keep email side effects outside critical request paths
`httpx`	Async provider HTTP calls	Reuse one client with app lifespan
`pydantic-settings`	Typed config and `.env` loading	Fail fast when keys are missing
`jinja2`	HTML email templates	Render once, then pass HTML to provider
`uvicorn`	Local ASGI server	Use production ASGI deployment settings

Add your API key to .env:

.env

SEQUENZY_API_KEY=sq_your_api_key_here

.env

RESEND_API_KEY=re_your_api_key_here

.env

SENDGRID_API_KEY=SG.your_api_key_here

Configure Settings with Pydantic

Use pydantic-settings to manage configuration. It reads from environment variables and .env files with full type validation:

config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
  sequenzy_api_key: str
  app_url: str = "https://yourapp.com"

  model_config = {"env_file": ".env"}

settings = Settings()

config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
  resend_api_key: str
  from_email: str = "Your App <noreply@yourdomain.com>"
  app_url: str = "https://yourapp.com"

  model_config = {"env_file": ".env"}

settings = Settings()

config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
  sendgrid_api_key: str
  from_email: str = "noreply@yourdomain.com"
  app_url: str = "https://yourapp.com"

  model_config = {"env_file": ".env"}

settings = Settings()

This is better than raw os.environ because you get validation at startup - if SEQUENZY_API_KEY is missing, FastAPI won't even start instead of crashing on the first email send.

Create the Email Client

Create a reusable async email client. Use httpx.AsyncClient with a managed lifecycle so connections are properly pooled and cleaned up:

email_client.py

import httpx
from config import settings

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.sequenzy.com/v1",
          headers={"Authorization": f"Bearer {settings.sequenzy_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized. Call startup() first.")
      return self._client

  async def send(self, to: str, subject: str, body: str) -> dict:
      response = await self.client.post(
          "/transactional/send",
          json={"to": to, "subject": subject, "body": body},
      )
      response.raise_for_status()
      return response.json()

email_client = EmailClient()

email_client.py

import httpx
from config import settings

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.resend.com",
          headers={"Authorization": f"Bearer {settings.resend_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized. Call startup() first.")
      return self._client

  async def send(self, to: str, subject: str, html: str) -> dict:
      response = await self.client.post(
          "/emails",
          json={
              "from": settings.from_email,
              "to": to,
              "subject": subject,
              "html": html,
          },
      )
      response.raise_for_status()
      return response.json()

email_client = EmailClient()

email_client.py

import httpx
from config import settings

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.sendgrid.com/v3",
          headers={"Authorization": f"Bearer {settings.sendgrid_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized. Call startup() first.")
      return self._client

  async def send(self, to: str, subject: str, html: str) -> dict:
      response = await self.client.post(
          "/mail/send",
          json={
              "personalizations": [{"to": [{"email": to}]}],
              "from": {"email": settings.from_email},
              "subject": subject,
              "content": [{"type": "text/html", "value": html}],
          },
      )
      response.raise_for_status()
      return {"sent": True}

email_client = EmailClient()

Key things happening here:

Connection pooling: httpx.AsyncClient reuses TCP connections across requests. Creating a new client per request wastes time on TLS handshakes.
Lifecycle management: The client opens on startup and closes on shutdown. No leaked connections.
Timeout: 30 seconds is generous but prevents hanging forever if the provider is down.

Wire Up the App Lifecycle

Use FastAPI's lifespan context manager to start and stop the email client:

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from email_client import email_client
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    await email_client.startup()
    yield
    await email_client.shutdown()
 
app = FastAPI(lifespan=lifespan)

This ensures the httpx.AsyncClient is properly initialized before your app handles requests and properly closed when the app shuts down.

Send Your First Email

The simplest possible email. Pydantic validates the request body, FastAPI handles the rest.

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from email_client import email_client

@asynccontextmanager
async def lifespan(app: FastAPI):
  await email_client.startup()
  yield
  await email_client.shutdown()

app = FastAPI(lifespan=lifespan)

class SendRequest(BaseModel):
  email: EmailStr
  name: str

@app.post("/api/send")
async def send_welcome(req: SendRequest):
  try:
      result = await email_client.send(
          to=req.email,
          subject=f"Welcome, {req.name}",
          body=f"<h1>Welcome, {req.name}</h1><p>Your account is ready.</p>",
      )
      return {"jobId": result["jobId"]}
  except Exception as e:
      raise HTTPException(status_code=500, detail=str(e))

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from email_client import email_client

@asynccontextmanager
async def lifespan(app: FastAPI):
  await email_client.startup()
  yield
  await email_client.shutdown()

app = FastAPI(lifespan=lifespan)

class SendRequest(BaseModel):
  email: EmailStr
  name: str

@app.post("/api/send")
async def send_welcome(req: SendRequest):
  try:
      result = await email_client.send(
          to=req.email,
          subject=f"Welcome, {req.name}",
          html=f"<h1>Welcome, {req.name}</h1><p>Your account is ready.</p>",
      )
      return {"id": result["id"]}
  except Exception as e:
      raise HTTPException(status_code=500, detail=str(e))

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from email_client import email_client

@asynccontextmanager
async def lifespan(app: FastAPI):
  await email_client.startup()
  yield
  await email_client.shutdown()

app = FastAPI(lifespan=lifespan)

class SendRequest(BaseModel):
  email: EmailStr
  name: str

@app.post("/api/send")
async def send_welcome(req: SendRequest):
  try:
      await email_client.send(
          to=req.email,
          subject=f"Welcome, {req.name}",
          html=f"<h1>Welcome, {req.name}</h1><p>Your account is ready.</p>",
      )
      return {"sent": True}
  except Exception as e:
      raise HTTPException(status_code=500, detail=str(e))

Test it:

uvicorn main:app --reload
curl -X POST http://localhost:8000/api/send \
  -H "Content-Type: application/json" \
  -d '{"email": "user@example.com", "name": "Alice"}'

Pydantic's EmailStr validates the email format before your code even runs. Invalid emails get a 422 response automatically.

Build Email Templates with Jinja2

Raw f-strings get messy fast. Jinja2 gives you inheritance, conditionals, loops, and auto-escaping:

project/
├── templates/
│   └── emails/
│       ├── base.html
│       ├── welcome.html
│       ├── password-reset.html
│       └── receipt.html
├── email_client.py
├── email_renderer.py
├── config.py
└── main.py

Start with a base template:

<!-- templates/emails/base.html -->
<!DOCTYPE html>
<html>
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body style="margin:0;padding:0;background-color:#f6f9fc;font-family:sans-serif;">
  <div style="max-width:480px;margin:40px auto;background:#fff;padding:40px;border-radius:8px;">
    {% block content %}{% endblock %}
    <hr style="border:none;border-top:1px solid #e5e7eb;margin:32px 0 16px;" />
    <p style="font-size:12px;color:#9ca3af;">
      &copy; {{ year }} Your App. All rights reserved.
    </p>
  </div>
</body>
</html>

Then extend it for each email type:

<!-- templates/emails/welcome.html -->
{% extends "base.html" %}
 
{% block content %}
<h1 style="font-size:24px;margin-bottom:16px;">Welcome, {{ name }}</h1>
<p style="font-size:16px;line-height:1.6;color:#374151;">
  Your account is ready. You can log in and start exploring.
</p>
<a href="{{ login_url }}"
   style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;font-size:14px;font-weight:600;margin-top:16px;">
  Go to Dashboard
</a>
{% endblock %}

Create a renderer that loads and renders templates:

# email_renderer.py
from datetime import datetime
from jinja2 import Environment, FileSystemLoader, select_autoescape
 
env = Environment(
    loader=FileSystemLoader("templates/emails"),
    autoescape=select_autoescape(["html"]),
)
 
def render_email(template_name: str, **kwargs) -> str:
    template = env.get_template(template_name)
    return template.render(year=datetime.now().year, **kwargs)

Use it in your routes:

main.py

from email_renderer import render_email

@app.post("/api/send-welcome")
async def send_welcome(req: SendRequest):
  html = render_email("welcome.html", name=req.name, login_url="https://app.yoursite.com")

  result = await email_client.send(
      to=req.email,
      subject=f"Welcome, {req.name}",
      body=html,
  )
  return {"jobId": result["jobId"]}

main.py

from email_renderer import render_email

@app.post("/api/send-welcome")
async def send_welcome(req: SendRequest):
  html = render_email("welcome.html", name=req.name, login_url="https://app.yoursite.com")

  result = await email_client.send(
      to=req.email,
      subject=f"Welcome, {req.name}",
      html=html,
  )
  return {"id": result["id"]}

main.py

from email_renderer import render_email

@app.post("/api/send-welcome")
async def send_welcome(req: SendRequest):
  html = render_email("welcome.html", name=req.name, login_url="https://app.yoursite.com")

  await email_client.send(
      to=req.email,
      subject=f"Welcome, {req.name}",
      html=html,
  )
  return {"sent": True}

Background Tasks

FastAPI's built-in BackgroundTasks lets you return a response immediately and send the email after. The email runs in the same process, after the response is sent:

main.py

from fastapi import BackgroundTasks
from email_renderer import render_email
import logging

logger = logging.getLogger(__name__)

async def send_welcome_email(email: str, name: str):
  try:
      html = render_email("welcome.html", name=name, login_url="https://app.yoursite.com")
      await email_client.send(to=email, subject=f"Welcome, {name}", body=html)
      logger.info(f"Welcome email sent to {email}")
  except Exception:
      logger.exception(f"Failed to send welcome email to {email}")

@app.post("/api/signup")
async def signup(req: SendRequest, background_tasks: BackgroundTasks):
  # Create user in database first...

  background_tasks.add_task(send_welcome_email, req.email, req.name)

  return {"message": "Account created"}

main.py

from fastapi import BackgroundTasks
from email_renderer import render_email
import logging

logger = logging.getLogger(__name__)

async def send_welcome_email(email: str, name: str):
  try:
      html = render_email("welcome.html", name=name, login_url="https://app.yoursite.com")
      await email_client.send(to=email, subject=f"Welcome, {name}", html=html)
      logger.info(f"Welcome email sent to {email}")
  except Exception:
      logger.exception(f"Failed to send welcome email to {email}")

@app.post("/api/signup")
async def signup(req: SendRequest, background_tasks: BackgroundTasks):
  # Create user in database first...

  background_tasks.add_task(send_welcome_email, req.email, req.name)

  return {"message": "Account created"}

main.py

from fastapi import BackgroundTasks
from email_renderer import render_email
import logging

logger = logging.getLogger(__name__)

async def send_welcome_email(email: str, name: str):
  try:
      html = render_email("welcome.html", name=name, login_url="https://app.yoursite.com")
      await email_client.send(to=email, subject=f"Welcome, {name}", html=html)
      logger.info(f"Welcome email sent to {email}")
  except Exception:
      logger.exception(f"Failed to send welcome email to {email}")

@app.post("/api/signup")
async def signup(req: SendRequest, background_tasks: BackgroundTasks):
  # Create user in database first...

  background_tasks.add_task(send_welcome_email, req.email, req.name)

  return {"message": "Account created"}

The BackgroundTasks parameter is injected by FastAPI automatically. The response returns immediately, and the email sends after. If the email fails, it logs the error but doesn't crash the request.

Dependency Injection for Email

FastAPI's Depends system lets you inject the email client into routes cleanly. This is especially useful if you want to swap the client in tests:

# dependencies.py
from email_client import email_client, EmailClient
 
async def get_email_client() -> EmailClient:
    return email_client

# routes/emails.py
from fastapi import APIRouter, Depends, BackgroundTasks
from pydantic import BaseModel, EmailStr
from dependencies import get_email_client
from email_client import EmailClient
 
router = APIRouter(prefix="/api")
 
class ContactRequest(BaseModel):
    email: EmailStr
    name: str
    message: str
 
@router.post("/contact")
async def contact(
    req: ContactRequest,
    background_tasks: BackgroundTasks,
    email: EmailClient = Depends(get_email_client),
):
    async def send():
        await email.send(
            to="you@yourcompany.com",
            subject=f"Contact from {req.name}",
            body=f"<p><strong>From:</strong> {req.email}</p><p>{req.message}</p>",
        )
 
    background_tasks.add_task(send)
    return {"message": "Message received"}

In tests, you override the dependency:

from fastapi.testclient import TestClient
from unittest.mock import AsyncMock
 
mock_client = AsyncMock()
app.dependency_overrides[get_email_client] = lambda: mock_client
 
client = TestClient(app)
response = client.post("/api/contact", json={...})
mock_client.send.assert_called_once()

Common Email Patterns for SaaS

Here are the emails almost every SaaS app needs, with production-ready implementations.

Password Reset

emails/password_reset.py

from email_client import email_client
from email_renderer import render_email
from config import settings

async def send_password_reset(email: str, reset_token: str):
  reset_url = f"{settings.app_url}/reset-password?token={reset_token}"
  html = render_email("password-reset.html", reset_url=reset_url)

  await email_client.send(
      to=email,
      subject="Reset your password",
      body=html,
  )

emails/password_reset.py

from email_client import email_client
from email_renderer import render_email
from config import settings

async def send_password_reset(email: str, reset_token: str):
  reset_url = f"{settings.app_url}/reset-password?token={reset_token}"
  html = render_email("password-reset.html", reset_url=reset_url)

  await email_client.send(
      to=email,
      subject="Reset your password",
      html=html,
  )

emails/password_reset.py

from email_client import email_client
from email_renderer import render_email
from config import settings

async def send_password_reset(email: str, reset_token: str):
  reset_url = f"{settings.app_url}/reset-password?token={reset_token}"
  html = render_email("password-reset.html", reset_url=reset_url)

  await email_client.send(
      to=email,
      subject="Reset your password",
      html=html,
  )

The template:

<!-- templates/emails/password-reset.html -->
{% extends "base.html" %}
 
{% block content %}
<h2 style="font-size:20px;margin-bottom:16px;">Password Reset</h2>
<p style="font-size:16px;line-height:1.6;color:#374151;">
  Click the button below to reset your password. This link expires in 1 hour.
</p>
<a href="{{ reset_url }}"
   style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;font-weight:600;">
  Reset Password
</a>
<p style="color:#6b7280;font-size:14px;margin-top:24px;">
  If you didn't request this, ignore this email.
</p>
{% endblock %}

Payment Receipt

emails/receipt.py

from email_client import email_client
from email_renderer import render_email

async def send_receipt(email: str, amount: int, plan: str, invoice_url: str):
  formatted = f"${amount / 100:.2f}"
  html = render_email(
      "receipt.html",
      amount=formatted,
      plan=plan,
      invoice_url=invoice_url,
  )

  await email_client.send(
      to=email,
      subject=f"Payment receipt - {formatted}",
      body=html,
  )

emails/receipt.py

from email_client import email_client
from email_renderer import render_email

async def send_receipt(email: str, amount: int, plan: str, invoice_url: str):
  formatted = f"${amount / 100:.2f}"
  html = render_email(
      "receipt.html",
      amount=formatted,
      plan=plan,
      invoice_url=invoice_url,
  )

  await email_client.send(
      to=email,
      subject=f"Payment receipt - {formatted}",
      html=html,
  )

emails/receipt.py

from email_client import email_client
from email_renderer import render_email

async def send_receipt(email: str, amount: int, plan: str, invoice_url: str):
  formatted = f"${amount / 100:.2f}"
  html = render_email(
      "receipt.html",
      amount=formatted,
      plan=plan,
      invoice_url=invoice_url,
  )

  await email_client.send(
      to=email,
      subject=f"Payment receipt - {formatted}",
      html=html,
  )

<!-- templates/emails/receipt.html -->
{% extends "base.html" %}
 
{% block content %}
<h2 style="font-size:20px;margin-bottom:16px;">Payment Received</h2>
<p style="font-size:16px;color:#374151;">Thanks for your payment. Here's your receipt:</p>
<table style="width:100%;border-collapse:collapse;margin:16px 0;">
  <tr>
    <td style="padding:8px;border-bottom:1px solid #e5e7eb;">Plan</td>
    <td style="padding:8px;border-bottom:1px solid #e5e7eb;text-align:right;">{{ plan }}</td>
  </tr>
  <tr>
    <td style="padding:8px;font-weight:600;">Total</td>
    <td style="padding:8px;text-align:right;font-weight:600;">{{ amount }}</td>
  </tr>
</table>
<a href="{{ invoice_url }}" style="color:#f97316;">View full invoice</a>
{% endblock %}

Stripe Webhook Handler

routes/stripe_webhook.py

import hmac
import hashlib
import time
from fastapi import APIRouter, Request, HTTPException, BackgroundTasks
from emails.receipt import send_receipt
from email_client import email_client
from config import settings

router = APIRouter()

def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
  parts = dict(item.split("=", 1) for item in signature.split(","))
  timestamp = parts.get("t", "")
  expected_sig = parts.get("v1", "")

  if not timestamp or not expected_sig:
      return False

  # Check timestamp is within 5 minutes
  if abs(time.time() - int(timestamp)) > 300:
      return False

  signed_payload = f"{timestamp}.{payload.decode()}"
  computed = hmac.new(
      secret.encode(),
      signed_payload.encode(),
      hashlib.sha256,
  ).hexdigest()

  return hmac.compare_digest(computed, expected_sig)

@router.post("/api/webhooks/stripe")
async def stripe_webhook(request: Request, background_tasks: BackgroundTasks):
  body = await request.body()
  signature = request.headers.get("stripe-signature", "")

  if not verify_stripe_signature(body, signature, settings.stripe_webhook_secret):
      raise HTTPException(status_code=400, detail="Invalid signature")

  event = await request.json()

  if event["type"] == "checkout.session.completed":
      session = event["data"]["object"]

      background_tasks.add_task(
          send_receipt,
          email=session["customer_email"],
          amount=session["amount_total"],
          plan=session.get("metadata", {}).get("plan", "Pro"),
          invoice_url=session.get("invoice_url", f"{settings.app_url}/billing"),
      )

  if event["type"] == "invoice.payment_failed":
      invoice = event["data"]["object"]

      async def send_failed_payment():
          await email_client.send(
              to=invoice["customer_email"],
              subject="Payment failed - action needed",
              body=f"""
                  <h2>Payment Failed</h2>
                  <p>We couldn't process your payment. Please update your billing info.</p>
                  <a href="{settings.app_url}/billing"
                     style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;">
                    Update Billing
                  </a>
              """,
          )

      background_tasks.add_task(send_failed_payment)

  return {"received": True}

routes/stripe_webhook.py

import hmac
import hashlib
import time
from fastapi import APIRouter, Request, HTTPException, BackgroundTasks
from emails.receipt import send_receipt
from email_client import email_client
from config import settings

router = APIRouter()

def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
  parts = dict(item.split("=", 1) for item in signature.split(","))
  timestamp = parts.get("t", "")
  expected_sig = parts.get("v1", "")

  if not timestamp or not expected_sig:
      return False

  if abs(time.time() - int(timestamp)) > 300:
      return False

  signed_payload = f"{timestamp}.{payload.decode()}"
  computed = hmac.new(
      secret.encode(),
      signed_payload.encode(),
      hashlib.sha256,
  ).hexdigest()

  return hmac.compare_digest(computed, expected_sig)

@router.post("/api/webhooks/stripe")
async def stripe_webhook(request: Request, background_tasks: BackgroundTasks):
  body = await request.body()
  signature = request.headers.get("stripe-signature", "")

  if not verify_stripe_signature(body, signature, settings.stripe_webhook_secret):
      raise HTTPException(status_code=400, detail="Invalid signature")

  event = await request.json()

  if event["type"] == "checkout.session.completed":
      session = event["data"]["object"]

      background_tasks.add_task(
          send_receipt,
          email=session["customer_email"],
          amount=session["amount_total"],
          plan=session.get("metadata", {}).get("plan", "Pro"),
          invoice_url=session.get("invoice_url", f"{settings.app_url}/billing"),
      )

  if event["type"] == "invoice.payment_failed":
      invoice = event["data"]["object"]

      async def send_failed_payment():
          await email_client.send(
              to=invoice["customer_email"],
              subject="Payment failed - action needed",
              html=f"""
                  <h2>Payment Failed</h2>
                  <p>We couldn't process your payment. Please update your billing info.</p>
                  <a href="{settings.app_url}/billing"
                     style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;">
                    Update Billing
                  </a>
              """,
          )

      background_tasks.add_task(send_failed_payment)

  return {"received": True}

routes/stripe_webhook.py

import hmac
import hashlib
import time
from fastapi import APIRouter, Request, HTTPException, BackgroundTasks
from emails.receipt import send_receipt
from email_client import email_client
from config import settings

router = APIRouter()

def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
  parts = dict(item.split("=", 1) for item in signature.split(","))
  timestamp = parts.get("t", "")
  expected_sig = parts.get("v1", "")

  if not timestamp or not expected_sig:
      return False

  if abs(time.time() - int(timestamp)) > 300:
      return False

  signed_payload = f"{timestamp}.{payload.decode()}"
  computed = hmac.new(
      secret.encode(),
      signed_payload.encode(),
      hashlib.sha256,
  ).hexdigest()

  return hmac.compare_digest(computed, expected_sig)

@router.post("/api/webhooks/stripe")
async def stripe_webhook(request: Request, background_tasks: BackgroundTasks):
  body = await request.body()
  signature = request.headers.get("stripe-signature", "")

  if not verify_stripe_signature(body, signature, settings.stripe_webhook_secret):
      raise HTTPException(status_code=400, detail="Invalid signature")

  event = await request.json()

  if event["type"] == "checkout.session.completed":
      session = event["data"]["object"]

      background_tasks.add_task(
          send_receipt,
          email=session["customer_email"],
          amount=session["amount_total"],
          plan=session.get("metadata", {}).get("plan", "Pro"),
          invoice_url=session.get("invoice_url", f"{settings.app_url}/billing"),
      )

  if event["type"] == "invoice.payment_failed":
      invoice = event["data"]["object"]

      async def send_failed_payment():
          await email_client.send(
              to=invoice["customer_email"],
              subject="Payment failed - action needed",
              html=f"""
                  <h2>Payment Failed</h2>
                  <p>We couldn't process your payment. Please update your billing info.</p>
                  <a href="{settings.app_url}/billing"
                     style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;">
                    Update Billing
                  </a>
              """,
          )

      background_tasks.add_task(send_failed_payment)

  return {"received": True}

Key things: we use request.body() to get the raw bytes for signature verification, then request.json() for the parsed event. The hmac.compare_digest function prevents timing attacks.

Error Handling

Emails fail. Networks time out. Rate limits hit. Here's how to handle it properly.

email_client.py

import httpx
import logging
from config import settings

logger = logging.getLogger(__name__)

class EmailSendError(Exception):
  def __init__(self, message: str, status: int, retryable: bool):
      super().__init__(message)
      self.status = status
      self.retryable = retryable

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.sequenzy.com/v1",
          headers={"Authorization": f"Bearer {settings.sequenzy_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized")
      return self._client

  async def send(self, to: str, subject: str, body: str) -> dict:
      try:
          response = await self.client.post(
              "/transactional/send",
              json={"to": to, "subject": subject, "body": body},
          )
      except httpx.TimeoutException:
          raise EmailSendError("Request timed out", 0, retryable=True)
      except httpx.ConnectError:
          raise EmailSendError("Connection failed", 0, retryable=True)

      if response.status_code == 429:
          raise EmailSendError("Rate limited", 429, retryable=True)
      if response.status_code == 401:
          raise EmailSendError("Invalid API key", 401, retryable=False)
      if response.status_code >= 500:
          raise EmailSendError(f"Server error: {response.text}", response.status_code, retryable=True)
      if not response.is_success:
          raise EmailSendError(f"Send failed: {response.text}", response.status_code, retryable=False)

      logger.info(f"Email sent to {to}: {subject}")
      return response.json()

email_client = EmailClient()

email_client.py

import httpx
import logging
from config import settings

logger = logging.getLogger(__name__)

class EmailSendError(Exception):
  def __init__(self, message: str, status: int, retryable: bool):
      super().__init__(message)
      self.status = status
      self.retryable = retryable

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.resend.com",
          headers={"Authorization": f"Bearer {settings.resend_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized")
      return self._client

  async def send(self, to: str, subject: str, html: str) -> dict:
      try:
          response = await self.client.post(
              "/emails",
              json={
                  "from": settings.from_email,
                  "to": to,
                  "subject": subject,
                  "html": html,
              },
          )
      except httpx.TimeoutException:
          raise EmailSendError("Request timed out", 0, retryable=True)
      except httpx.ConnectError:
          raise EmailSendError("Connection failed", 0, retryable=True)

      if response.status_code == 429:
          raise EmailSendError("Rate limited", 429, retryable=True)
      if response.status_code in (401, 403):
          raise EmailSendError("Invalid API key", response.status_code, retryable=False)
      if response.status_code >= 500:
          raise EmailSendError(f"Server error: {response.text}", response.status_code, retryable=True)
      if not response.is_success:
          raise EmailSendError(f"Send failed: {response.text}", response.status_code, retryable=False)

      logger.info(f"Email sent to {to}: {subject}")
      return response.json()

email_client = EmailClient()

email_client.py

import httpx
import logging
from config import settings

logger = logging.getLogger(__name__)

class EmailSendError(Exception):
  def __init__(self, message: str, status: int, retryable: bool):
      super().__init__(message)
      self.status = status
      self.retryable = retryable

class EmailClient:
  def __init__(self):
      self._client: httpx.AsyncClient | None = None

  async def startup(self):
      self._client = httpx.AsyncClient(
          base_url="https://api.sendgrid.com/v3",
          headers={"Authorization": f"Bearer {settings.sendgrid_api_key}"},
          timeout=30.0,
      )

  async def shutdown(self):
      if self._client:
          await self._client.aclose()

  @property
  def client(self) -> httpx.AsyncClient:
      if not self._client:
          raise RuntimeError("Email client not initialized")
      return self._client

  async def send(self, to: str, subject: str, html: str) -> dict:
      try:
          response = await self.client.post(
              "/mail/send",
              json={
                  "personalizations": [{"to": [{"email": to}]}],
                  "from": {"email": settings.from_email},
                  "subject": subject,
                  "content": [{"type": "text/html", "value": html}],
              },
          )
      except httpx.TimeoutException:
          raise EmailSendError("Request timed out", 0, retryable=True)
      except httpx.ConnectError:
          raise EmailSendError("Connection failed", 0, retryable=True)

      if response.status_code == 429:
          raise EmailSendError("Rate limited", 429, retryable=True)
      if response.status_code in (401, 403):
          raise EmailSendError("Invalid API key", response.status_code, retryable=False)
      if response.status_code >= 500:
          raise EmailSendError(f"Server error: {response.text}", response.status_code, retryable=True)
      if not response.is_success:
          raise EmailSendError(f"Send failed: {response.text}", response.status_code, retryable=False)

      logger.info(f"Email sent to {to}: {subject}")
      return {"sent": True}

email_client = EmailClient()

For critical emails (password resets, receipts), add a retry wrapper:

# retry.py
import asyncio
from email_client import EmailSendError
 
async def with_retry(fn, max_retries: int = 3):
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except EmailSendError as e:
            if not e.retryable or attempt == max_retries:
                raise
            delay = 2 ** attempt
            await asyncio.sleep(delay)
 
# Usage
from emails.password_reset import send_password_reset
 
await with_retry(lambda: send_password_reset("user@example.com", token))

Rate Limiting with Middleware

Add basic rate limiting to prevent abuse on public-facing endpoints:

# middleware.py
import time
from collections import defaultdict
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
 
class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, max_per_minute: int = 10):
        super().__init__(app)
        self.max_per_minute = max_per_minute
        self.requests: dict[str, list[float]] = defaultdict(list)
 
    async def dispatch(self, request: Request, call_next):
        # Only rate limit email endpoints
        if not request.url.path.startswith("/api/send"):
            return await call_next(request)
 
        client_ip = request.client.host if request.client else "unknown"
        now = time.time()
        minute_ago = now - 60
 
        # Clean old entries and check limit
        self.requests[client_ip] = [
            t for t in self.requests[client_ip] if t > minute_ago
        ]
 
        if len(self.requests[client_ip]) >= self.max_per_minute:
            raise HTTPException(status_code= 429, detail="Too many requests")
 
        self.requests[client_ip].append(now)
        return await call_next(request)

# main.py
from middleware import RateLimitMiddleware
 
app = FastAPI(lifespan=lifespan)
app.add_middleware(RateLimitMiddleware, max_per_minute=10)

Going to Production

Before you start sending real emails, handle these things.

1. Verify Your Domain

Without this, your emails go straight to spam. No exceptions.

2. Use a Dedicated Sending Domain

Send from something like mail.yourapp.com instead of your root domain. If your email reputation takes a hit, your main domain stays clean.

3. Manage Settings Properly

Use pydantic-settings (shown above) for all configuration. It validates at startup, reads from .env files, and gives you type safety. Never use raw os.environ.get() for critical config.

4. Set Up Structured Logging

import logging
import json
 
class JSONFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
        })
 
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logging.basicConfig(handlers=[handler], level=logging.INFO)

5. Use Gunicorn with Uvicorn Workers

For production, run with Gunicorn as the process manager:

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

This gives you multiple worker processes for better throughput and reliability.

6. Health Check Endpoint

@app.get("/health")
async def health():
    return {"status": "ok"}

Beyond Transactional: Marketing Emails and Sequences

At some point, you'll want more than one-off transactional emails. You'll want to:

Send onboarding sequences that guide new users through your product over several days
Run marketing campaigns to announce features or share updates
Automate lifecycle emails based on what users do (or don't do) in your app
Track engagement to see which emails get opened and clicked

Here's what subscriber management looks like with httpx:

# Add a subscriber when they sign up
await client.post(
    "/subscribers",
    json={
        "email": "user@example.com",
        "firstName": "Jane",
        "tags": ["signed-up"],
        "customAttributes": {"plan": "free", "source": "organic"},
    },
)
 
# Tag them when they upgrade
await client.post(
    "/subscribers/tags",
    json={"email": "user@example.com", "tag": "customer"},
)
 
# Track events to trigger automated sequences
await client.post(
    "/subscribers/events",
    json={
        "email": "user@example.com",
        "event": "onboarding.completed",
        "properties": {"completedSteps": 5},
    },
)

You set up sequences in the Sequenzy dashboard (onboarding drip, trial conversion, churn prevention), and the API triggers them based on what happens in your app.

FAQ

Should I use the Resend/SendGrid Python SDK or `httpx`?

Can I use `smtplib` instead of an email API?

What about Celery for background email tasks?

How do I test email sending?

What's the difference between `BackgroundTasks` and `asyncio.create_task`?

How do I handle email bounces?

Can I send attachments?

How do I send to multiple recipients?

Why `pydantic-settings` instead of `python-dotenv`?

How do I preview email templates during development?

Render the template and write it to a file, then open in a browser:

html = render_email("welcome.html", name="Alice", login_url="#")
with open("/tmp/preview.html", "w") as f:
    f.write(html)
# Open /tmp/preview.html in your browser

Or create a preview endpoint (only in development):

@app.get("/preview/{template_name}")
async def preview_email(template_name: str):
    html = render_email(f"{template_name}.html", name="Alice", login_url="#")
    return HTMLResponse(html)

Wrapping Up

Here's what we covered:

Async email sending with httpx.AsyncClient for non-blocking HTTP calls
Pydantic settings for validated, type-safe configuration
Connection pooling with managed client lifecycle via lifespan
Jinja2 templates with inheritance for maintainable email HTML
BackgroundTasks for sending emails without blocking responses
Dependency injection for clean, testable email routes
Error handling with custom exceptions and retries
Stripe webhooks with hmac signature verification
Production checklist: domain verification, structured logging, Gunicorn workers

The code in this guide is production-ready. Copy the patterns that fit your app, swap in your provider of choice, and start sending.

Pick an Email Provider

Install

Configure Settings with Pydantic

Create the Email Client

Wire Up the App Lifecycle

Send Your First Email

Build Email Templates with Jinja2

Background Tasks

Dependency Injection for Email

Common Email Patterns for SaaS

Password Reset

Payment Receipt

Stripe Webhook Handler

Error Handling

Rate Limiting with Middleware

Going to Production

1. Verify Your Domain

2. Use a Dedicated Sending Domain

3. Manage Settings Properly

4. Set Up Structured Logging

5. Use Gunicorn with Uvicorn Workers

6. Health Check Endpoint

Beyond Transactional: Marketing Emails and Sequences

FAQ

Should I use the Resend/SendGrid Python SDK or httpx?

Can I use smtplib instead of an email API?

What about Celery for background email tasks?

How do I test email sending?

What's the difference between BackgroundTasks and asyncio.create_task?

How do I handle email bounces?

Can I send attachments?

How do I send to multiple recipients?

Why pydantic-settings instead of python-dotenv?

How do I preview email templates during development?

Wrapping Up

Frequently Asked Questions

Should I use FastAPI's async capabilities for email sending?

How do I send emails in the background in FastAPI?

How do I validate email addresses in FastAPI?

How do I handle email API errors in FastAPI?

Can I use Jinja2 templates for emails in FastAPI?

How do I test FastAPI email endpoints?

How do I rate limit email endpoints in FastAPI?

Should I use FastAPI's dependency injection for the email client?

How do I send emails with file attachments in FastAPI?

How do I configure different email providers for dev and production in FastAPI?

Pick an Email Provider

Install

Configure Settings with Pydantic

Create the Email Client

Wire Up the App Lifecycle

Send Your First Email

Build Email Templates with Jinja2

Background Tasks

Dependency Injection for Email

Common Email Patterns for SaaS

Password Reset

Payment Receipt

Stripe Webhook Handler

Error Handling

Rate Limiting with Middleware

Going to Production

1. Verify Your Domain

2. Use a Dedicated Sending Domain

3. Manage Settings Properly

4. Set Up Structured Logging

5. Use Gunicorn with Uvicorn Workers

6. Health Check Endpoint

Beyond Transactional: Marketing Emails and Sequences

FAQ

Should I use the Resend/SendGrid Python SDK or httpx?

Can I use smtplib instead of an email API?

What about Celery for background email tasks?

How do I test email sending?

What's the difference between BackgroundTasks and asyncio.create_task?

How do I handle email bounces?

Can I send attachments?

How do I send to multiple recipients?

Why pydantic-settings instead of python-dotenv?

How do I preview email templates during development?

Should I use the Resend/SendGrid Python SDK or `httpx`?

Can I use `smtplib` instead of an email API?

What's the difference between `BackgroundTasks` and `asyncio.create_task`?

Why `pydantic-settings` instead of `python-dotenv`?

Should I use the Resend/SendGrid Python SDK or `httpx`?

Can I use `smtplib` instead of an email API?

What's the difference between `BackgroundTasks` and `asyncio.create_task`?

Why `pydantic-settings` instead of `python-dotenv`?