diff --git a/.github/scripts/check_strict_types.sh b/.github/scripts/check_strict_types.sh
index 4855a9ec9..116478d5c 100755
--- a/.github/scripts/check_strict_types.sh
+++ b/.github/scripts/check_strict_types.sh
@@ -15,7 +15,9 @@ import atexit
 
 import posthog
 from posthog import FeatureFlagEvaluations, FlagValue, Posthog
-from posthog.emails import AsyncEmails, Emails, SendEmailResult
+from datetime import datetime, timezone
+
+from posthog.emails import AsyncEmails, CancelEmailResult, EmailList, Emails, SendEmailResult, UpdateEmailResult
 from posthog.ai.evaluations import (
     AsyncOfflineEvaluations,
     BooleanScorerConfig,
@@ -46,15 +48,27 @@ _ = (flag_value, all_flags, enabled, payload, evaluations, active)
 emails = Emails("phs_example")
 template_send: SendEmailResult = emails.send(to="user@example.com", template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", template_data={"name": "Ada"})
 rendered_send: SendEmailResult = emails.send(to="user@example.com", template_key="password-reset", subject="Reset", html="<p>Reset</p>", from_="Acme <security@example.com>", reply_to=["help@example.com"], capture_engagement=False, max_retries=0)
+scheduled_send: SendEmailResult = emails.send(to="user@example.com", template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", scheduled_at=datetime(2026, 12, 1, tzinfo=timezone.utc))
+scheduled_status: str = scheduled_send.status
+scheduled_time: datetime | None = scheduled_send.scheduled_at
+moved: UpdateEmailResult = emails.update(scheduled_send.id, scheduled_at="2026-12-02T09:00:00+02:00")
+canceled: CancelEmailResult = emails.cancel(scheduled_send.id)
+page: EmailList = emails.list(cursor=None, limit=50)
+next_page: EmailList | None = emails.list(cursor=page.next) if page.next else None
+first_time: datetime = page.results[0].scheduled_at
 tracked_send: SendEmailResult = emails.send(to="user@example.com", template_key="welcome", subject="Welcome", html="<p>Hi</p>", tracking={"clicks": True, "engagement": False})
 untracked_send: SendEmailResult = emails.send(to="user@example.com", template_key="welcome", subject="Welcome", html="<p>Hi</p>", tracking=False)
 not_sensitive_send: SendEmailResult = emails.send(to="user@example.com", template_key="newsletter", subject="News", html="<p>News</p>", sensitive=False)
-_ = (template_send, rendered_send, tracked_send, untracked_send)
+_ = (template_send, rendered_send, tracked_send, untracked_send, scheduled_status, scheduled_time, moved, canceled, next_page, first_time)
 
 
 async def async_email_types() -> None:
-    sent: SendEmailResult = await AsyncEmails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>")
-    _ = sent
+    async_emails = AsyncEmails("phs_example")
+    sent: SendEmailResult = await async_emails.send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>", scheduled_at="2026-12-01T09:00:00Z")
+    moved: UpdateEmailResult = await async_emails.update(sent.id, scheduled_at=datetime(2026, 12, 2, tzinfo=timezone.utc))
+    canceled: CancelEmailResult = await async_emails.cancel(sent.id)
+    page: EmailList = await async_emails.list()
+    _ = (sent, moved, canceled, page)
 
 
 async def offline_evaluation_types(
@@ -114,6 +128,11 @@ Emails("phs_example").send(to="user@example.com", template_id="0190f6d2-7c1e-7b3
 Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>", tracking={"open": False})  # expected-type-error
 Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>", template_data={"name": "Ada"})  # expected-type-error
 Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>", cc="team@example.com")  # expected-type-error
+Emails("phs_example").send(to="user@example.com", template_key="welcome", subject="Hi", html="<p>Hi</p>", scheduled_at=1796115600)  # expected-type-error
+Emails("phs_example").update("0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b")  # expected-type-error
+Emails("phs_example").update("0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b", "2026-12-01T09:00:00Z")  # expected-type-error
+Emails("phs_example").list("scheduled")  # expected-type-error
+Emails("phs_example").list(status="sent")  # expected-type-error
 
 
 async def invalid_async_email() -> None:
diff --git a/.sampo/changesets/scheduled-email-sends.md b/.sampo/changesets/scheduled-email-sends.md
new file mode 100644
index 000000000..c0170b3c2
--- /dev/null
+++ b/.sampo/changesets/scheduled-email-sends.md
@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Schedule a transactional email with `scheduled_at` on `emails.send()`, up to 30 days ahead, and move, cancel or list scheduled emails with `emails.update()`, `emails.cancel()` and `emails.list()`. A naive `datetime` raises `ValueError` before any request. Send results now carry `status` and `scheduled_at`.
diff --git a/posthog/emails.py b/posthog/emails.py
index 2839a2dd6..e6073aa6f 100644
--- a/posthog/emails.py
+++ b/posthog/emails.py
@@ -71,12 +71,29 @@
 - ``client_disabled``: the client is disabled, so it sends nothing.
 - ``client_closed``: the client is shut down.
 
-Status: ``emails.get(message_id)`` reads what happened to a sent email, for
+Status: ``emails.get(message_id)`` reads what happened to an email, for
 30 days after the send, as an :class:`EmailStatus`. A key with only
 ``transactional_email:read`` can read statuses but not send. ``get`` raises
 ``not_found`` (404) for an unknown ID, an email older than 30 days, or for a
 few seconds after the send.
 
+Scheduling: pass ``scheduled_at``, an aware ``datetime`` or an ISO 8601
+string with an offset, to send up to 30 days later. The result's ``status`` is
+``scheduled``. Until the email goes out, ``emails.list()`` finds it,
+``emails.update(id, scheduled_at=...)`` moves it, and ``emails.cancel(id)``
+stops it. A naive ``datetime`` raises ``ValueError`` before any request.
+
+```python
+from datetime import datetime, timedelta, timezone
+
+result = posthog.emails.send(
+    to="user@example.com",
+    template_id="<template UUID>",
+    scheduled_at=datetime.now(timezone.utc) + timedelta(days=1),
+)
+posthog.emails.cancel(result.id)
+```
+
 Cleanup: ``shutdown()`` on the client closes its email connections. Close a
 standalone :class:`Emails` with ``close()`` or a ``with`` block.
 
@@ -98,11 +115,12 @@
 import threading
 import time
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import datetime, timezone
 from types import TracebackType
 from typing import (
     Any,
     Callable,
+    Literal,
     Mapping,
     Optional,
     Sequence,
@@ -111,6 +129,7 @@
     Union,
     overload,
 )
+from urllib.parse import urlencode
 from uuid import UUID, uuid4
 
 import requests
@@ -128,26 +147,36 @@
 
 __all__ = [
     "AsyncEmails",
+    "CancelEmailResult",
     "CodeEmailArgs",
     "EmailError",
+    "EmailList",
     "EmailStatus",
     "EmailTracking",
     "EmailWarning",
     "Emails",
+    "ScheduledEmail",
     "SendEmailResult",
     "TemplateEmailArgs",
+    "UpdateEmailResult",
 ]
 
 log = logging.getLogger("posthog")
 
 _SEND_PATH = "/api/projects/@current/messaging/send/email/"
 _STATUS_PATH = "/api/projects/@current/messaging/emails/{message_id}/"
+_LIST_PATH = "/api/projects/@current/messaging/emails/"
+_MAX_LIST_LIMIT = 100
 _TEMPLATE_KEY = re.compile(r"[a-zA-Z0-9_-]{1,400}")
 _MAX_RETRY_DELAY = 30.0
 _IDEMPOTENCY_KEY = re.compile(r"^[\x21-\x7e]{1,255}$")
 _HEADER_TOKEN = re.compile(r"[\x21-\x7e]+")
 _IN_FLIGHT_CODES = frozenset({"idempotency_key_in_flight"})
 _QUOTA_CODES = frozenset({"quota_limited"})
+# Conflicts a retry can't resolve: the email changed under the request, as posthog-node names them.
+_SETTLED_CONFLICT_CODES = frozenset(
+    {"template_outdated", "already_sending", "canceled"}
+)
 
 
 class EmailError(Exception):
@@ -201,20 +230,94 @@ class EmailWarning:
     fix: Optional[str] = None
 
 
+_SCHEDULING_UNSUPPORTED = EmailWarning(
+    code="scheduling_unsupported",
+    message="This PostHog predates scheduled sends, so it ignored scheduled_at and sent the email now.",
+    fix="Update PostHog before you schedule emails with it.",
+)
+
+
 @dataclass(frozen=True)
 class SendEmailResult:
-    """The queued email. ``id`` is its message ID.
+    """The accepted email. ``id`` is its message ID.
 
     Attributes:
         id: The email's message ID.
         replayed: ``True`` when the response says PostHog replayed an earlier send for
             the same ``idempotency_key`` instead of sending again.
-        warnings: Warnings PostHog returned about the request.
+        warnings: Warnings PostHog returned about the request. A send with
+            ``scheduled_at`` that a PostHog without scheduled sends delivered
+            at once adds ``scheduling_unsupported``.
+        status: ``"scheduled"`` when the email waits for its ``scheduled_at``,
+            or ``"queued"`` when it goes out now.
+        scheduled_at: When a scheduled email goes out, in UTC. ``None`` for an
+            email that goes out now, also when ``scheduled_at`` was at or
+            before now.
     """
 
     id: str
     replayed: bool = False
     warnings: tuple[EmailWarning, ...] = ()
+    status: str = "queued"
+    scheduled_at: Optional[datetime] = None
+
+
+@dataclass(frozen=True)
+class UpdateEmailResult:
+    """A scheduled email after :meth:`Emails.update` moved it.
+
+    Attributes:
+        id: The email's message ID.
+        status: ``"scheduled"``, until the email goes out.
+        scheduled_at: When the email goes out now, in UTC.
+    """
+
+    id: str
+    status: str
+    scheduled_at: datetime
+
+
+@dataclass(frozen=True)
+class CancelEmailResult:
+    """A scheduled email after :meth:`Emails.cancel` stopped it.
+
+    Attributes:
+        id: The email's message ID.
+        status: ``"canceled"``.
+    """
+
+    id: str
+    status: str
+
+
+@dataclass(frozen=True)
+class ScheduledEmail:
+    """One email that waits for its send time, as :meth:`Emails.list` returns it.
+
+    Attributes:
+        id: The email's message ID.
+        to: The recipient's email address.
+        scheduled_at: When the email goes out, in UTC.
+        created_at: When PostHog accepted the send.
+    """
+
+    id: str
+    to: str
+    scheduled_at: datetime
+    created_at: datetime
+
+
+@dataclass(frozen=True)
+class EmailList:
+    """One page of :meth:`Emails.list`.
+
+    Attributes:
+        results: The emails on this page, the soonest first.
+        next: Pass it as ``cursor`` to read the next page. ``None`` on the last page.
+    """
+
+    results: tuple[ScheduledEmail, ...]
+    next: Optional[str] = None
 
 
 class _RecipientArgs(TypedDict):
@@ -222,6 +325,7 @@ class _RecipientArgs(TypedDict):
     distinct_id: NotRequired[Optional[str]]
     idempotency_key: NotRequired[Optional[str]]
     max_retries: NotRequired[Optional[int]]
+    scheduled_at: NotRequired[Optional[Union[datetime, str]]]
 
 
 class TemplateEmailArgs(_RecipientArgs):
@@ -236,6 +340,8 @@ class TemplateEmailArgs(_RecipientArgs):
         idempotency_key: 1 to 255 visible ASCII characters.
         max_retries: Overrides the client's maximum number of retries for this
             send. ``0`` turns retries off.
+        scheduled_at: When to send the email, up to 30 days ahead. See
+            :meth:`Emails.send`.
     """
 
     template_id: Union[str, UUID]
@@ -316,6 +422,8 @@ class CodeEmailArgs(_RecipientArgs):
             first send's result, also when its workflow was paused since.
         max_retries: Overrides the client's maximum number of retries for this
             send. ``0`` turns retries off.
+        scheduled_at: When to send the email, up to 30 days ahead. See
+            :meth:`Emails.send`.
     """
 
     template_key: str
@@ -336,6 +444,7 @@ class CodeEmailArgs(_RecipientArgs):
         "from_",
         "reply_to",
         "capture_engagement",
+        "scheduled_at",
         "sensitive",
         "tracking",
     }
@@ -348,6 +457,8 @@ class EmailStatus:
 
     ``status`` is one of:
 
+    - ``scheduled``: the email waits for its ``scheduled_at``.
+    - ``canceled``: :meth:`Emails.cancel` stopped the email before it went out.
     - ``queued``: the email waits to go out.
     - ``sent``: PostHog handed the email to the email provider.
     - ``failed``: the email was not sent, and ``reason`` says why.
@@ -381,6 +492,7 @@ class _PreparedSend:
     body: bytes
     idempotency_key: str
     max_retries: Optional[int] = None
+    scheduled: bool = False
 
 
 def _prepare(args: Mapping[str, Any]) -> _PreparedSend:
@@ -404,6 +516,7 @@ def _prepare(args: Mapping[str, Any]) -> _PreparedSend:
             args.get("distinct_id"),
             args.get("idempotency_key"),
             args.get("max_retries"),
+            args.get("scheduled_at"),
         )
     return _prepare_code_send(args)
 
@@ -438,11 +551,14 @@ def _prepare_code_send(args: Mapping[str, Any]) -> _PreparedSend:
     if distinct_id is not None:
         payload["distinct_id"] = distinct_id
     payload.update(_sender(args))
+    if "scheduled_at" in args:
+        payload["scheduled_at"] = _scheduled_at(args["scheduled_at"])
     return _PreparedSend(
         path=_SEND_PATH,
         body=_encode(payload),
         idempotency_key=idempotency_key,
         max_retries=max_retries,
+        scheduled="scheduled_at" in payload,
     )
 
 
@@ -511,6 +627,7 @@ def _prepare_send(
     distinct_id: Optional[str],
     idempotency_key: Optional[str],
     max_retries: Optional[int] = None,
+    scheduled_at: Optional[Union[datetime, str]] = None,
 ) -> _PreparedSend:
     to, distinct_id, idempotency_key = _recipient(to, distinct_id, idempotency_key)
     max_retries = _retry_limit(max_retries)
@@ -530,11 +647,14 @@ def _prepare_send(
     }
     if distinct_id is not None:
         payload["distinct_id"] = distinct_id
+    if scheduled_at is not None:
+        payload["scheduled_at"] = _scheduled_at(scheduled_at)
     return _PreparedSend(
         path=_SEND_PATH,
         body=json.dumps(payload, cls=DatetimeSerializer).encode("utf-8"),
         idempotency_key=idempotency_key,
         max_retries=max_retries,
+        scheduled="scheduled_at" in payload,
     )
 
 
@@ -580,6 +700,57 @@ def _status_path(message_id: Union[str, UUID]) -> str:
     return _STATUS_PATH.format(message_id=message_uuid)
 
 
+def _cancel_path(message_id: Union[str, UUID]) -> str:
+    return _status_path(message_id) + "cancel/"
+
+
+def _scheduled_at(value: Any) -> str:
+    """The send time on the wire: an aware ``datetime`` as UTC ISO 8601, a string with an offset as given."""
+    if isinstance(value, datetime):
+        if value.utcoffset() is None:
+            raise ValueError(
+                "scheduled_at has no timezone. Pass an aware datetime, such as "
+                "datetime.now(timezone.utc) + timedelta(hours=1)."
+            )
+        return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
+    if isinstance(value, str):
+        parsed = _timestamp(value)
+        if parsed is None or parsed.utcoffset() is None:
+            raise ValueError(
+                "scheduled_at must be an ISO 8601 time with a timezone offset, "
+                "such as 2026-12-01T09:00:00Z or 2026-12-01T09:00:00+02:00."
+            )
+        return value
+    raise TypeError("scheduled_at must be an aware datetime or an ISO 8601 string.")
+
+
+def _prepare_update(scheduled_at: Any) -> bytes:
+    if scheduled_at is None:
+        raise TypeError("update() needs scheduled_at, the email's new send time.")
+    return _encode({"scheduled_at": _scheduled_at(scheduled_at)})
+
+
+def _list_path(status: Any, cursor: Any, limit: Any) -> str:
+    if status != "scheduled":
+        raise ValueError(
+            "status must be 'scheduled': only scheduled emails can be listed."
+        )
+    query: list[tuple[str, str]] = [("status", status)]
+    if cursor is not None:
+        if not isinstance(cursor, str):
+            raise TypeError("cursor must be the next value of the previous page.")
+        if not cursor:
+            raise ValueError("cursor must not be empty.")
+        query.append(("cursor", cursor))
+    if limit is not None:
+        if isinstance(limit, bool) or not isinstance(limit, int):
+            raise TypeError("limit must be an integer.")
+        if not 1 <= limit <= _MAX_LIST_LIMIT:
+            raise ValueError(f"limit must be from 1 to {_MAX_LIST_LIMIT}.")
+        query.append(("limit", str(limit)))
+    return f"{_LIST_PATH}?{urlencode(query)}"
+
+
 def _validate_options(
     timeout: float, max_retries: int, migrated_from: Optional[str]
 ) -> None:
@@ -622,7 +793,7 @@ def _error_name(status: Optional[int], code: str) -> str:
     if status == 409:
         return (
             "validation_error"
-            if code == "template_outdated"
+            if code in _SETTLED_CONFLICT_CODES
             else "concurrent_idempotent_requests"
         )
     if status == 422:
@@ -655,7 +826,10 @@ def _retry_after(header: Optional[str]) -> Optional[float]:
 
 
 def _parse_response(
-    status: int, payload: Any, retry_after_header: Optional[str]
+    status: int,
+    payload: Any,
+    retry_after_header: Optional[str],
+    scheduled: bool = False,
 ) -> SendEmailResult:
     fields = payload if isinstance(payload, dict) else {}
     if 200 <= status < 300:
@@ -665,20 +839,27 @@ def _parse_response(
                 "api_error", "PostHog answered without a message ID.", status
             )
         warnings = fields.get("warnings")
+        server_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)
+        )
+        # A PostHog without scheduled sends drops scheduled_at, sends at once and answers no status.
+        unscheduled = (
+            (_SCHEDULING_UNSUPPORTED,) if scheduled and "status" not in fields else ()
+        )
         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)
-            ),
+            status=_text(fields.get("status")) or "queued",
+            scheduled_at=_timestamp(fields.get("scheduled_at")),
+            warnings=server_warnings + unscheduled,
         )
     raise _response_error(fields, status, retry_after_header)
 
@@ -724,6 +905,85 @@ def _parse_status_response(
     )
 
 
+def _parse_update_response(
+    status: int, payload: Any, retry_after_header: Optional[str]
+) -> UpdateEmailResult:
+    fields = _success_fields(status, payload, retry_after_header)
+    message_id = _text(fields.get("id"))
+    email_status = _text(fields.get("status"))
+    scheduled_at = _timestamp(fields.get("scheduled_at"))
+    if message_id is None or email_status is None or scheduled_at is None:
+        raise _sdk_error(
+            "api_error", "PostHog answered without the email's send time.", status
+        )
+    return UpdateEmailResult(
+        id=message_id, status=email_status, scheduled_at=scheduled_at
+    )
+
+
+def _parse_cancel_response(
+    status: int, payload: Any, retry_after_header: Optional[str]
+) -> CancelEmailResult:
+    fields = _success_fields(status, payload, retry_after_header)
+    message_id = _text(fields.get("id"))
+    if message_id is None or fields.get("status") != "canceled":
+        raise _sdk_error(
+            "api_error", "PostHog answered without a canceled email.", status
+        )
+    return CancelEmailResult(id=message_id, status="canceled")
+
+
+def _parse_list_response(
+    status: int, payload: Any, retry_after_header: Optional[str]
+) -> EmailList:
+    fields = _success_fields(status, payload, retry_after_header)
+    rows = fields.get("results")
+    if not isinstance(rows, list):
+        raise _sdk_error(
+            "api_error", "PostHog answered without a list of emails.", status
+        )
+    results = []
+    for row in rows:
+        row = row if isinstance(row, dict) else {}
+        message_id = _text(row.get("id"))
+        to = _text(row.get("to"))
+        scheduled_at = _timestamp(row.get("scheduled_at"))
+        created_at = _timestamp(row.get("created_at"))
+        if (
+            message_id is None
+            or to is None
+            or scheduled_at is None
+            or created_at is None
+        ):
+            # Dropping a row would hide an email that still goes out.
+            raise _sdk_error(
+                "api_error",
+                "PostHog answered with an email the client can't read.",
+                status,
+            )
+        results.append(
+            ScheduledEmail(
+                id=message_id, to=to, scheduled_at=scheduled_at, created_at=created_at
+            )
+        )
+    next_cursor = fields.get("next", "")
+    if next_cursor is not None and _text(next_cursor) is None:
+        # Losing the cursor would end a caller's paging loop early.
+        raise _sdk_error(
+            "api_error", "PostHog answered without the next page's cursor.", status
+        )
+    return EmailList(results=tuple(results), next=next_cursor)
+
+
+def _success_fields(
+    status: int, payload: Any, retry_after_header: Optional[str]
+) -> Mapping[str, Any]:
+    fields = payload if isinstance(payload, dict) else {}
+    if not 200 <= status < 300:
+        raise _response_error(fields, status, retry_after_header)
+    return fields
+
+
 def _response_error(
     fields: Mapping[str, Any], status: int, retry_after_header: Optional[str]
 ) -> EmailError:
@@ -743,6 +1003,7 @@ def _response_error(
 
 def _is_retryable(error: EmailError) -> bool:
     # A send carries an idempotency key, so a retry can't queue a second email. A status read changes nothing.
+    # A cancel, or a move to a later time, ends the same however often it lands, so it needs no key.
     if error.status_code is None:
         return error.code == "network_error"
     if error.status_code == 409:
@@ -794,11 +1055,22 @@ def _headers(
 
 _SEND_NOT_DONE = "no email was sent"
 _GET_NOT_DONE = "no email status was read"
+_UPDATE_NOT_DONE = "no email was moved"
+_CANCEL_NOT_DONE = "no email was canceled"
+_LIST_NOT_DONE = "no emails were listed"
 _SEND_KEY_NEEDED = "Sending email needs secret_key, a project secret API key with the transactional_email:write scope."
 _GET_KEY_NEEDED = (
     "Reading an email status needs secret_key, a project secret API key with the "
     "transactional_email:read or transactional_email:write scope."
 )
+_MANAGE_KEY_NEEDED = (
+    "Moving or canceling a scheduled email needs secret_key, a project secret API key "
+    "with the transactional_email:write scope."
+)
+_LIST_KEY_NEEDED = (
+    "Listing scheduled emails needs secret_key, a project secret API key with the "
+    "transactional_email:read or transactional_email:write scope."
+)
 _ResultT = TypeVar("_ResultT")
 _Parse = Callable[[int, Any, Optional[str]], _ResultT]
 
@@ -927,10 +1199,20 @@ def send(self, **kwargs: Any) -> SendEmailResult:
                 outside this call, too.
             max_retries: Overrides the client's maximum number of retries for
                 this send. ``0`` turns retries off.
+            scheduled_at: When to send the email, up to 30 days after this
+                call: an aware ``datetime``, or an ISO 8601 string with an
+                offset, such as ``"2026-12-01T09:00:00+02:00"``. A naive
+                ``datetime`` or a string without an offset raises
+                ``ValueError`` before any request, so the email never goes out
+                in the wrong timezone. A time at or before now sends now. Move
+                or stop a scheduled email with :meth:`update` and
+                :meth:`cancel`, and find it with :meth:`list`. The idempotency
+                key covers the time: the same key with another time raises
+                ``idempotency_key_reused``.
 
         Returns:
-            The queued email. A result means PostHog queued the email, not that
-            it was delivered.
+            The accepted email. A result means PostHog queued or scheduled the
+            email, not that it was delivered. ``status`` says which.
 
         Raises:
             EmailError: PostHog did not queue the email, or the client could
@@ -953,6 +1235,11 @@ def send(self, **kwargs: Any) -> SendEmailResult:
                 html=rendered_html,
                 distinct_id="user_123",
             )
+            emails.send(
+                to="user@example.com",
+                template_id="0190f6d2-7c1e-7b3a-9f5e-1c2d3e4f5a6b",
+                scheduled_at=datetime.now(timezone.utc) + timedelta(days=1),
+            )
             ```
         """
         prepared = _prepare(kwargs)
@@ -976,7 +1263,9 @@ def _deliver(self, prepared: _PreparedSend, secret_key: str) -> SendEmailResult:
             prepared.path,
             prepared.body,
             headers,
-            _parse_response,
+            lambda status, payload, retry_after: _parse_response(
+                status, payload, retry_after, prepared.scheduled
+            ),
             prepared.max_retries,
         )
 
@@ -1012,6 +1301,125 @@ def get(self, message_id: Union[str, UUID]) -> EmailStatus:
         finally:
             finish()
 
+    def update(
+        self, message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]
+    ) -> UpdateEmailResult:
+        """Move a scheduled email to another send time.
+
+        Only the time changes. To change the content, the recipient or the
+        sender, :meth:`cancel` the email and send it again.
+
+        Retries like :meth:`send`. A move to a later time ends the same
+        however often it lands, so it takes no idempotency key. A move to now
+        starts the send, so when its answer is lost, the retry can raise
+        ``already_sending`` for an email that was moved: read it with
+        :meth:`get`.
+
+        Args:
+            message_id: The ``id`` that :meth:`send` returned.
+            scheduled_at: The new send time, as :meth:`send` takes it. It must
+                be within 30 days after the first send call. A time at or
+                before now sends the email now.
+
+        Returns:
+            The email with its new send time.
+
+        Raises:
+            EmailError: The email was not moved. ``code`` is
+                ``already_sending`` once PostHog has started sending it,
+                ``canceled`` after a cancel, ``invalid_scheduled_at`` for a time
+                too far ahead, and ``not_found`` for an unknown ID or an email
+                that went out. ``name`` is ``validation_error`` for the first
+                three: a retry can't change them.
+            TypeError, ValueError: An argument has the wrong type or format.
+
+        Examples:
+            ```python
+            emails.update(result.id, scheduled_at=datetime.now(timezone.utc) + timedelta(hours=2))
+            ```
+        """
+        path = _status_path(message_id)
+        body = _prepare_update(scheduled_at)
+        finish = self._start(_UPDATE_NOT_DONE)
+        try:
+            headers = _headers(
+                self._secret_key(_MANAGE_KEY_NEEDED), self.migrated_from, has_body=True
+            )
+            return self._call("PATCH", path, body, headers, _parse_update_response)
+        finally:
+            finish()
+
+    def cancel(self, message_id: Union[str, UUID]) -> CancelEmailResult:
+        """Stop a scheduled email before it goes out.
+
+        A canceled email never goes out, and :meth:`get` reads ``canceled``.
+        Canceling it again also succeeds. Retries like :meth:`send`.
+
+        Args:
+            message_id: The ``id`` that :meth:`send` returned.
+
+        Returns:
+            The canceled email.
+
+        Raises:
+            EmailError: The email was not canceled. ``code`` is
+                ``already_sending``, named ``validation_error``, once PostHog
+                has started sending it, and ``not_found`` for an unknown ID or
+                an email that went out.
+            ValueError: ``message_id`` is not a message ID.
+        """
+        path = _cancel_path(message_id)
+        finish = self._start(_CANCEL_NOT_DONE)
+        try:
+            headers = _headers(self._secret_key(_MANAGE_KEY_NEEDED), self.migrated_from)
+            return self._call("POST", path, None, headers, _parse_cancel_response)
+        finally:
+            finish()
+
+    def list(
+        self,
+        *,
+        status: Literal["scheduled"] = "scheduled",
+        cursor: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> EmailList:
+        """List the project's emails that wait for their send time, the soonest first.
+
+        Reads one page. Pass the page's ``next`` as ``cursor`` to read the
+        next one. A list is never retried: call it again.
+
+        Args:
+            status: ``"scheduled"``, the only status PostHog lists.
+            cursor: The ``next`` of the previous page.
+            limit: Emails per page, 1 to 100. PostHog's default is 100.
+
+        Returns:
+            One page of scheduled emails.
+
+        Raises:
+            EmailError: PostHog did not answer with a page it could read, such
+                as one without its ``next`` cursor.
+            TypeError, ValueError: An argument has the wrong type or format.
+
+        Examples:
+            ```python
+            page = emails.list()
+            while True:
+                for email in page.results:
+                    print(email.id, email.to, email.scheduled_at)
+                if page.next is None:
+                    break
+                page = emails.list(cursor=page.next)
+            ```
+        """
+        path = _list_path(status, cursor, limit)
+        finish = self._start(_LIST_NOT_DONE)
+        try:
+            headers = _headers(self._secret_key(_LIST_KEY_NEEDED), self.migrated_from)
+            return self._call("GET", path, None, headers, _parse_list_response, 0)
+        finally:
+            finish()
+
     def _call(
         self,
         method: str,
@@ -1160,7 +1568,9 @@ async def _deliver(
             prepared.path,
             prepared.body,
             headers,
-            _parse_response,
+            lambda status, payload, retry_after: _parse_response(
+                status, payload, retry_after, prepared.scheduled
+            ),
             prepared.max_retries,
         )
 
@@ -1178,6 +1588,61 @@ async def get(self, message_id: Union[str, UUID]) -> EmailStatus:
         finally:
             finish()
 
+    async def update(
+        self, message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]
+    ) -> UpdateEmailResult:
+        """Move a scheduled email to another send time.
+
+        Takes the same arguments and raises the same errors as
+        :meth:`Emails.update`.
+        """
+        path = _status_path(message_id)
+        body = _prepare_update(scheduled_at)
+        finish = self._start(_UPDATE_NOT_DONE)
+        try:
+            headers = _headers(
+                self._secret_key(_MANAGE_KEY_NEEDED), self.migrated_from, has_body=True
+            )
+            return await self._call(
+                "PATCH", path, body, headers, _parse_update_response
+            )
+        finally:
+            finish()
+
+    async def cancel(self, message_id: Union[str, UUID]) -> CancelEmailResult:
+        """Stop a scheduled email before it goes out.
+
+        Takes the same argument and raises the same errors as
+        :meth:`Emails.cancel`.
+        """
+        path = _cancel_path(message_id)
+        finish = self._start(_CANCEL_NOT_DONE)
+        try:
+            headers = _headers(self._secret_key(_MANAGE_KEY_NEEDED), self.migrated_from)
+            return await self._call("POST", path, None, headers, _parse_cancel_response)
+        finally:
+            finish()
+
+    async def list(
+        self,
+        *,
+        status: Literal["scheduled"] = "scheduled",
+        cursor: Optional[str] = None,
+        limit: Optional[int] = None,
+    ) -> EmailList:
+        """List the project's emails that wait for their send time, the soonest first.
+
+        Takes the same arguments and raises the same errors as
+        :meth:`Emails.list`.
+        """
+        path = _list_path(status, cursor, limit)
+        finish = self._start(_LIST_NOT_DONE)
+        try:
+            headers = _headers(self._secret_key(_LIST_KEY_NEEDED), self.migrated_from)
+            return await self._call("GET", path, None, headers, _parse_list_response, 0)
+        finally:
+            finish()
+
     async def _call(
         self,
         method: str,
diff --git a/posthog/test/test_emails.py b/posthog/test/test_emails.py
index b13b8a0c6..224278686 100644
--- a/posthog/test/test_emails.py
+++ b/posthog/test/test_emails.py
@@ -4,7 +4,7 @@
 import json
 import logging
 import threading
-from datetime import datetime, timezone
+from datetime import date, datetime, timedelta, timezone
 from decimal import Decimal
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from typing import Any, Iterator, Optional, Union
@@ -900,3 +900,518 @@ def test_rejects_a_code_send_that_mixes_or_misses_fields(_name, overrides, error
         )
 
     assert adapter.requests == []
+
+
+# Scheduled sends: schedule, move, cancel and list.
+
+CANCEL_URL = f"{STATUS_URL}cancel/"
+LIST_URL = "https://us.posthog.com/api/projects/@current/messaging/emails/"
+LATER = datetime(2026, 12, 1, 9, 0, tzinfo=timezone(timedelta(hours=2)))
+
+
+@parameterized.expand(
+    [
+        ("aware_datetime_goes_as_utc", LATER, "2026-12-01T07:00:00Z"),
+        ("utc_string_as_given", "2026-12-01T07:00:00Z", "2026-12-01T07:00:00Z"),
+        (
+            "offset_string_as_given",
+            "2026-12-01T09:00:00.250+02:00",
+            "2026-12-01T09:00:00.250+02:00",
+        ),
+        ("space_separated_as_given", "2026-12-01 07:00Z", "2026-12-01 07:00Z"),
+    ]
+)
+def test_send_schedules_both_forms_for_a_time_with_an_offset(_name, value, wire):
+    emails, adapter = emails_with(
+        [
+            accepted(status="scheduled", scheduled_at="2026-12-01T07:00:00Z"),
+            accepted(status="scheduled", scheduled_at="2026-12-01T07:00:00Z"),
+        ]
+    )
+
+    template_result = send(emails, scheduled_at=value)
+    code_result = send_code(emails, html=HTML, scheduled_at=value)
+
+    for result in (template_result, code_result):
+        assert result.status == "scheduled"
+        assert result.scheduled_at == datetime(2026, 12, 1, 7, tzinfo=timezone.utc)
+        assert result.warnings == ()
+    assert [
+        json.loads(request.body)["scheduled_at"] for request in adapter.requests
+    ] == [
+        wire,
+        wire,
+    ]
+
+
+@parameterized.expand(
+    [
+        ("left_out", lambda emails: send(emails)),
+        ("passed_as_none", lambda emails: send(emails, scheduled_at=None)),
+        (
+            "code_form_as_none",
+            lambda emails: send_code(emails, html=HTML, scheduled_at=None),
+        ),
+    ]
+)
+def test_a_send_now_carries_no_scheduled_at_and_reads_queued(_name, call):
+    emails, adapter = emails_with([accepted()])
+
+    result = call(emails)
+
+    assert (result.status, result.scheduled_at, result.warnings) == ("queued", None, ())
+    assert "scheduled_at" not in json.loads(adapter.requests[0].body)
+
+
+@parameterized.expand(
+    [
+        ("template_form", lambda emails: send(emails, scheduled_at=LATER)),
+        (
+            "code_form",
+            lambda emails: send_code(emails, html=HTML, scheduled_at=LATER),
+        ),
+    ]
+)
+def test_a_server_without_scheduled_sends_warns_that_it_sent_now(_name, call):
+    # A PostHog that predates scheduled sends drops scheduled_at, sends at once and answers no status.
+    emails, _ = emails_with([accepted()])
+
+    result = call(emails)
+
+    assert (result.status, result.scheduled_at) == ("queued", None)
+    assert [w.code for w in result.warnings] == ["scheduling_unsupported"]
+
+
+def test_a_time_at_or_before_now_reads_as_the_servers_queued_answer():
+    emails, _ = emails_with([accepted(status="queued", scheduled_at=None)])
+
+    result = send(emails, scheduled_at="2020-01-01T00:00:00Z")
+
+    assert (result.status, result.scheduled_at, result.warnings) == ("queued", None, ())
+
+
+@parameterized.expand(
+    [
+        ("naive_datetime", datetime(2026, 12, 1, 9), ValueError),
+        ("string_without_offset", "2026-12-01T09:00:00", ValueError),
+        ("date_only_string", "2026-12-01", ValueError),
+        ("natural_language", "in 1 min", ValueError),
+        ("not_a_date", "2026-13-01T09:00:00Z", ValueError),
+        ("empty_string", "", ValueError),
+        ("unix_timestamp", 1796115600, TypeError),
+        ("date", date(2026, 12, 1), TypeError),
+    ]
+)
+def test_refuses_a_time_without_a_timezone_before_any_request(_name, value, error):
+    emails, adapter = emails_with([])
+
+    for call in (
+        lambda: send(emails, scheduled_at=value),
+        lambda: send_code(emails, html=HTML, scheduled_at=value),
+        lambda: emails.update(MESSAGE_ID, scheduled_at=value),
+    ):
+        with pytest.raises(error, match="scheduled_at"):
+            call()
+
+    assert adapter.requests == []
+
+
+def test_update_moves_a_scheduled_email_and_returns_its_new_time():
+    emails, adapter = emails_with(
+        [
+            (
+                200,
+                {
+                    "id": MESSAGE_ID,
+                    "status": "scheduled",
+                    "scheduled_at": "2026-12-01T07:00:00Z",
+                },
+                {},
+            )
+        ]
+    )
+
+    result = emails.update(MESSAGE_ID, scheduled_at=LATER)
+
+    assert (result.id, result.status, result.scheduled_at) == (
+        MESSAGE_ID,
+        "scheduled",
+        datetime(2026, 12, 1, 7, tzinfo=timezone.utc),
+    )
+    [request] = adapter.requests
+    assert (request.method, request.url) == ("PATCH", STATUS_URL)
+    assert json.loads(request.body) == {"scheduled_at": "2026-12-01T07:00:00Z"}
+    assert request.headers["Content-Type"] == "application/json"
+    assert request.headers["Authorization"] == f"Bearer {SECRET_KEY}"
+    assert "Idempotency-Key" not in request.headers
+
+
+def test_cancel_stops_a_scheduled_email():
+    emails, adapter = emails_with([(200, {"id": MESSAGE_ID, "status": "canceled"}, {})])
+
+    result = emails.cancel(MESSAGE_ID)
+
+    assert (result.id, result.status) == (MESSAGE_ID, "canceled")
+    [request] = adapter.requests
+    assert (request.method, request.url, request.body) == ("POST", CANCEL_URL, None)
+    assert "Idempotency-Key" not in request.headers
+
+
+def list_reply(next_cursor: Optional[str] = None) -> Reply:
+    return (
+        200,
+        {
+            "results": [
+                {
+                    "id": MESSAGE_ID,
+                    "to": "user@example.com",
+                    "scheduled_at": "2026-12-01T07:00:00Z",
+                    "created_at": "2026-11-12T10:00:00.123000Z",
+                }
+            ],
+            "next": next_cursor,
+        },
+        {},
+    )
+
+
+@parameterized.expand(
+    [
+        ("first_page", {}, "?status=scheduled", "cursor_2"),
+        (
+            "next_page",
+            {"cursor": "cursor_2", "limit": 50},
+            "?status=scheduled&cursor=cursor_2&limit=50",
+            None,
+        ),
+    ]
+)
+def test_list_reads_scheduled_emails_a_page_at_a_time(_name, args, query, next_cursor):
+    emails, adapter = emails_with([list_reply(next_cursor)])
+
+    page = emails.list(**args)
+
+    [request] = adapter.requests
+    assert (request.method, request.url) == ("GET", LIST_URL + query)
+    [email] = page.results
+    assert (email.id, email.to, email.scheduled_at, email.created_at) == (
+        MESSAGE_ID,
+        "user@example.com",
+        datetime(2026, 12, 1, 7, tzinfo=timezone.utc),
+        datetime(2026, 11, 12, 10, 0, 0, 123000, tzinfo=timezone.utc),
+    )
+    assert page.next == next_cursor
+
+
+@parameterized.expand(
+    [
+        ("zero", {"limit": 0}, ValueError),
+        ("over_a_page", {"limit": 101}, ValueError),
+        ("not_an_int", {"limit": "10"}, TypeError),
+        ("a_bool", {"limit": True}, TypeError),
+        ("empty_cursor", {"cursor": ""}, ValueError),
+        ("empty_status", {"status": ""}, ValueError),
+        ("another_status", {"status": "sent"}, ValueError),
+    ]
+)
+def test_list_refuses_bad_arguments_before_any_request(_name, args, error):
+    emails, adapter = emails_with([])
+
+    with pytest.raises(error):
+        emails.list(**args)
+
+    assert adapter.requests == []
+
+
+@parameterized.expand(
+    [
+        ("update", lambda emails, id: emails.update(id, scheduled_at=LATER)),
+        ("cancel", lambda emails, id: emails.cancel(id)),
+    ]
+)
+def test_update_and_cancel_refuse_an_id_that_is_not_a_message_id(_name, call):
+    emails, adapter = emails_with([])
+
+    with pytest.raises(ValueError, match="message_id"):
+        call(emails, "msg_1")
+
+    assert adapter.requests == []
+
+
+def test_update_needs_a_time():
+    emails, adapter = emails_with([])
+
+    with pytest.raises(TypeError, match="scheduled_at"):
+        emails.update(MESSAGE_ID, scheduled_at=None)
+
+    assert adapter.requests == []
+
+
+MANAGE_CALLS = [
+    (
+        "update",
+        lambda emails: emails.update(MESSAGE_ID, scheduled_at=LATER),
+        (
+            200,
+            {
+                "id": MESSAGE_ID,
+                "status": "scheduled",
+                "scheduled_at": "2026-12-01T07:00:00Z",
+            },
+            {},
+        ),
+    ),
+    (
+        "cancel",
+        lambda emails: emails.cancel(MESSAGE_ID),
+        (200, {"id": MESSAGE_ID, "status": "canceled"}, {}),
+    ),
+]
+
+
+@parameterized.expand(
+    [
+        (f"{name}_{case}", call, success, first_reply, retried)
+        for name, call, success in MANAGE_CALLS
+        for case, first_reply, retried in [
+            ("rate_limited", rejected(429, "throttled"), True),
+            ("service_unavailable", rejected(503, "delivery_unavailable"), True),
+            ("connection_error", requests.exceptions.ConnectionError("down"), True),
+            ("already_sending", rejected(409, "already_sending"), False),
+            ("canceled", rejected(409, "canceled"), False),
+            ("not_found", rejected(404, "not_found"), False),
+            ("too_far", rejected(400, "invalid_scheduled_at"), False),
+            ("over_quota", rejected(429, "quota_limited"), False),
+        ]
+    ]
+)
+def test_update_and_cancel_retry_like_a_send(
+    _name, call, success, first_reply, retried
+):
+    emails, adapter = emails_with([first_reply, success])
+
+    if retried:
+        assert call(emails).id == MESSAGE_ID
+        assert len(adapter.requests) == 2
+    else:
+        with pytest.raises(EmailError):
+            call(emails)
+        assert len(adapter.requests) == 1
+
+
+@parameterized.expand(
+    [
+        ("move_while_sending", MANAGE_CALLS[0][1], "already_sending"),
+        ("move_after_a_cancel", MANAGE_CALLS[0][1], "canceled"),
+        ("cancel_while_sending", MANAGE_CALLS[1][1], "already_sending"),
+    ]
+)
+def test_a_cancel_or_move_too_late_is_a_validation_error(_name, call, code):
+    emails, adapter = emails_with([rejected(409, code)])
+
+    with pytest.raises(EmailError) as raised:
+        call(emails)
+
+    # As with template_outdated, a retry can't resolve it, so posthog-node names it this way too.
+    assert (raised.value.status_code, raised.value.code, raised.value.name) == (
+        409,
+        code,
+        "validation_error",
+    )
+    assert len(adapter.requests) == 1
+
+
+@parameterized.expand(
+    [
+        ("service_unavailable", rejected(503, "delivery_unavailable")),
+        ("rate_limited", rejected(429, "throttled")),
+        ("connection_error", requests.exceptions.ConnectionError("down")),
+    ]
+)
+def test_list_never_retries(_name, first_reply):
+    emails, adapter = emails_with([first_reply, list_reply()])
+
+    with pytest.raises(EmailError):
+        emails.list()
+
+    assert len(adapter.requests) == 1
+
+
+@parameterized.expand(
+    [
+        (
+            "update",
+            lambda emails: emails.update(MESSAGE_ID, scheduled_at=LATER),
+            "missing_secret_key",
+        ),
+        ("cancel", lambda emails: emails.cancel(MESSAGE_ID), "missing_secret_key"),
+        ("list", lambda emails: emails.list(), "missing_secret_key"),
+    ]
+)
+def test_manage_calls_need_the_secret_key_and_an_enabled_client(_name, call, code):
+    adapter = FakeAdapter([])
+    session = requests.Session()
+    session.mount("https://", adapter)
+
+    with pytest.raises(EmailError) as missing:
+        call(Emails(None, session=session))
+    with pytest.raises(EmailError) as disabled:
+        call(Emails(SECRET_KEY, session=session, disabled=True))
+
+    assert (missing.value.code, disabled.value.code) == (code, "client_disabled")
+    assert adapter.requests == []
+
+
+@parameterized.expand(
+    [
+        ("row_without_fields", {"results": [{"id": MESSAGE_ID}], "next": None}),
+        ("next_not_a_string", {**list_reply()[1], "next": 123}),
+        ("next_empty", {**list_reply()[1], "next": ""}),
+        ("next_left_out", {"results": list_reply()[1]["results"]}),
+    ]
+)
+def test_a_malformed_list_page_raises_instead_of_dropping_emails(_name, body):
+    # A page that loses its next cursor would end a caller's loop early and hide emails that still go out.
+    emails, _ = emails_with([(200, body, {})])
+
+    with pytest.raises(EmailError) as raised:
+        emails.list()
+
+    assert raised.value.code == "api_error"
+
+
+@parameterized.expand(
+    [
+        ("no_status", {"id": MESSAGE_ID}),
+        ("still_scheduled", {"id": MESSAGE_ID, "status": "scheduled"}),
+    ]
+)
+def test_a_cancel_reply_without_a_canceled_email_raises(_name, body):
+    emails, _ = emails_with([(200, body, {})])
+
+    with pytest.raises(EmailError) as raised:
+        emails.cancel(MESSAGE_ID)
+
+    assert raised.value.code == "api_error"
+
+
+def async_emails_with(replies: list[httpx.Response]):
+    seen: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        return replies.pop(0)
+
+    emails = AsyncEmails(
+        SECRET_KEY,
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+    )
+    return emails, seen
+
+
+def unavailable() -> httpx.Response:
+    return httpx.Response(503, json={"code": "delivery_unavailable"})
+
+
+@pytest.mark.asyncio
+async def test_async_send_schedules_an_email():
+    emails, seen = async_emails_with(
+        [
+            httpx.Response(
+                202,
+                json={
+                    "message_id": MESSAGE_ID,
+                    "status": "scheduled",
+                    "scheduled_at": "2026-12-01T07:00:00Z",
+                },
+            )
+        ]
+    )
+
+    sent = await emails.send(
+        to="user@example.com", template_id=TEMPLATE_ID, scheduled_at=LATER
+    )
+
+    assert (sent.status, sent.scheduled_at, sent.warnings) == (
+        "scheduled",
+        datetime(2026, 12, 1, 7, tzinfo=timezone.utc),
+        (),
+    )
+    assert [(request.method, str(request.url)) for request in seen] == [
+        ("POST", SEND_URL)
+    ]
+    assert json.loads(seen[0].content)["scheduled_at"] == "2026-12-01T07:00:00Z"
+
+
+@pytest.mark.asyncio
+async def test_async_update_retries_an_unavailable_service_and_moves_the_email():
+    emails, seen = async_emails_with(
+        [
+            unavailable(),
+            httpx.Response(
+                200,
+                json={
+                    "id": MESSAGE_ID,
+                    "status": "scheduled",
+                    "scheduled_at": "2026-12-02T07:00:00Z",
+                },
+            ),
+        ]
+    )
+
+    with mock.patch("posthog.emails.asyncio.sleep"):
+        moved = await emails.update(
+            MESSAGE_ID, scheduled_at="2026-12-02T09:00:00+02:00"
+        )
+
+    assert moved.scheduled_at == datetime(2026, 12, 2, 7, tzinfo=timezone.utc)
+    assert [(request.method, str(request.url)) for request in seen] == [
+        ("PATCH", STATUS_URL),
+        ("PATCH", STATUS_URL),
+    ]
+    assert json.loads(seen[0].content) == {"scheduled_at": "2026-12-02T09:00:00+02:00"}
+    assert "Idempotency-Key" not in seen[0].headers
+
+
+@pytest.mark.asyncio
+async def test_async_cancel_retries_an_unavailable_service_and_cancels_the_email():
+    emails, seen = async_emails_with(
+        [
+            unavailable(),
+            httpx.Response(200, json={"id": MESSAGE_ID, "status": "canceled"}),
+        ]
+    )
+
+    with mock.patch("posthog.emails.asyncio.sleep"):
+        canceled = await emails.cancel(MESSAGE_ID)
+
+    assert canceled.status == "canceled"
+    assert [(request.method, str(request.url)) for request in seen] == [
+        ("POST", CANCEL_URL),
+        ("POST", CANCEL_URL),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_async_list_reads_a_page():
+    emails, seen = async_emails_with([httpx.Response(200, json=list_reply()[1])])
+
+    page = await emails.list(limit=10)
+
+    assert [email.id for email in page.results] == [MESSAGE_ID]
+    assert [str(request.url) for request in seen] == [
+        LIST_URL + "?status=scheduled&limit=10"
+    ]
+
+
+@pytest.mark.asyncio
+async def test_async_list_never_retries():
+    emails, seen = async_emails_with(
+        [unavailable(), httpx.Response(200, json=list_reply()[1])]
+    )
+
+    with mock.patch("posthog.emails.asyncio.sleep"):
+        with pytest.raises(EmailError):
+            await emails.list()
+
+    assert len(seen) == 1
diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt
index 8eec881b9..b94d26fb7 100644
--- a/references/public_api_snapshot.txt
+++ b/references/public_api_snapshot.txt
@@ -912,6 +912,8 @@ attribute posthog.debug = False
 attribute posthog.default_client = None
 attribute posthog.disable_geoip = True
 attribute posthog.disabled = False
+attribute posthog.emails.CancelEmailResult.id: str
+attribute posthog.emails.CancelEmailResult.status: str
 attribute posthog.emails.CodeEmailArgs.capture_engagement: NotRequired[Optional[bool]]
 attribute posthog.emails.CodeEmailArgs.from_: NotRequired[Optional[str]]
 attribute posthog.emails.CodeEmailArgs.html: NotRequired[Optional[str]]
@@ -930,6 +932,8 @@ 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.EmailList.next: Optional[str] = None
+attribute posthog.emails.EmailList.results: tuple[ScheduledEmail, ...]
 attribute posthog.emails.EmailStatus.created_at: datetime
 attribute posthog.emails.EmailStatus.id: str
 attribute posthog.emails.EmailStatus.reason: Optional[str] = None
@@ -941,11 +945,20 @@ attribute posthog.emails.EmailTracking.opens: bool
 attribute posthog.emails.EmailWarning.code: str
 attribute posthog.emails.EmailWarning.fix: Optional[str] = None
 attribute posthog.emails.EmailWarning.message: str
+attribute posthog.emails.ScheduledEmail.created_at: datetime
+attribute posthog.emails.ScheduledEmail.id: str
+attribute posthog.emails.ScheduledEmail.scheduled_at: datetime
+attribute posthog.emails.ScheduledEmail.to: str
 attribute posthog.emails.SendEmailResult.id: str
 attribute posthog.emails.SendEmailResult.replayed: bool = False
+attribute posthog.emails.SendEmailResult.scheduled_at: Optional[datetime] = None
+attribute posthog.emails.SendEmailResult.status: str = 'queued'
 attribute posthog.emails.SendEmailResult.warnings: tuple[EmailWarning, ...] = ()
 attribute posthog.emails.TemplateEmailArgs.template_data: NotRequired[Optional[Mapping[str, Any]]]
 attribute posthog.emails.TemplateEmailArgs.template_id: Union[str, UUID]
+attribute posthog.emails.UpdateEmailResult.id: str
+attribute posthog.emails.UpdateEmailResult.scheduled_at: datetime
+attribute posthog.emails.UpdateEmailResult.status: str
 attribute posthog.enable_exception_autocapture = False
 attribute posthog.enable_exception_autocapture_rate_limiting = False
 attribute posthog.enable_full_ai_capture = False
@@ -1375,14 +1388,18 @@ class posthog.client.Client(project_api_key: str, host=None, debug=False, max_qu
 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, migrated_from: Optional[str] = None)
+class posthog.emails.CancelEmailResult(id: str, status: str)
 class posthog.emails.CodeEmailArgs 
 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.EmailList(results: tuple[ScheduledEmail, ...], next: Optional[str] = None)
 class posthog.emails.EmailStatus(id: str, status: str, created_at: datetime, status_at: datetime, reason: Optional[str] = None)
 class posthog.emails.EmailTracking 
 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, migrated_from: Optional[str] = None)
-class posthog.emails.SendEmailResult(id: str, replayed: bool = False, warnings: tuple[EmailWarning, ...] = ())
+class posthog.emails.ScheduledEmail(id: str, to: str, scheduled_at: datetime, created_at: datetime)
+class posthog.emails.SendEmailResult(id: str, replayed: bool = False, warnings: tuple[EmailWarning, ...] = (), status: str = 'queued', scheduled_at: Optional[datetime] = None)
 class posthog.emails.TemplateEmailArgs 
+class posthog.emails.UpdateEmailResult(id: str, status: str, scheduled_at: datetime)
 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)
@@ -1883,11 +1900,17 @@ 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.cancel(message_id: Union[str, UUID]) -> CancelEmailResult
 method posthog.emails.AsyncEmails.get(message_id: Union[str, UUID]) -> EmailStatus
+method posthog.emails.AsyncEmails.list(*, status: Literal['scheduled'] = 'scheduled', cursor: Optional[str] = None, limit: Optional[int] = None) -> EmailList
 method posthog.emails.AsyncEmails.send(**kwargs: Any) -> SendEmailResult
+method posthog.emails.AsyncEmails.update(message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]) -> UpdateEmailResult
+method posthog.emails.Emails.cancel(message_id: Union[str, UUID]) -> CancelEmailResult
 method posthog.emails.Emails.close() -> None
 method posthog.emails.Emails.get(message_id: Union[str, UUID]) -> EmailStatus
+method posthog.emails.Emails.list(*, status: Literal['scheduled'] = 'scheduled', cursor: Optional[str] = None, limit: Optional[int] = None) -> EmailList
 method posthog.emails.Emails.send(**kwargs: Any) -> SendEmailResult
+method posthog.emails.Emails.update(message_id: Union[str, UUID], *, scheduled_at: Union[datetime, str]) -> UpdateEmailResult
 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)