# Send email from Flask

Send transactional email from Flask with emails.sh. A complete app with a send route, a background thread, and Jinja email templates.

Flask has no mail layer of its own, so you are wiring this yourself and there is less to fight. This page gives you a full app.py with a send helper, a route that uses it, and the Jinja template it renders. It runs as written with two files.

## Setup

1. **Create a key** https://emails.sh/dashboard issues a key starting with esh_.
1. **Load it into config** app.config.from_prefixed_env() reads every FLASK_* variable, or read os.environ directly. Keep it out of the module you import from a template context.
1. **Add the send helper** One function that takes to, subject, and html. Every route calls it, so the endpoint and the auth header live in exactly one place.
1. **Render bodies with Jinja** render_template("email/welcome.html", name=name) gives you autoescaping, so a user whose name contains markup cannot inject it into your email.

## Install

```bash
pip install emailssh
```

## .env

```bash
EMAILSSH_API_KEY=esh_your_key_here
EMAILSSH_FROM="Acme <hello@acme.com>"
```

## The app

app.py (Flask 3.x)

```py
import os
import threading

import requests
from flask import Flask, jsonify, render_template, request

app = Flask(__name__)

API_KEY = os.environ["EMAILSSH_API_KEY"]
FROM = os.environ.get("EMAILSSH_FROM", "Acme <onboarding@emails.sh>")
ENDPOINT = "https://emails.sh/v1/emails"


def send_email(to, subject, html, text=None, idempotency_key=None):
    """One place that knows the endpoint, the header, and the timeout."""
    payload = {
        "from": FROM,
        "to": [to],
        "subject": subject,
        "html": html,
    }
    if text:
        payload["text"] = text
    if idempotency_key:
        payload["idempotency_key"] = idempotency_key

    response = requests.post(
        ENDPOINT,
        json=payload,
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=15,
    )
    response.raise_for_status()
    return response.json()


def send_in_background(app, **kwargs):
    """A thread is enough for one small app. Anything with real volume wants a
    queue (Celery, RQ, Dramatiq) so a restart does not drop pending sends."""

    def run():
        with app.app_context():
            try:
                send_email(**kwargs)
            except Exception as error:
                app.logger.exception("emails.sh send failed: %s", error)

    threading.Thread(target=run, daemon=True).start()


@app.post("/signup")
def signup():
    data = request.get_json(silent=True) or request.form
    email = (data.get("email") or "").strip()
    name = (data.get("name") or "there").strip()

    if "@" not in email:
        return jsonify({"error": "A valid email is required"}), 400

    html = render_template("email/welcome.html", name=name)
    send_in_background(
        app,
        to=email,
        subject="Welcome to Acme",
        html=html,
        text=f"Hi {name}, your Acme account is ready. Sign in at https://acme.com/login",
        idempotency_key=f"welcome:{email.lower()}",
    )

    return jsonify({"ok": True})


if __name__ == "__main__":
    app.run(debug=True)
```

## The template

templates/email/welcome.html

```
<!doctype html>
<html>
	<body style="font-family: system-ui, sans-serif; line-height: 1.5; color: #111">
		<p>Hi {{ name }},</p>
		<p>Your Acme account is ready.</p>
		<p><a href="https://acme.com/login">Sign in</a></p>
		<p style="color: #666; font-size: 13px">
			You are receiving this because you created an account at acme.com.
		</p>
	</body>
</html>
```

## The webhook receiver

webhooks.py, mounted with app.register_blueprint(webhooks)

```py
import hashlib
import hmac
import os

from flask import Blueprint, current_app, request

webhooks = Blueprint("webhooks", __name__)
SECRET = os.environ["EMAILSSH_WEBHOOK_SECRET"].encode()


@webhooks.post("/webhooks/emailssh")
def emailssh_events():
    signature = request.headers.get("x-emailssh-signature", "")
    expected = hmac.new(SECRET, request.get_data(), hashlib.sha256).hexdigest()

    # compare_digest, not ==, so the check does not leak the secret through
    # how long it takes to fail.
    if not hmac.compare_digest(expected, signature):
        return {"error": "bad signature"}, 401

    event = request.get_json()
    kind = event.get("type")

    if kind in ("email.bounced", "email.complained"):
        current_app.logger.warning("Suppress %s after %s", event["data"]["to"][0], kind)

    return "", 204
```

## Worth knowing

### requests.post with no timeout can hang forever

Without timeout=, a stalled connection holds the worker thread until the OS gives up, which under gunicorn means one fewer worker for minutes. Always pass a timeout.

### A thread is a starting point, not a queue

A daemon thread dies when the process restarts, so a deploy mid-send loses that email. Once sends matter, move to Celery or RQ where the job survives a restart.

### Render with Jinja, do not f-string user input into HTML

render_template autoescapes, so a name of <script>alert(1)</script> arrives as text. An f-string does not, and that markup ends up in your email.

### raise_for_status is what tells you it failed

The API returns 422 invalid_from_domain as a normal HTTP response. Without raise_for_status the call looks successful and nothing arrives.


## Questions

### How do I send an email in Flask without Flask-Mail?

One requests.post to https://emails.sh/v1/emails with a Bearer token, as in send_email above. Flask-Mail exists to configure SMTP, which you no longer need.

### How do I stop the send from blocking the response?

The background thread above is enough for low volume. For anything higher, put the send in a Celery or RQ task.

### Where do I put the API key?

In the environment, read at import time. Do not put it in a config file you commit, and do not pass it through to a template context.


Docs: https://emails.sh/docs.md