diff --git a/.sampo/changesets/transactional-emails.md b/.sampo/changesets/transactional-emails.md
new file mode 100644
index 000000000..cf264b7e3
--- /dev/null
+++ b/.sampo/changesets/transactional-emails.md
@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Send transactional email from PostHog email templates with `client.emails.send()` on `Posthog` and `AsyncPosthog`, or with `Emails` and `AsyncEmails` from `posthog.emails` in a process that only sends email. A send needs a project secret API key, goes to the app host, sends an idempotency key, and retries network errors, rate limits and an unavailable service with the same key. Set `app_host`, `email_request_timeout_seconds` and `email_request_max_retries` on the client to configure sends. Errors raise `EmailError` with PostHog's error `code` and a Resend-style `name`.
diff --git a/posthog/async_client.py b/posthog/async_client.py
index 186bff727..f8e1dcdcd 100644
--- a/posthog/async_client.py
+++ b/posthog/async_client.py
@@ -10,7 +10,7 @@
 import warnings
 import weakref
 from datetime import datetime, timezone
-from typing import Any, Dict, Mapping, Optional, Union
+from typing import Any, Callable, Dict, Mapping, Optional, Union
 from uuid import UUID, uuid4
 
 from typing_extensions import Unpack
@@ -70,6 +70,7 @@
     try_attach_code_variables_to_frames,
 )
 from .consumer import MAX_MSG_SIZE
+from .emails import AsyncEmails, _client_emails, _validate_options
 from .feature_flag_evaluations import (
     FeatureFlagEvaluations,
     _EvaluatedFlagRecord,
@@ -129,6 +130,9 @@ def __init__(
         personal_api_key: Optional[str] = None,
         feature_flags_request_timeout_seconds: float = 3,
         feature_flags_request_max_retries: int = 1,
+        app_host: Optional[str] = None,
+        email_request_timeout_seconds: float = 15,
+        email_request_max_retries: int = 2,
     ) -> None:
         if flush_at <= 0:
             raise ValueError("flush_at must be greater than zero")
@@ -173,6 +177,10 @@ def __init__(
             else resolved_secret_key
         ) or None
         self.personal_api_key = self.secret_key
+        _validate_options(email_request_timeout_seconds, email_request_max_retries)
+        self.app_host = app_host
+        self.email_request_timeout_seconds = email_request_timeout_seconds
+        self.email_request_max_retries = email_request_max_retries
         self.feature_flags_request_timeout_seconds = (
             feature_flags_request_timeout_seconds
         )
@@ -247,6 +255,48 @@ def __init__(
         self._duplicate_client_registry_key: Optional[tuple[str, str]] = None
         self._register_duplicate_client()
 
+    _emails: Optional[AsyncEmails] = None
+
+    @property
+    def emails(self) -> AsyncEmails:
+        """
+        Send transactional email from PostHog email templates.
+
+        Needs ``secret_key``, a project secret API key with the ``hog_flow:write``
+        scope, and ``posthog[async]``. Sends go to ``app_host``, or to the PostHog
+        Cloud app host of ``host``. ``email_request_timeout_seconds`` and
+        ``email_request_max_retries`` configure each send. A disabled or shut-down
+        client sends nothing. See ``posthog.emails`` for setup.
+
+        Examples:
+            ```python
+            await client.emails.send(
+                to="user@example.com",
+                template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b",
+                distinct_id="user_123",
+            )
+            ```
+        """
+        if self._emails is None:
+            self._emails = _client_emails(
+                AsyncEmails,
+                self.secret_key,
+                self.host,
+                lambda: self.disabled,
+                self._admit_email_send,
+                app_host=self.app_host,
+                timeout=self.email_request_timeout_seconds,
+                max_retries=self.email_request_max_retries,
+            )
+        return self._emails
+
+    def _admit_email_send(self) -> Optional[Callable[[], None]]:
+        # Shutdown waits for in-flight operations before it closes the email transport.
+        if not self._accepting or self._closed:
+            return None
+        completion = self._start_inflight_operation()
+        return lambda: self._finish_inflight_operation(completion)
+
     async def __aenter__(self) -> AsyncClient:
         self._ensure_workers_started()
         return self
@@ -1217,6 +1267,13 @@ async def shutdown(self) -> None:
                 self.log.exception("Failed to close async capture transport")
                 errors.append(error)
 
+            if self._emails is not None:
+                try:
+                    await self._emails.aclose()
+                except Exception as error:
+                    self.log.exception("Failed to close the async email transport")
+                    errors.append(error)
+
             try:
                 self._unregister_duplicate_client()
             except Exception as error:
diff --git a/posthog/client.py b/posthog/client.py
index bff6def5d..6330b8aa7 100644
--- a/posthog/client.py
+++ b/posthog/client.py
@@ -53,6 +53,7 @@
     set_context_session as _context_set_context_session,
     tag as _context_tag,
 )
+from posthog.emails import Emails, _client_emails, _no_op, _validate_options
 from posthog.exception_capture import ExceptionCapture
 from posthog._logging import _configure_posthog_logging
 from posthog.exception_utils import (
@@ -722,6 +723,9 @@ def __init__(
         _use_ai_lane=False,
         _enable_multimodal_capture=False,
         traces: Optional[dict] = None,
+        app_host: Optional[str] = None,
+        email_request_timeout_seconds: float = 15,
+        email_request_max_retries: int = 2,
     ):
         """
         Initialize a new PostHog client instance.
@@ -773,6 +777,15 @@ def __init__(
             feature_flags_request_max_retries: Number of retries for feature flag
                 requests after network, transport, or timeout failures. Defaults
                 to 1. Set to 0 to disable retries.
+            app_host: Your PostHog app host, such as ``https://posthog.example.com``.
+                ``emails`` sends there. Needed only when ``host`` is not PostHog
+                Cloud, for example a reverse proxy, which must not receive the
+                secret key.
+            email_request_timeout_seconds: Timeout in seconds for each email send
+                request. Defaults to 15.
+            email_request_max_retries: Number of retries for an email send after a
+                network error, a rate limit, or an unavailable service. Defaults to 2.
+                Set to 0 when a job runner already retries the send.
             super_properties: Properties merged into every captured event.
             enable_exception_autocapture: Automatically capture uncaught
                 exceptions.
@@ -1037,6 +1050,10 @@ def __init__(
             else resolved_secret_key
         ) or None
         self.personal_api_key = self.secret_key
+        _validate_options(email_request_timeout_seconds, email_request_max_retries)
+        self.app_host = app_host
+        self.email_request_timeout_seconds = email_request_timeout_seconds
+        self.email_request_max_retries = email_request_max_retries
         if debug:
             # Ensures that debug level messages are logged when debug mode is on.
             # Otherwise, defaults to WARNING level. See https://docs.python.org/3/howto/logging.html#what-happens-if-no-configuration-is-provided
@@ -1131,6 +1148,42 @@ def _set_library_identity(self, library_id: str, library_version: str) -> None:
     def _request_identity_kwargs(self) -> Dict[str, str]:
         return {"_user_agent": self._sdk_info} if self._sdk_info != _USER_AGENT else {}
 
+    _emails: Optional[Emails] = None
+
+    @property
+    def emails(self) -> Emails:
+        """
+        Send transactional email from PostHog email templates.
+
+        Needs ``secret_key``, a project secret API key with the ``hog_flow:write``
+        scope. Sends go to ``app_host``, or to the PostHog Cloud app host of
+        ``host``. ``email_request_timeout_seconds`` and ``email_request_max_retries``
+        configure each send. A disabled or shut-down client sends nothing. See
+        ``posthog.emails`` for setup.
+
+        Examples:
+            ```python
+            client.emails.send(
+                to="user@example.com",
+                template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b",
+                template_data={"name": "Ada"},
+                distinct_id="user_123",
+            )
+            ```
+        """
+        if self._emails is None:
+            self._emails = _client_emails(
+                Emails,
+                self.secret_key,
+                self.host,
+                lambda: self.disabled,
+                lambda: None if self._shutdown_requested else _no_op,
+                app_host=self.app_host,
+                timeout=self.email_request_timeout_seconds,
+                max_retries=self.email_request_max_retries,
+            )
+        return self._emails
+
     @property
     def queue(self) -> Queue:
         """The analytics lane's queue (kept for backwards compatibility)."""
@@ -2320,6 +2373,8 @@ def _reinit_after_fork(self):
         self._lifecycle_lock = threading.Lock()
         self._lifecycle_condition = threading.Condition(self._lifecycle_lock)
         self._lifecycle_owner = None
+        # The parent's email session and its lock are not safe to share with the child.
+        self._emails = None
         self._deferred_lifecycle_thread_pending = False
         self._deferred_lifecycle_dirty = False
         self._lifecycle_callback_context = ContextVar(
@@ -2958,6 +3013,12 @@ def _join_once(
                 self._shutdown_flag_definition_cache_provider,
                 errors,
             )
+            if self._emails is not None:
+                self._run_lifecycle_cleanup(
+                    "Failed to close the email transport during lifecycle cleanup",
+                    self._emails.close,
+                    errors,
+                )
             self._run_lifecycle_cleanup(
                 "Failed to unregister client during lifecycle cleanup",
                 self._unregister_duplicate_client,
diff --git a/posthog/emails.py b/posthog/emails.py
new file mode 100644
index 000000000..a2258728c
--- /dev/null
+++ b/posthog/emails.py
@@ -0,0 +1,743 @@
+"""Transactional email, sent from an email template in PostHog.
+
+Setup:
+
+1. In PostHog, open the email template under Workflows, make it transactional,
+   and copy its ID. ``template_id`` is that UUID.
+2. Create a project secret API key (``phs_...``) with Workflow write access
+   (``hog_flow:write``) in your project settings. The project API key
+   (``phc_...``) captures events but cannot send email.
+3. Send from the client you already use, or from :class:`Emails` in a process
+   that only sends email:
+
+```python
+from posthog import Posthog
+
+posthog = Posthog("phc_...", secret_key="phs_...", host="https://eu.i.posthog.com")
+result = posthog.emails.send(
+    to="user@example.com",
+    template_id="<template UUID>",
+    template_data={"name": "Ada"},  # fills {{ variables.name }}
+    distinct_id="user_123",
+    idempotency_key="welcome:user_123",
+)
+print(result.id)  # the queued email
+```
+
+``posthog.emails`` above is the client's property. For async code, use
+``AsyncPosthog`` or :class:`AsyncEmails` with ``posthog[async]`` installed.
+
+Where sends go: a client on PostHog Cloud sends to its region's app host,
+``us.posthog.com`` or ``eu.posthog.com``. For any other ``host``, such as a
+reverse proxy, set ``app_host`` on the client, because the proxy must not
+receive the secret key. :class:`Emails` takes the app host as ``host``.
+
+What a result means: PostHog queued the email. Delivery happens later, and the
+template's workflow in PostHog shows its sends, deliveries and failures.
+
+Timeouts and retries: each request times out after 15 seconds. A send retries
+network errors, rate limits and an unavailable service twice, waiting about 1
+and then 2 seconds, or the server's ``Retry-After`` if that is longer. A
+``Retry-After`` over 30 seconds raises at once. Every attempt sends the same
+idempotency key, so a retry never sends twice. Set
+``email_request_max_retries=0`` on the client, or ``max_retries=0`` on
+:class:`Emails`, when a job runner already retries the send, and pass an
+``idempotency_key`` that names the email, such as ``welcome:user_123``.
+
+Errors: a send that is not queued raises :class:`EmailError`. Branch on
+``code``. The ones a setup usually hits:
+
+- ``missing_secret_key``: the client has no ``secret_key``.
+- ``missing_app_host``: the client's ``host`` is not PostHog Cloud. Set ``app_host``.
+- ``authentication_failed`` (401) or ``permission_denied`` (403): the key is
+  wrong, or it lacks ``hog_flow:write``.
+- ``not_found`` (404): the template ID is wrong, or transactional email is not
+  enabled for the project.
+- ``client_disabled``: the client is disabled, so it sends nothing.
+- ``client_closed``: the client is shut down.
+
+Cleanup: ``shutdown()`` on the client closes its email connections. Close a
+standalone :class:`Emails` with ``close()`` or a ``with`` block.
+
+Testing without network calls: pass your own transport, ``session`` to
+:class:`Emails` or ``http_client`` to :class:`AsyncEmails`, with a fake adapter
+or ``httpx.MockTransport``. Any library that mocks ``requests`` or ``httpx``
+works too. A disabled client sends nothing and raises ``client_disabled``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import importlib
+import json
+import logging
+import math
+import random
+import re
+import threading
+import time
+from dataclasses import dataclass
+from types import TracebackType
+from typing import Any, Callable, Mapping, Optional, TypeVar, Union
+from uuid import UUID, uuid4
+
+import requests
+
+from .capture_v1 import _parse_retry_after
+from .request import (
+    EU_APP_ENDPOINT,
+    US_APP_ENDPOINT,
+    DatetimeSerializer,
+    determine_app_host,
+)
+from .utils import clean
+from .version import VERSION
+
+__all__ = ["AsyncEmails", "EmailError", "EmailWarning", "Emails", "SendEmailResult"]
+
+log = logging.getLogger("posthog")
+
+_SEND_PATH = "/api/projects/@current/messaging_templates/{template_id}/send/"
+_MAX_RETRY_DELAY = 30.0
+_IDEMPOTENCY_KEY = re.compile(r"^[\x21-\x7e]{1,255}$")
+_IN_FLIGHT_CODES = frozenset({"idempotency_key_in_flight"})
+_QUOTA_CODES = frozenset({"quota_limited"})
+
+
+class EmailError(Exception):
+    """PostHog did not queue the email, or the client could not tell whether it did.
+
+    Branch on ``code``, PostHog's stable error code, or on ``name``, which uses
+    Resend's error names. The message never contains the secret key or the
+    request body.
+    """
+
+    def __init__(
+        self,
+        *,
+        name: str,
+        code: str,
+        message: str,
+        status_code: Optional[int] = None,
+        why: Optional[str] = None,
+        fix: Optional[str] = None,
+        docs_url: Optional[str] = None,
+        field: Optional[str] = None,
+        retry_after: Optional[float] = None,
+    ) -> None:
+        super().__init__(message)
+        self.name = name
+        self.code = code
+        self.message = message
+        self.status_code = status_code
+        self.why = why
+        self.fix = fix
+        self.docs_url = docs_url
+        self.field = field
+        self.retry_after = retry_after
+
+
+@dataclass(frozen=True)
+class EmailWarning:
+    code: str
+    message: str
+    fix: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class SendEmailResult:
+    """The queued email. ``id`` is its message ID."""
+
+    id: str
+    replayed: bool = False
+    warnings: tuple[EmailWarning, ...] = ()
+
+
+@dataclass(frozen=True)
+class _PreparedSend:
+    path: str
+    body: bytes
+    idempotency_key: str
+
+
+def _prepare_send(
+    to: str,
+    template_id: Union[str, UUID],
+    template_data: Optional[Mapping[str, Any]],
+    distinct_id: Optional[str],
+    idempotency_key: Optional[str],
+) -> _PreparedSend:
+    if not isinstance(to, str):
+        raise TypeError("to must be one email address as a string.")
+    if not to.strip():
+        raise ValueError("to must not be empty.")
+    try:
+        template_uuid = (
+            template_id if isinstance(template_id, UUID) else UUID(template_id)
+        )
+    except (TypeError, ValueError, AttributeError):
+        raise ValueError("template_id must be the template's UUID.") from None
+    if template_data is not None and not isinstance(template_data, Mapping):
+        raise TypeError("template_data must be a mapping.")
+    if distinct_id is not None and (
+        not isinstance(distinct_id, str) or not distinct_id
+    ):
+        raise ValueError("distinct_id must be a nonempty string.")
+    if idempotency_key is None:
+        idempotency_key = str(uuid4())
+    elif not isinstance(idempotency_key, str) or not _IDEMPOTENCY_KEY.match(
+        idempotency_key
+    ):
+        raise ValueError(
+            "idempotency_key must be 1 to 255 visible ASCII characters without spaces."
+        )
+
+    payload: dict[str, Any] = {
+        "to": to,
+        "message_data": clean(dict(template_data or {})),
+    }
+    if distinct_id is not None:
+        payload["identifiers"] = {"distinct_id": distinct_id}
+    return _PreparedSend(
+        path=_SEND_PATH.format(template_id=template_uuid),
+        body=json.dumps(payload, cls=DatetimeSerializer).encode("utf-8"),
+        idempotency_key=idempotency_key,
+    )
+
+
+def _validate_options(timeout: float, max_retries: int) -> None:
+    if (
+        isinstance(timeout, bool)
+        or not isinstance(timeout, (int, float))
+        or not math.isfinite(timeout)
+        or timeout <= 0
+    ):
+        raise ValueError("timeout must be a finite positive number of seconds.")
+    if (
+        isinstance(max_retries, bool)
+        or not isinstance(max_retries, int)
+        or max_retries < 0
+    ):
+        raise ValueError("max_retries must be a nonnegative integer.")
+
+
+def _text(value: Any) -> Optional[str]:
+    return value if isinstance(value, str) and value else None
+
+
+def _error_name(status: Optional[int], code: str) -> str:
+    if code == "missing_secret_key" or status == 401:
+        return "missing_api_key"
+    if status is None:
+        return "service_unavailable" if code == "network_error" else "application_error"
+    if status == 400:
+        return "validation_error"
+    if status == 403:
+        return "invalid_permission"
+    if status == 404:
+        return "not_found"
+    if status == 409:
+        return (
+            "validation_error"
+            if code == "template_outdated"
+            else "concurrent_idempotent_requests"
+        )
+    if status == 422:
+        # The template send route also answers 422 for send checks such as recipient_suppressed.
+        return (
+            "invalid_idempotent_request"
+            if code == "idempotency_key_reused"
+            else "validation_error"
+        )
+    if status == 429:
+        return (
+            "monthly_quota_exceeded" if code in _QUOTA_CODES else "rate_limit_exceeded"
+        )
+    if status == 503:
+        return "service_unavailable"
+    return "application_error"
+
+
+def _sdk_error(code: str, message: str, status: Optional[int] = None) -> EmailError:
+    return EmailError(
+        name=_error_name(status, code), code=code, message=message, status_code=status
+    )
+
+
+def _retry_after(header: Optional[str]) -> Optional[float]:
+    seconds = _parse_retry_after(header)
+    if seconds is None or not math.isfinite(seconds):
+        return None
+    return max(0.0, seconds)
+
+
+def _parse_response(
+    status: int, payload: Any, retry_after_header: Optional[str]
+) -> SendEmailResult:
+    fields = payload if isinstance(payload, dict) else {}
+    if 200 <= status < 300:
+        message_id = _text(fields.get("message_id"))
+        if message_id is None:
+            raise _sdk_error(
+                "api_error", "PostHog answered without a message ID.", status
+            )
+        warnings = fields.get("warnings")
+        return SendEmailResult(
+            id=message_id,
+            replayed=fields.get("replayed") is True,
+            warnings=tuple(
+                EmailWarning(
+                    code=_text(warning.get("code")) or "warning",
+                    message=_text(warning.get("detail"))
+                    or _text(warning.get("message"))
+                    or "",
+                    fix=_text(warning.get("fix")),
+                )
+                for warning in (warnings if isinstance(warnings, list) else [])
+                if isinstance(warning, dict)
+            ),
+        )
+    code = _text(fields.get("code")) or "api_error"
+    raise EmailError(
+        name=_error_name(status, code),
+        code=code,
+        message=_text(fields.get("detail")) or f"PostHog answered HTTP {status}.",
+        status_code=status,
+        why=_text(fields.get("why")),
+        fix=_text(fields.get("fix")),
+        docs_url=_text(fields.get("docs_url")),
+        field=_text(fields.get("attr")),
+        retry_after=_retry_after(retry_after_header),
+    )
+
+
+def _is_retryable(error: EmailError) -> bool:
+    # Every request carries an idempotency key, so a retry can't queue a second email.
+    if error.status_code is None:
+        return error.code == "network_error"
+    if error.status_code == 409:
+        return error.code in _IN_FLIGHT_CODES
+    if error.status_code == 429:
+        return error.code not in _QUOTA_CODES
+    return error.status_code == 503
+
+
+def _retry_delay(error: EmailError, attempt: int, max_retries: int) -> Optional[float]:
+    if attempt >= max_retries or not _is_retryable(error):
+        return None
+    # Retrying before a longer Retry-After would break the server's rate limit.
+    if error.retry_after is not None and error.retry_after > _MAX_RETRY_DELAY:
+        return None
+    backoff = min(_MAX_RETRY_DELAY, 2.0**attempt * random.uniform(0.8, 1.2))
+    return max(backoff, error.retry_after or 0.0)
+
+
+def _no_op() -> None:
+    pass
+
+
+def _log_warnings(result: SendEmailResult) -> None:
+    for warning in result.warnings:
+        log.warning("PostHog email warning %s: %s", warning.code, warning.message)
+
+
+def _headers(secret_key: str, idempotency_key: str) -> dict[str, str]:
+    return {
+        "Authorization": f"Bearer {secret_key}",
+        "Accept": "application/json",
+        "Content-Type": "application/json",
+        "Idempotency-Key": idempotency_key,
+        "User-Agent": f"posthog-python/{VERSION}",
+    }
+
+
+class _BearerAuth(requests.auth.AuthBase):
+    def __init__(self, secret_key: str, idempotency_key: str) -> None:
+        self._headers = _headers(secret_key, idempotency_key)
+
+    def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
+        # Setting auth explicitly stops requests from substituting .netrc credentials.
+        request.headers.update(self._headers)
+        return request
+
+
+class _EmailsBase:
+    def __init__(
+        self,
+        secret_key: Optional[str],
+        *,
+        host: Optional[str],
+        timeout: float,
+        max_retries: int,
+        disabled: bool,
+    ) -> None:
+        _validate_options(timeout, max_retries)
+        self.secret_key = secret_key.strip() if isinstance(secret_key, str) else None
+        self.host = determine_app_host(host)
+        self.timeout = timeout
+        self.max_retries = max_retries
+        self.disabled = disabled
+        self._parent_disabled: Callable[[], bool] = lambda: False
+        self._admit: Callable[[], Optional[Callable[[], None]]] = lambda: _no_op
+        self._host_error: Optional[str] = None
+
+    def _is_disabled(self) -> bool:
+        return self.disabled or self._parent_disabled()
+
+    def _start_send(self) -> Callable[[], None]:
+        finish = self._admit()
+        if finish is None:
+            raise _sdk_error(
+                "client_closed",
+                "The PostHog client is shut down, so no email was sent.",
+            )
+        return finish
+
+    def _secret_key(self) -> str:
+        if self._host_error:
+            raise _sdk_error("missing_app_host", self._host_error)
+        if not self.secret_key:
+            raise _sdk_error(
+                "missing_secret_key",
+                "Sending email needs secret_key, a project secret API key with the hog_flow:write scope.",
+            )
+        return self.secret_key
+
+
+class Emails(_EmailsBase):
+    """Sends transactional email from PostHog email templates.
+
+    ``host`` is your PostHog app host. PostHog Cloud ingestion hosts map to the
+    app host, so ``us.i.posthog.com`` sends to ``us.posthog.com``. Any other
+    host is used as given.
+
+    Each send retries network errors, rate limits and an unavailable service up
+    to ``max_retries`` times, with the same idempotency key. Set
+    ``max_retries=0`` when a job runner already retries the send. ``timeout``
+    is in seconds, per request. Pass ``session`` to send through your own
+    ``requests.Session``, for example a test double.
+    """
+
+    def __init__(
+        self,
+        secret_key: Optional[str],
+        *,
+        host: Optional[str] = None,
+        timeout: float = 15,
+        max_retries: int = 2,
+        disabled: bool = False,
+        session: Optional[requests.Session] = None,
+    ) -> None:
+        super().__init__(
+            secret_key,
+            host=host,
+            timeout=timeout,
+            max_retries=max_retries,
+            disabled=disabled,
+        )
+        self._session: Optional[requests.Session] = session
+        self._owns_session = session is None
+        self._session_lock = threading.Lock()
+
+    def send(
+        self,
+        *,
+        to: str,
+        template_id: Union[str, UUID],
+        template_data: Optional[Mapping[str, Any]] = None,
+        distinct_id: Optional[str] = None,
+        idempotency_key: Optional[str] = None,
+    ) -> SendEmailResult:
+        """Queue one email from a transactional email template.
+
+        Args:
+            to: One recipient email address.
+            template_id: The UUID of a template made transactional in PostHog.
+            template_data: Values for the template's ``{{ variables.* }}``
+                placeholders. Values are encoded the way capture encodes
+                properties.
+            distinct_id: The recipient's distinct ID in your product. The
+                email's events go to this person. Without it, they go to a
+                person whose distinct ID is the email address.
+            idempotency_key: 1 to 255 visible ASCII characters. A send with a
+                key that PostHog has already accepted queues nothing and returns
+                the first send's result. Without it, the client makes a key
+                that covers its own retries. Pass a key derived from what the
+                email is for, such as ``welcome:user_123``, to cover retries
+                outside this call, too.
+
+        Returns:
+            The queued email. A result means PostHog queued the email, not that
+            it was delivered.
+
+        Raises:
+            EmailError: PostHog did not queue the email, or the client could
+                not tell whether it did.
+            TypeError, ValueError: An argument has the wrong type or format.
+
+        Examples:
+            ```python
+            emails.send(
+                to="user@example.com",
+                template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b",
+                template_data={"name": "Ada"},
+                distinct_id="user_123",
+                idempotency_key="welcome:user_123",
+            )
+            ```
+        """
+        prepared = _prepare_send(
+            to, template_id, template_data, distinct_id, idempotency_key
+        )
+        if self._is_disabled():
+            raise _sdk_error(
+                "client_disabled",
+                "The PostHog client is disabled, so no email was sent.",
+            )
+        finish = self._start_send()
+        try:
+            return self._deliver(prepared)
+        finally:
+            finish()
+
+    def _deliver(self, prepared: _PreparedSend) -> SendEmailResult:
+        auth = _BearerAuth(self._secret_key(), prepared.idempotency_key)
+        session = self._get_session()
+        url = self.host + prepared.path
+        for attempt in range(self.max_retries + 1):
+            try:
+                response = session.request(
+                    "POST",
+                    url,
+                    data=prepared.body,
+                    auth=auth,
+                    timeout=self.timeout,
+                    allow_redirects=False,
+                )
+                try:
+                    try:
+                        payload = response.json()
+                    except ValueError:
+                        payload = None
+                    result = _parse_response(
+                        response.status_code,
+                        payload,
+                        response.headers.get("Retry-After"),
+                    )
+                finally:
+                    response.close()
+                _log_warnings(result)
+                return result
+            except requests.exceptions.RequestException as exc:
+                error = _sdk_error("network_error", "The request to PostHog failed.")
+                error.__cause__ = exc
+            except EmailError as exc:
+                error = exc
+            delay = _retry_delay(error, attempt, self.max_retries)
+            if delay is None:
+                raise error
+            time.sleep(delay)
+        raise AssertionError("unreachable")
+
+    def close(self) -> None:
+        """Close the HTTP connections this client opened. A ``session`` you passed in stays open."""
+        with self._session_lock:
+            if self._session is not None and self._owns_session:
+                self._session.close()
+                self._session = None
+
+    def __enter__(self) -> Emails:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[type[BaseException]],
+        exc: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
+        self.close()
+
+    def _get_session(self) -> requests.Session:
+        with self._session_lock:
+            if self._session is None:
+                session = requests.Session()
+                session.mount("https://", requests.adapters.HTTPAdapter(max_retries=0))
+                session.mount("http://", requests.adapters.HTTPAdapter(max_retries=0))
+                self._session = session
+            return self._session
+
+
+class AsyncEmails(_EmailsBase):
+    """The async counterpart of :class:`Emails`. Needs ``posthog[async]``.
+
+    Pass ``http_client`` to send through your own ``httpx.AsyncClient``, for
+    example one with an ``httpx.MockTransport`` in tests.
+    """
+
+    def __init__(
+        self,
+        secret_key: Optional[str],
+        *,
+        host: Optional[str] = None,
+        timeout: float = 15,
+        max_retries: int = 2,
+        disabled: bool = False,
+        http_client: Any = None,
+    ) -> None:
+        super().__init__(
+            secret_key,
+            host=host,
+            timeout=timeout,
+            max_retries=max_retries,
+            disabled=disabled,
+        )
+        self._client: Any = http_client
+        self._owns_client = http_client is None
+
+    async def send(
+        self,
+        *,
+        to: str,
+        template_id: Union[str, UUID],
+        template_data: Optional[Mapping[str, Any]] = None,
+        distinct_id: Optional[str] = None,
+        idempotency_key: Optional[str] = None,
+    ) -> SendEmailResult:
+        """Queue one email from a transactional email template.
+
+        Takes the same arguments and raises the same errors as
+        :meth:`Emails.send`.
+
+        Examples:
+            ```python
+            await emails.send(
+                to="user@example.com",
+                template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b",
+                template_data={"name": "Ada"},
+                distinct_id="user_123",
+            )
+            ```
+        """
+        prepared = _prepare_send(
+            to, template_id, template_data, distinct_id, idempotency_key
+        )
+        if self._is_disabled():
+            raise _sdk_error(
+                "client_disabled",
+                "The PostHog client is disabled, so no email was sent.",
+            )
+        finish = self._start_send()
+        try:
+            return await self._deliver(prepared)
+        finally:
+            finish()
+
+    async def _deliver(self, prepared: _PreparedSend) -> SendEmailResult:
+        headers = _headers(self._secret_key(), prepared.idempotency_key)
+        httpx, client = self._get_client()
+        url = self.host + prepared.path
+        for attempt in range(self.max_retries + 1):
+            try:
+                # An injected client may carry its own auth and timeout, so set both per request: auth=None keeps
+                # the secret key's Bearer header, and the send timeout applies whatever the client's default.
+                response = await client.post(
+                    url,
+                    content=prepared.body,
+                    headers=headers,
+                    auth=None,
+                    timeout=self.timeout,
+                )
+                try:
+                    try:
+                        payload = response.json()
+                    except ValueError:
+                        payload = None
+                    result = _parse_response(
+                        response.status_code,
+                        payload,
+                        response.headers.get("Retry-After"),
+                    )
+                finally:
+                    await response.aclose()
+                _log_warnings(result)
+                return result
+            except httpx.RequestError as exc:
+                error = _sdk_error("network_error", "The request to PostHog failed.")
+                error.__cause__ = exc
+            except EmailError as exc:
+                error = exc
+            delay = _retry_delay(error, attempt, self.max_retries)
+            if delay is None:
+                raise error
+            await asyncio.sleep(delay)
+        raise AssertionError("unreachable")
+
+    async def aclose(self) -> None:
+        """Close the HTTP connections this client opened. An ``http_client`` you passed in stays open."""
+        if self._client is not None and self._owns_client:
+            client, self._client = self._client, None
+            await client.aclose()
+
+    async def __aenter__(self) -> AsyncEmails:
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[type[BaseException]],
+        exc: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
+        await self.aclose()
+
+    def _get_client(self) -> tuple[Any, Any]:
+        try:
+            httpx = importlib.import_module("httpx")
+        except ImportError:
+            raise RuntimeError(
+                "Async email sending needs httpx. Install it with `posthog[async]`."
+            ) from None
+        if self._client is None:
+            self._client = httpx.AsyncClient(
+                timeout=self.timeout, follow_redirects=False
+            )
+        return httpx, self._client
+
+
+_EmailsT = TypeVar("_EmailsT", bound=_EmailsBase)
+
+
+def _client_emails(
+    factory: Callable[..., _EmailsT],
+    secret_key: Optional[str],
+    client_host: Optional[str],
+    is_disabled: Callable[[], bool],
+    admit: Callable[[], Optional[Callable[[], None]]],
+    *,
+    app_host: Optional[str],
+    timeout: float,
+    max_retries: int,
+) -> _EmailsT:
+    if app_host:
+        emails = factory(
+            secret_key, host=app_host, timeout=timeout, max_retries=max_retries
+        )
+        emails._parent_disabled = is_disabled
+        emails._admit = admit
+        return emails
+    inferred_host = determine_app_host(client_host)
+    is_cloud = inferred_host in (US_APP_ENDPOINT, EU_APP_ENDPOINT)
+    emails = factory(
+        secret_key,
+        host=inferred_host if is_cloud else None,
+        timeout=timeout,
+        max_retries=max_retries,
+    )
+    if not is_cloud:
+        # A custom host is often an ingestion proxy, which must not receive the secret key.
+        emails._host_error = (
+            "The client's host is not PostHog Cloud, so the client can't tell where to send email. "
+            "Set app_host to your PostHog app host on the client."
+        )
+    emails._parent_disabled = is_disabled
+    emails._admit = admit
+    return emails
diff --git a/posthog/request.py b/posthog/request.py
index 9b7b1b1b2..f1816444d 100644
--- a/posthog/request.py
+++ b/posthog/request.py
@@ -186,6 +186,8 @@ def disable_connection_reuse() -> None:
 
 US_INGESTION_ENDPOINT = "https://us.i.posthog.com"
 EU_INGESTION_ENDPOINT = "https://eu.i.posthog.com"
+US_APP_ENDPOINT = "https://us.posthog.com"
+EU_APP_ENDPOINT = "https://eu.posthog.com"
 DEFAULT_HOST = US_INGESTION_ENDPOINT
 USER_AGENT = "posthog-python/" + VERSION
 
@@ -213,6 +215,17 @@ def determine_server_host(host: Optional[str]) -> str:
         return host_or_default
 
 
+def determine_app_host(host: Optional[str]) -> str:
+    """Determines the host for app API calls. PostHog Cloud ingestion hosts don't serve them."""
+    trimmed_host = remove_trailing_slash(normalize_host(host))
+    if trimmed_host in (US_INGESTION_ENDPOINT, "https://app.posthog.com"):
+        return US_APP_ENDPOINT
+    elif trimmed_host == EU_INGESTION_ENDPOINT:
+        return EU_APP_ENDPOINT
+    else:
+        return trimmed_host
+
+
 def post(
     api_key: str,
     host: Optional[str] = None,
diff --git a/posthog/test/test_emails.py b/posthog/test/test_emails.py
new file mode 100644
index 000000000..056bb4fca
--- /dev/null
+++ b/posthog/test/test_emails.py
@@ -0,0 +1,464 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from datetime import datetime, timezone
+from decimal import Decimal
+from typing import Any, Optional, Union
+from unittest import mock
+
+import httpx
+import pytest
+import requests
+from parameterized import parameterized
+
+from posthog import AsyncPosthog, Posthog
+from posthog.emails import AsyncEmails, EmailError, Emails
+from posthog.version import VERSION
+
+SECRET_KEY = "phs_test_secret"
+TEMPLATE_ID = "0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b"
+SEND_URL = f"https://us.posthog.com/api/projects/@current/messaging_templates/{TEMPLATE_ID}/send/"
+
+Reply = Union[Exception, tuple[int, Optional[dict[str, Any]], dict[str, str]]]
+
+
+def accepted(message_id: str = "msg_1", **fields: Any) -> Reply:
+    return (202, {"message_id": message_id, **fields}, {})
+
+
+def rejected(status: int, code: Optional[str] = None, **headers: str) -> Reply:
+    body = (
+        {"type": "invalid_request", "code": code, "detail": "Nope."} if code else None
+    )
+    return (status, body, headers)
+
+
+class FakeAdapter(requests.adapters.BaseAdapter):
+    def __init__(self, replies: list[Reply]) -> None:
+        super().__init__()
+        self.replies = list(replies)
+        self.requests: list[requests.PreparedRequest] = []
+
+    def send(
+        self, request: requests.PreparedRequest, **kwargs: Any
+    ) -> requests.Response:
+        self.requests.append(request)
+        reply = self.replies.pop(0)
+        if isinstance(reply, Exception):
+            raise reply
+        status, body, headers = reply
+        response = requests.Response()
+        response.status_code = status
+        response._content = b"" if body is None else json.dumps(body).encode()
+        response.headers.update(headers)
+        response.request = request
+        response.url = request.url or ""
+        return response
+
+    def close(self) -> None:
+        pass
+
+
+def emails_with(replies: list[Reply], **options: Any) -> tuple[Emails, FakeAdapter]:
+    adapter = FakeAdapter(replies)
+    session = requests.Session()
+    session.mount("https://", adapter)
+    return Emails(SECRET_KEY, session=session, **options), adapter
+
+
+def response(status: int, body: dict[str, Any]) -> requests.Response:
+    reply = requests.Response()
+    reply.status_code = status
+    reply._content = json.dumps(body).encode()
+    return reply
+
+
+def send(emails: Emails, **overrides: Any):
+    return emails.send(
+        **{"to": "user@example.com", "template_id": TEMPLATE_ID, **overrides}
+    )
+
+
+@pytest.fixture(autouse=True)
+def no_sleep():
+    with (
+        mock.patch("posthog.emails.time.sleep") as sleep,
+        mock.patch("posthog.emails.random.uniform", return_value=1.0),
+    ):
+        yield sleep
+
+
+@parameterized.expand(
+    [
+        (
+            "with_distinct_id",
+            {"distinct_id": "user_123"},
+            {"identifiers": {"distinct_id": "user_123"}},
+        ),
+        ("without_distinct_id", {}, {}),
+    ]
+)
+def test_send_posts_the_template_to_the_app_host(_name, overrides, extra_body):
+    emails, adapter = emails_with([accepted("msg_1")])
+
+    result = send(
+        emails,
+        template_data={
+            "total": Decimal("19.5"),
+            "expires_at": datetime(2026, 11, 20, tzinfo=timezone.utc),
+        },
+        idempotency_key="welcome:user_123",
+        **overrides,
+    )
+
+    assert result.id == "msg_1"
+    assert result.replayed is False
+    [request] = adapter.requests
+    assert (request.method, request.url) == ("POST", SEND_URL)
+    assert request.headers["Authorization"] == f"Bearer {SECRET_KEY}"
+    assert request.headers["Idempotency-Key"] == "welcome:user_123"
+    assert request.headers["User-Agent"] == f"posthog-python/{VERSION}"
+    assert json.loads(request.body) == {
+        "to": "user@example.com",
+        "message_data": {"total": 19.5, "expires_at": "2026-11-20T00:00:00+00:00"},
+        **extra_body,
+    }
+
+
+@parameterized.expand(
+    [
+        ("default", None, "https://us.posthog.com"),
+        ("us_ingestion", "https://us.i.posthog.com", "https://us.posthog.com"),
+        ("eu_ingestion", "https://eu.i.posthog.com/", "https://eu.posthog.com"),
+        ("legacy_app", "https://app.posthog.com", "https://us.posthog.com"),
+    ]
+)
+def test_sends_go_to_the_app_host(_name, host, expected):
+    assert Emails(SECRET_KEY, host=host).host == expected
+    # The client turns app hosts into ingestion hosts, so the emails API maps them back.
+    client_host = host or "https://us.posthog.com"
+    client = Posthog(
+        "phc_test", host=client_host, secret_key=SECRET_KEY, sync_mode=True
+    )
+    assert client.emails.host == expected
+
+
+def test_a_client_with_a_custom_host_never_sends_it_the_secret_key():
+    assert (
+        Emails(SECRET_KEY, host="https://ph.example.com/").host
+        == "https://ph.example.com"
+    )
+    client = Posthog(
+        "phc_test", host="https://ph.example.com", secret_key=SECRET_KEY, sync_mode=True
+    )
+
+    with mock.patch.object(requests.Session, "request") as request:
+        with pytest.raises(EmailError) as raised:
+            send(client.emails)
+
+    assert raised.value.code == "missing_app_host"
+    request.assert_not_called()
+
+
+def test_client_email_options_set_the_host_timeout_and_retries():
+    client = Posthog(
+        "phc_test",
+        host="https://ph.example.com",
+        secret_key=SECRET_KEY,
+        sync_mode=True,
+        app_host="https://posthog.example.com/",
+        email_request_timeout_seconds=3,
+        email_request_max_retries=0,
+    )
+    unavailable = response(503, {"code": "delivery_unavailable"})
+
+    with mock.patch.object(
+        requests.Session, "request", return_value=unavailable
+    ) as request:
+        with pytest.raises(EmailError):
+            send(client.emails)
+
+    request.assert_called_once()
+    assert request.call_args.args[1].startswith("https://posthog.example.com/api/")
+    assert request.call_args.kwargs["timeout"] == 3
+
+
+@parameterized.expand(
+    [
+        ("rate_limited", rejected(429, "throttled"), True),
+        ("over_quota", rejected(429, "quota_limited"), False),
+        ("service_unavailable", rejected(503, "delivery_unavailable"), True),
+        ("same_key_in_flight", rejected(409, "idempotency_key_in_flight"), True),
+        ("template_outdated", rejected(409, "template_outdated"), False),
+        ("invalid_recipient", rejected(400, "invalid_recipient"), False),
+        ("server_error", rejected(500), False),
+        ("connection_error", requests.exceptions.ConnectionError("down"), True),
+        ("read_timeout", requests.exceptions.ReadTimeout("slow"), True),
+    ]
+)
+def test_retries_only_what_waiting_can_fix(_name, first_reply, retried):
+    emails, adapter = emails_with([first_reply, accepted()])
+
+    if retried:
+        assert send(emails).id == "msg_1"
+        first, second = adapter.requests
+        assert first.headers["Idempotency-Key"] == second.headers["Idempotency-Key"]
+    else:
+        with pytest.raises(EmailError):
+            send(emails)
+        assert len(adapter.requests) == 1
+
+
+@parameterized.expand(
+    [
+        (400, "invalid_recipient", "validation_error"),
+        (401, "authentication_failed", "missing_api_key"),
+        (403, "permission_denied", "invalid_permission"),
+        (404, "not_found", "not_found"),
+        (409, "idempotency_key_in_flight", "concurrent_idempotent_requests"),
+        (409, "template_outdated", "validation_error"),
+        (422, "idempotency_key_reused", "invalid_idempotent_request"),
+        (422, "recipient_suppressed", "validation_error"),
+        (429, "throttled", "rate_limit_exceeded"),
+        (429, "quota_limited", "monthly_quota_exceeded"),
+        (503, "delivery_unavailable", "service_unavailable"),
+        (500, None, "application_error"),
+    ]
+)
+def test_errors_keep_posthogs_code_and_resends_name(status, code, name):
+    emails, _ = emails_with([rejected(status, code)], max_retries=0)
+
+    with pytest.raises(EmailError) as raised:
+        send(emails)
+
+    assert (raised.value.status_code, raised.value.code, raised.value.name) == (
+        status,
+        code or "api_error",
+        name,
+    )
+    assert SECRET_KEY not in str(raised.value)
+
+
+def test_network_failure_after_retries_is_service_unavailable():
+    failure = requests.exceptions.ConnectionError("down")
+    emails, adapter = emails_with([failure, failure, failure])
+
+    with pytest.raises(EmailError) as raised:
+        send(emails)
+
+    assert len(adapter.requests) == 3
+    assert (raised.value.code, raised.value.name) == (
+        "network_error",
+        "service_unavailable",
+    )
+    assert isinstance(raised.value.__cause__, requests.exceptions.ConnectionError)
+
+
+def test_waits_for_retry_after_and_gives_up_on_a_longer_one(no_sleep):
+    emails, adapter = emails_with(
+        [rejected(429, "throttled", **{"Retry-After": "5"}), accepted()]
+    )
+    send(emails)
+    no_sleep.assert_called_once_with(5.0)
+
+    emails, adapter = emails_with(
+        [rejected(429, "throttled", **{"Retry-After": "120"})]
+    )
+    with pytest.raises(EmailError) as raised:
+        send(emails)
+    assert raised.value.retry_after == 120.0
+    assert len(adapter.requests) == 1
+
+
+def test_max_retries_zero_leaves_retries_to_the_caller():
+    emails, adapter = emails_with(
+        [rejected(503, "delivery_unavailable")], max_retries=0
+    )
+
+    with pytest.raises(EmailError):
+        send(emails)
+
+    assert len(adapter.requests) == 1
+
+
+def test_an_acceptance_without_a_message_id_is_not_retried():
+    emails, adapter = emails_with([(202, {}, {}), accepted()])
+
+    with pytest.raises(EmailError) as raised:
+        send(emails)
+
+    assert raised.value.code == "api_error"
+    assert len(adapter.requests) == 1
+
+
+def test_replays_and_warnings_reach_the_caller(caplog):
+    warning = {
+        "code": "template_superseded",
+        "detail": "Newer content.",
+        "fix": "Upload.",
+    }
+    emails, _ = emails_with([accepted("msg_1", replayed=True, warnings=[warning])])
+
+    with caplog.at_level(logging.WARNING, logger="posthog"):
+        result = send(emails)
+
+    assert result.replayed is True
+    assert [(w.code, w.fix) for w in result.warnings] == [
+        ("template_superseded", "Upload.")
+    ]
+    assert "template_superseded" in caplog.text
+
+
+def test_missing_secret_key_fails_before_sending():
+    emails = Emails(None)
+
+    with mock.patch.object(requests.Session, "request") as request:
+        with pytest.raises(EmailError) as raised:
+            send(emails)
+
+    assert (raised.value.code, raised.value.name) == (
+        "missing_secret_key",
+        "missing_api_key",
+    )
+    request.assert_not_called()
+
+
+def test_a_disabled_client_sends_nothing():
+    client = Posthog("phc_test", secret_key=SECRET_KEY, sync_mode=True)
+    emails = client.emails
+
+    client.disabled = True
+    with mock.patch.object(requests.Session, "request") as request:
+        with pytest.raises(EmailError) as raised:
+            send(emails)
+
+    assert (raised.value.code, raised.value.name) == (
+        "client_disabled",
+        "application_error",
+    )
+    request.assert_not_called()
+
+
+def test_a_shut_down_client_sends_no_email():
+    client = Posthog("phc_test", secret_key=SECRET_KEY, sync_mode=True)
+    emails = client.emails
+
+    client.shutdown()
+
+    with mock.patch.object(requests.Session, "request") as request:
+        for handle in (emails, client.emails):
+            with pytest.raises(EmailError) as raised:
+                send(handle)
+            assert raised.value.code == "client_closed"
+    request.assert_not_called()
+    assert emails._session is None
+
+
+@parameterized.expand(
+    [
+        ("several_recipients", {"to": ["a@example.com", "b@example.com"]}, TypeError),
+        ("empty_recipient", {"to": " "}, ValueError),
+        ("template_name", {"template_id": "welcome"}, ValueError),
+        ("template_data_list", {"template_data": [1, 2]}, TypeError),
+        ("key_with_space", {"idempotency_key": "welcome user"}, ValueError),
+        ("key_too_long", {"idempotency_key": "k" * 256}, ValueError),
+    ]
+)
+def test_rejects_invalid_arguments_before_sending(_name, overrides, error):
+    emails, adapter = emails_with([])
+
+    with pytest.raises(error):
+        send(emails, **overrides)
+
+    assert adapter.requests == []
+
+
+@pytest.mark.asyncio
+async def test_async_send_retries_with_the_same_key():
+    seen: list[httpx.Request] = []
+    replies = [
+        httpx.Response(503, json={"code": "delivery_unavailable"}),
+        httpx.Response(202, json={"message_id": "msg_1"}),
+    ]
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        return replies.pop(0)
+
+    emails = AsyncEmails(
+        SECRET_KEY,
+        timeout=3,
+        http_client=httpx.AsyncClient(
+            transport=httpx.MockTransport(handler), auth=("user", "pass"), timeout=None
+        ),
+    )
+
+    with mock.patch("posthog.emails.asyncio.sleep") as sleep:
+        result = await emails.send(
+            to="user@example.com", template_id=TEMPLATE_ID, distinct_id="user_123"
+        )
+
+    assert result.id == "msg_1"
+    sleep.assert_awaited_once()
+    assert [str(request.url) for request in seen] == [SEND_URL, SEND_URL]
+    assert seen[0].headers["Idempotency-Key"] == seen[1].headers["Idempotency-Key"]
+    assert seen[0].headers["Authorization"] == f"Bearer {SECRET_KEY}"
+    assert seen[0].extensions["timeout"] == dict.fromkeys(
+        ["connect", "read", "write", "pool"], 3
+    )
+    assert json.loads(seen[0].content)["identifiers"] == {"distinct_id": "user_123"}
+
+
+@pytest.mark.asyncio
+async def test_async_disabled_client_sends_nothing():
+    handler = mock.Mock(side_effect=AssertionError("a disabled client sent a request"))
+    emails = AsyncEmails(
+        SECRET_KEY,
+        disabled=True,
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+    )
+
+    with pytest.raises(EmailError) as raised:
+        await emails.send(to="user@example.com", template_id=TEMPLATE_ID)
+
+    assert raised.value.code == "client_disabled"
+    handler.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_async_shutdown_waits_for_a_send_in_flight_and_then_refuses_sends():
+    started, release = asyncio.Event(), asyncio.Event()
+    replies = [
+        httpx.Response(503, json={"code": "delivery_unavailable"}),
+        httpx.Response(202, json={"message_id": "msg_1"}),
+    ]
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        started.set()
+        await release.wait()
+        return replies.pop(0)
+
+    client = AsyncPosthog("phc_test", secret_key=SECRET_KEY)
+    emails = client.emails
+    emails._client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+
+    with mock.patch("posthog.emails._retry_delay", return_value=0.0):
+        sending = asyncio.create_task(
+            emails.send(to="user@example.com", template_id=TEMPLATE_ID)
+        )
+        await started.wait()
+        shutting_down = asyncio.create_task(client.shutdown())
+        for _ in range(5):
+            await asyncio.sleep(0)
+        assert not shutting_down.done()
+        release.set()
+
+        assert (await sending).id == "msg_1"
+        await shutting_down
+
+    assert emails._client is None
+    with pytest.raises(EmailError) as raised:
+        await emails.send(to="user@example.com", template_id=TEMPLATE_ID)
+    assert raised.value.code == "client_closed"
diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt
index feb33578b..406f8c99b 100644
--- a/references/public_api_snapshot.txt
+++ b/references/public_api_snapshot.txt
@@ -245,6 +245,7 @@ alias posthog.client.DEFAULT_CODE_VARIABLES_IGNORE_PATTERNS -> posthog.exception
 alias posthog.client.DEFAULT_CODE_VARIABLES_MASK_PATTERNS -> posthog.exception_utils.DEFAULT_CODE_VARIABLES_MASK_PATTERNS
 alias posthog.client.DEFAULT_CODE_VARIABLES_MASK_URL_CREDENTIALS -> posthog.exception_utils.DEFAULT_CODE_VARIABLES_MASK_URL_CREDENTIALS
 alias posthog.client.EVENTS_ENDPOINT -> posthog.request.EVENTS_ENDPOINT
+alias posthog.client.Emails -> posthog.emails.Emails
 alias posthog.client.ExceptionArg -> posthog.args.ExceptionArg
 alias posthog.client.ExceptionCapture -> posthog.exception_capture.ExceptionCapture
 alias posthog.client.FeatureFlag -> posthog.types.FeatureFlag
@@ -725,6 +726,7 @@ attribute posthog.args.OptionalSetArgs.properties: NotRequired[Optional[Dict[str
 attribute posthog.args.OptionalSetArgs.timestamp: NotRequired[Optional[Union[datetime, str]]]
 attribute posthog.args.OptionalSetArgs.uuid: NotRequired[Optional[Union[str, UUID]]]
 attribute posthog.async_client.AsyncClient.api_key = (project_api_key or '').strip()
+attribute posthog.async_client.AsyncClient.app_host = app_host
 attribute posthog.async_client.AsyncClient.before_send = before_send
 attribute posthog.async_client.AsyncClient.capture_compression = _resolve_capture_compression(capture_compression, gzip_fallback=gzip)
 attribute posthog.async_client.AsyncClient.capture_exception_code_variables = capture_exception_code_variables
@@ -738,6 +740,9 @@ attribute posthog.async_client.AsyncClient.debug = debug
 attribute posthog.async_client.AsyncClient.disable_geoip = disable_geoip
 attribute posthog.async_client.AsyncClient.disabled = disabled or not self.api_key
 attribute posthog.async_client.AsyncClient.distinct_ids_feature_flags_reported = SizeLimitedDict(_MAX_DICT_SIZE, set)
+attribute posthog.async_client.AsyncClient.email_request_max_retries = email_request_max_retries
+attribute posthog.async_client.AsyncClient.email_request_timeout_seconds = email_request_timeout_seconds
+attribute posthog.async_client.AsyncClient.emails: AsyncEmails
 attribute posthog.async_client.AsyncClient.feature_flags_request_max_retries = max(0, feature_flags_request_max_retries)
 attribute posthog.async_client.AsyncClient.feature_flags_request_timeout_seconds = feature_flags_request_timeout_seconds
 attribute posthog.async_client.AsyncClient.gzip = gzip
@@ -775,6 +780,7 @@ attribute posthog.capture_v1.CaptureV1Error.drops = drops or []
 attribute posthog.capture_v1.CaptureV1Error.request_id = request_id
 attribute posthog.capture_v1.CaptureV1Error.retry_exhausted = retry_exhausted or []
 attribute posthog.client.Client.api_key = (project_api_key or '').strip()
+attribute posthog.client.Client.app_host = app_host
 attribute posthog.client.Client.capture_compression = _resolve_capture_compression(capture_compression, gzip_fallback=gzip)
 attribute posthog.client.Client.capture_exception_code_variables = capture_exception_code_variables
 attribute posthog.client.Client.capture_mode = _resolve_capture_mode(capture_mode)
@@ -789,6 +795,9 @@ attribute posthog.client.Client.debug = debug
 attribute posthog.client.Client.disable_geoip = disable_geoip
 attribute posthog.client.Client.disabled = disabled or not self.api_key
 attribute posthog.client.Client.distinct_ids_feature_flags_reported = SizeLimitedDict(MAX_DICT_SIZE, set)
+attribute posthog.client.Client.email_request_max_retries = email_request_max_retries
+attribute posthog.client.Client.email_request_timeout_seconds = email_request_timeout_seconds
+attribute posthog.client.Client.emails: Emails
 attribute posthog.client.Client.enable_exception_autocapture = enable_exception_autocapture
 attribute posthog.client.Client.enable_exception_autocapture_rate_limiting = enable_exception_autocapture_rate_limiting
 attribute posthog.client.Client.enable_full_ai_capture = enable_full_ai_capture is True or _use_ai_lane is True or _enable_multimodal_capture is True
@@ -870,6 +879,21 @@ attribute posthog.debug = False
 attribute posthog.default_client = None
 attribute posthog.disable_geoip = True
 attribute posthog.disabled = False
+attribute posthog.emails.EmailError.code = code
+attribute posthog.emails.EmailError.docs_url = docs_url
+attribute posthog.emails.EmailError.field = field
+attribute posthog.emails.EmailError.fix = fix
+attribute posthog.emails.EmailError.message = message
+attribute posthog.emails.EmailError.name = name
+attribute posthog.emails.EmailError.retry_after = retry_after
+attribute posthog.emails.EmailError.status_code = status_code
+attribute posthog.emails.EmailError.why = why
+attribute posthog.emails.EmailWarning.code: str
+attribute posthog.emails.EmailWarning.fix: Optional[str] = None
+attribute posthog.emails.EmailWarning.message: str
+attribute posthog.emails.SendEmailResult.id: str
+attribute posthog.emails.SendEmailResult.replayed: bool = False
+attribute posthog.emails.SendEmailResult.warnings: tuple[EmailWarning, ...] = ()
 attribute posthog.enable_exception_autocapture = False
 attribute posthog.enable_exception_autocapture_rate_limiting = False
 attribute posthog.enable_full_ai_capture = False
@@ -1095,6 +1119,7 @@ attribute posthog.request.APIError.message = message
 attribute posthog.request.APIError.retry_after = retry_after
 attribute posthog.request.APIError.status = status
 attribute posthog.request.DEFAULT_HOST = US_INGESTION_ENDPOINT
+attribute posthog.request.EU_APP_ENDPOINT = 'https://eu.posthog.com'
 attribute posthog.request.EU_INGESTION_ENDPOINT = 'https://eu.i.posthog.com'
 attribute posthog.request.EVENTS_ENDPOINT = '/batch/'
 attribute posthog.request.GetResponse.data: Any
@@ -1109,6 +1134,7 @@ attribute posthog.request.RequestsConnectionError = requests.exceptions.Connecti
 attribute posthog.request.RequestsTimeout = requests.exceptions.Timeout
 attribute posthog.request.SocketOptions = List[Tuple[int, int, Union[int, bytes]]]
 attribute posthog.request.USER_AGENT = 'posthog-python/' + VERSION
+attribute posthog.request.US_APP_ENDPOINT = 'https://us.posthog.com'
 attribute posthog.request.US_INGESTION_ENDPOINT = 'https://us.i.posthog.com'
 attribute posthog.secret_key = None
 attribute posthog.send = True
@@ -1274,15 +1300,20 @@ class posthog.ai.types.TokenUsage
 class posthog.ai.types.ToolInProgress 
 class posthog.args.OptionalCaptureArgs 
 class posthog.args.OptionalSetArgs 
-class posthog.async_client.AsyncClient(project_api_key: str, host: Optional[str] = None, debug: bool = False, max_queue_size: int = 10000, send: bool = True, on_error=None, flush_at: int = 100, flush_interval: float = 5.0, gzip: bool = False, max_retries: int = 3, timeout: float = 15, thread: int = 1, disabled: bool = False, disable_geoip: bool = True, is_server: bool = True, historical_migration: bool = False, super_properties: Optional[dict[str, Any]] = None, before_send=None, log_captured_exceptions: bool = False, project_root: Optional[str] = None, capture_exception_code_variables: bool = False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: Optional[list[str]] = None, capture_mode: Optional[Union[CaptureMode, str]] = None, capture_compression: Optional[Union[CaptureCompression, str]] = None, capture_trace_context: bool = False, secret_key: Optional[str] = None, personal_api_key: Optional[str] = None, feature_flags_request_timeout_seconds: float = 3, feature_flags_request_max_retries: int = 1)
+class posthog.async_client.AsyncClient(project_api_key: str, host: Optional[str] = None, debug: bool = False, max_queue_size: int = 10000, send: bool = True, on_error=None, flush_at: int = 100, flush_interval: float = 5.0, gzip: bool = False, max_retries: int = 3, timeout: float = 15, thread: int = 1, disabled: bool = False, disable_geoip: bool = True, is_server: bool = True, historical_migration: bool = False, super_properties: Optional[dict[str, Any]] = None, before_send=None, log_captured_exceptions: bool = False, project_root: Optional[str] = None, capture_exception_code_variables: bool = False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: Optional[list[str]] = None, capture_mode: Optional[Union[CaptureMode, str]] = None, capture_compression: Optional[Union[CaptureCompression, str]] = None, capture_trace_context: bool = False, secret_key: Optional[str] = None, personal_api_key: Optional[str] = None, feature_flags_request_timeout_seconds: float = 3, feature_flags_request_max_retries: int = 1, app_host: Optional[str] = None, email_request_timeout_seconds: float = 15, email_request_max_retries: int = 2)
 class posthog.async_client.AsyncPosthog 
 class posthog.bucketed_rate_limiter.BucketedRateLimiter(bucket_size: Number, refill_rate: Number, refill_interval_seconds: Number, on_bucket_rate_limited: Optional[Callable[[Hashable], None]] = None, clock: Callable[[], float] = time.monotonic)
 class posthog.capture_compression.CaptureCompression 
 class posthog.capture_mode.CaptureMode 
 class posthog.capture_v1.CaptureV1Error(status: int | str, message: str, *, retry_after: Optional[float] = None, request_id: Optional[str] = None, attempts: Optional[int] = None, retry_exhausted: Optional[list[str]] = None, drops: Optional[list[tuple[str, Optional[str]]]] = None)
-class posthog.client.Client(project_api_key: str, host=None, debug=False, max_queue_size=10000, send=True, on_error=None, flush_at=100, flush_interval=5.0, gzip=False, max_retries=3, sync_mode=False, timeout: float = 15, thread=1, poll_interval=30, personal_api_key=None, disabled=False, disable_geoip=True, is_server=True, historical_migration=False, feature_flags_request_timeout_seconds: float = 3, feature_flags_request_max_retries=1, super_properties=None, enable_exception_autocapture=False, log_captured_exceptions=False, project_root=None, privacy_mode=False, before_send=None, flag_fallback_cache_url=None, enable_local_evaluation=True, flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None, capture_exception_code_variables=False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: list[str] | None = None, enable_exception_autocapture_rate_limiting=False, exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, capture_mode: Optional[Union[CaptureMode, str]] = None, capture_compression: Optional[Union[CaptureCompression, str]] = None, secret_key=None, metrics: Optional[dict] = None, enable_full_ai_capture=False, capture_trace_context=False, _use_ai_lane=False, _enable_multimodal_capture=False, traces: Optional[dict] = None)
+class posthog.client.Client(project_api_key: str, host=None, debug=False, max_queue_size=10000, send=True, on_error=None, flush_at=100, flush_interval=5.0, gzip=False, max_retries=3, sync_mode=False, timeout: float = 15, thread=1, poll_interval=30, personal_api_key=None, disabled=False, disable_geoip=True, is_server=True, historical_migration=False, feature_flags_request_timeout_seconds: float = 3, feature_flags_request_max_retries=1, super_properties=None, enable_exception_autocapture=False, log_captured_exceptions=False, project_root=None, privacy_mode=False, before_send=None, flag_fallback_cache_url=None, enable_local_evaluation=True, flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None, capture_exception_code_variables=False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: list[str] | None = None, enable_exception_autocapture_rate_limiting=False, exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, capture_mode: Optional[Union[CaptureMode, str]] = None, capture_compression: Optional[Union[CaptureCompression, str]] = None, secret_key=None, metrics: Optional[dict] = None, enable_full_ai_capture=False, capture_trace_context=False, _use_ai_lane=False, _enable_multimodal_capture=False, traces: Optional[dict] = None, app_host: Optional[str] = None, email_request_timeout_seconds: float = 15, email_request_max_retries: int = 2)
 class posthog.consumer.Consumer(queue, api_key, flush_at=100, host=None, on_error=None, flush_interval=5.0, gzip=False, retries=10, timeout: float = 15, historical_migration=False, endpoint=EVENTS_ENDPOINT, max_msg_size=MAX_MSG_SIZE, capture_mode=CaptureMode.V0, capture_compression=CaptureCompression.NONE)
 class posthog.contexts.ContextScope(parent=None, fresh: bool = False, capture_exceptions: bool = True, client: Optional[Client] = None)
+class posthog.emails.AsyncEmails(secret_key: Optional[str], *, host: Optional[str] = None, timeout: float = 15, max_retries: int = 2, disabled: bool = False, http_client: Any = None)
+class posthog.emails.EmailError(*, name: str, code: str, message: str, status_code: Optional[int] = None, why: Optional[str] = None, fix: Optional[str] = None, docs_url: Optional[str] = None, field: Optional[str] = None, retry_after: Optional[float] = None)
+class posthog.emails.EmailWarning(code: str, message: str, fix: Optional[str] = None)
+class posthog.emails.Emails(secret_key: Optional[str], *, host: Optional[str] = None, timeout: float = 15, max_retries: int = 2, disabled: bool = False, session: Optional[requests.Session] = None)
+class posthog.emails.SendEmailResult(id: str, replayed: bool = False, warnings: tuple[EmailWarning, ...] = ())
 class posthog.exception_capture.ExceptionCapture(client: Client, rate_limiting_enabled=False, bucket_size=DEFAULT_BUCKET_SIZE, refill_rate=DEFAULT_REFILL_RATE, refill_interval_seconds=DEFAULT_REFILL_INTERVAL_SECONDS)
 class posthog.exception_utils.AnnotatedValue(value, metadata)
 class posthog.exception_utils.VariableSizeLimiter(max_size=DEFAULT_TOTAL_VARIABLES_SIZE_LIMIT)
@@ -1521,6 +1552,7 @@ function posthog.mcp.session_token.read_mcp_session_header(headers: Any) -> Opti
 function posthog.mcp.tools.get_more_tools_result() -> Dict[str, Any]
 function posthog.new_context(fresh: bool = False, capture_exceptions: Optional[bool] = None, client: Optional[Client] = None)
 function posthog.request.batch_post(api_key: str, host: Optional[str] = None, gzip: bool = False, timeout: float = 15, path: str = EVENTS_ENDPOINT, **kwargs) -> requests.Response
+function posthog.request.determine_app_host(host: Optional[str]) -> str
 function posthog.request.determine_server_host(host: Optional[str]) -> str
 function posthog.request.disable_connection_reuse() -> None
 function posthog.request.enable_keep_alive() -> None
@@ -1777,6 +1809,10 @@ method posthog.contexts.ContextScope.set_code_variables_mask_url_credentials(ena
 method posthog.contexts.ContextScope.set_device_id(device_id: str)
 method posthog.contexts.ContextScope.set_distinct_id(distinct_id: str)
 method posthog.contexts.ContextScope.set_session_id(session_id: str)
+method posthog.emails.AsyncEmails.aclose() -> None
+method posthog.emails.AsyncEmails.send(*, to: str, template_id: Union[str, UUID], template_data: Optional[Mapping[str, Any]] = None, distinct_id: Optional[str] = None, idempotency_key: Optional[str] = None) -> SendEmailResult
+method posthog.emails.Emails.close() -> None
+method posthog.emails.Emails.send(*, to: str, template_id: Union[str, UUID], template_data: Optional[Mapping[str, Any]] = None, distinct_id: Optional[str] = None, idempotency_key: Optional[str] = None) -> SendEmailResult
 method posthog.exception_capture.ExceptionCapture.capture_exception(exception, metadata=None)
 method posthog.exception_capture.ExceptionCapture.close()
 method posthog.exception_capture.ExceptionCapture.exception_handler(exc_type, exc_value, exc_traceback)
@@ -1911,6 +1947,7 @@ module posthog.capture_v1
 module posthog.client
 module posthog.consumer
 module posthog.contexts
+module posthog.emails
 module posthog.exception_capture
 module posthog.exception_utils
 module posthog.feature_flag_evaluations
diff --git a/typings/requests/__init__.pyi b/typings/requests/__init__.pyi
index 9e92acd35..1c545b680 100644
--- a/typings/requests/__init__.pyi
+++ b/typings/requests/__init__.pyi
@@ -1,6 +1,7 @@
 from typing import Any
 
 from . import adapters as adapters, auth as auth, exceptions as exceptions
+from .auth import AuthBase as _AuthBase
 
 class PreparedRequest:
     headers: dict[str, str]
@@ -27,6 +28,7 @@ class Session:
         params: dict[str, str | int] | None = ...,
         timeout: float | None = ...,
         allow_redirects: bool = ...,
+        auth: _AuthBase | None = ...,
     ) -> Response: ...
     def post(
         self,