diff --git a/src/backend/core/api/jmap/__init__.py b/src/backend/core/api/jmap/__init__.py
new file mode 100644
index 000000000..29dc9d2a0
--- /dev/null
+++ b/src/backend/core/api/jmap/__init__.py
@@ -0,0 +1 @@
+"""JMAP API module for the messages application."""
diff --git a/src/backend/core/api/jmap/errors.py b/src/backend/core/api/jmap/errors.py
new file mode 100644
index 000000000..3e4dbfc9d
--- /dev/null
+++ b/src/backend/core/api/jmap/errors.py
@@ -0,0 +1,61 @@
+"""JMAP error types as per RFC 8620."""
+
+
+class JMAPError(Exception):
+    """Base JMAP error."""
+
+    error_type: str = "serverFail"
+
+    def __init__(self, description: str = ""):
+        self.description = description
+        super().__init__(description)
+
+    def to_response(self, call_id: str) -> list:
+        """Convert to JMAP error response tuple."""
+        return [
+            "error",
+            {"type": self.error_type, "description": self.description},
+            call_id,
+        ]
+
+
+class UnknownCapabilityError(JMAPError):
+    """The request used a capability not advertised in the session."""
+
+    error_type = "unknownCapability"
+
+
+class UnknownMethodError(JMAPError):
+    """The method name is not recognized."""
+
+    error_type = "unknownMethod"
+
+
+class InvalidArgumentsError(JMAPError):
+    """One or more arguments are of the wrong type or invalid."""
+
+    error_type = "invalidArguments"
+
+
+class InvalidResultReferenceError(JMAPError):
+    """A result reference could not be resolved."""
+
+    error_type = "invalidResultReference"
+
+
+class AccountNotFoundError(JMAPError):
+    """The accountId does not correspond to a valid account."""
+
+    error_type = "accountNotFound"
+
+
+class ForbiddenError(JMAPError):
+    """The action is forbidden for the authenticated user."""
+
+    error_type = "forbidden"
+
+
+class ServerFailError(JMAPError):
+    """An unexpected server error occurred."""
+
+    error_type = "serverFail"
diff --git a/src/backend/core/api/jmap/methods.py b/src/backend/core/api/jmap/methods.py
new file mode 100644
index 000000000..3ea50826c
--- /dev/null
+++ b/src/backend/core/api/jmap/methods.py
@@ -0,0 +1,861 @@
+"""JMAP method registry and handlers."""
+
+import json
+import logging
+from datetime import datetime
+from datetime import timezone as dt_timezone
+from typing import Any
+
+from django.db.models import OuterRef, Subquery, Value
+from django.db.models.functions import Concat
+from django.utils import timezone
+
+from core import models
+from core.mda.draft import create_draft
+from core.mda.outbound import prepare_outbound_message
+from core.mda.outbound_tasks import send_message_task
+
+from .errors import (
+    InvalidArgumentsError,
+    InvalidResultReferenceError,
+    UnknownMethodError,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class MethodRegistry:
+    """Registry for JMAP method handlers."""
+
+    _methods: dict[str, type["BaseMethod"]] = {}
+
+    @classmethod
+    def register(cls, method_name: str):
+        """Decorator to register a method handler."""
+
+        def decorator(handler_class: type["BaseMethod"]):
+            cls._methods[method_name] = handler_class
+            return handler_class
+
+        return decorator
+
+    @classmethod
+    def get_handler(cls, method_name: str) -> type["BaseMethod"]:
+        """Get a method handler by name."""
+        handler = cls._methods.get(method_name)
+        if not handler:
+            raise UnknownMethodError(f"Unknown method: {method_name}")
+        return handler
+
+    @classmethod
+    def get_registered_methods(cls) -> list[str]:
+        """Get list of registered method names."""
+        return list(cls._methods.keys())
+
+
+class JMAPContext:
+    """Execution context for JMAP methods."""
+
+    def __init__(self, user, results_by_call_id: dict[str, dict]):
+        self.user = user
+        self.results_by_call_id = results_by_call_id
+        self.implicit_responses: list[list] = []
+        self.current_call_id: str = ""
+
+
+class BaseMethod:
+    """Base class for JMAP method handlers."""
+
+    def __init__(self, context: JMAPContext):
+        self.context = context
+
+    def execute(self, args: dict) -> dict:
+        """Execute the method with the given arguments."""
+        raise NotImplementedError
+
+    def _get_state(self) -> str:
+        """Get current state string (timestamp-based)."""
+        return datetime.now(dt_timezone.utc).isoformat()
+
+    def _get_account_id(self, args: dict) -> str:
+        """Get and validate accountId from args."""
+        account_id = args.get("accountId")
+        if account_id is None:
+            raise InvalidArgumentsError("Missing required argument: accountId")
+        # Account ID should match the user's ID
+        if str(self.context.user.id) != str(account_id):
+            raise InvalidArgumentsError(f"Invalid accountId: {account_id}")
+        return account_id
+
+    def resolve_value(self, value: Any) -> Any:
+        """Resolve a value, handling back-references."""
+        if isinstance(value, dict) and "resultOf" in value:
+            return self._resolve_reference(value)
+        return value
+
+    def _resolve_reference(self, ref: dict) -> Any:
+        """Resolve a single ResultReference."""
+        call_id = ref.get("resultOf")
+        path = ref.get("path", "/")
+
+        if call_id not in self.context.results_by_call_id:
+            raise InvalidResultReferenceError(f"Unknown callId: {call_id}")
+
+        result = self.context.results_by_call_id[call_id]
+        return self._navigate_path(result, path)
+
+    def _navigate_path(self, obj: Any, path: str) -> Any:
+        """Navigate a JSONPointer-like path."""
+        if path in ("/", ""):
+            return obj
+
+        parts = path.strip("/").split("/")
+        current = obj
+
+        for part in parts:
+            if part == "*":
+                # Wildcard: extract from all items in list
+                if not isinstance(current, list):
+                    raise InvalidResultReferenceError(
+                        f"Cannot use '*' on non-list: {type(current)}"
+                    )
+                # Return the remaining path applied to each item
+                remaining_parts = parts[parts.index(part) + 1 :]
+                if remaining_parts:
+                    remaining_path = "/" + "/".join(remaining_parts)
+                    return [
+                        self._navigate_path(item, remaining_path) for item in current
+                    ]
+                return current
+            elif isinstance(current, dict):
+                if part not in current:
+                    raise InvalidResultReferenceError(f"Key not found: {part}")
+                current = current[part]
+            elif isinstance(current, list):
+                try:
+                    idx = int(part)
+                    current = current[idx]
+                except (ValueError, IndexError) as e:
+                    raise InvalidResultReferenceError(
+                        f"Invalid list index: {part}"
+                    ) from e
+            else:
+                raise InvalidResultReferenceError(
+                    f"Cannot navigate into {type(current)}"
+                )
+
+        return current
+
+
+def resolve_args(args: dict, context: JMAPContext) -> dict:
+    """Resolve all back-references in method arguments."""
+    resolved = {}
+    base_method = BaseMethod(context)
+
+    for key, value in args.items():
+        if key.startswith("#"):
+            # Key starts with # - this is a back-reference
+            actual_key = key[1:]
+            resolved[actual_key] = base_method.resolve_value(value)
+        else:
+            resolved[key] = value
+
+    return resolved
+
+
+# ---------- Mailbox Methods ----------
+
+
+@MethodRegistry.register("Mailbox/query")
+class MailboxQuery(BaseMethod):
+    """Query mailboxes accessible to the user."""
+
+    def execute(self, args: dict) -> dict:
+        account_id = self._get_account_id(args)
+        filter_obj = args.get("filter", {})
+        position = args.get("position", 0)
+        limit = args.get("limit")
+
+        # Get user's accessible mailboxes
+        mailboxes = models.Mailbox.objects.filter(accesses__user=self.context.user)
+
+        # Apply name filter
+        if name_filter := filter_obj.get("name"):
+            # Annotate with full email address and filter
+            mailboxes = mailboxes.annotate(
+                full_email=Concat("local_part", Value("@"), "domain__name")
+            ).filter(full_email__icontains=name_filter)
+
+        # Get IDs
+        mailbox_ids = list(mailboxes.values_list("id", flat=True))
+
+        # Apply pagination
+        if limit is not None:
+            mailbox_ids = mailbox_ids[position : position + limit]
+        else:
+            mailbox_ids = mailbox_ids[position:]
+
+        return {
+            "accountId": account_id,
+            "queryState": self._get_state(),
+            "canCalculateChanges": False,
+            "position": position,
+            "ids": [str(mid) for mid in mailbox_ids],
+            "total": len(mailbox_ids),
+        }
+
+
+@MethodRegistry.register("Mailbox/get")
+class MailboxGet(BaseMethod):
+    """Get mailbox details by IDs."""
+
+    def execute(self, args: dict) -> dict:
+        account_id = self._get_account_id(args)
+        ids = self.resolve_value(args.get("ids"))
+        properties = args.get("properties")
+
+        if ids is None:
+            # If ids is null, return all mailboxes
+            mailboxes = models.Mailbox.objects.filter(accesses__user=self.context.user)
+        else:
+            # Fetch specific mailboxes with access check
+            mailboxes = models.Mailbox.objects.filter(
+                id__in=ids, accesses__user=self.context.user
+            )
+
+        result_list = []
+        for mailbox in mailboxes.select_related("domain"):
+            mailbox_data = self._serialize_mailbox(mailbox, properties)
+            result_list.append(mailbox_data)
+
+        found_ids = {str(m.id) for m in mailboxes}
+        not_found = [mid for mid in (ids or []) if mid not in found_ids]
+
+        return {
+            "accountId": account_id,
+            "state": self._get_state(),
+            "list": result_list,
+            "notFound": not_found,
+        }
+
+    def _serialize_mailbox(
+        self, mailbox: models.Mailbox, properties: list | None
+    ) -> dict:
+        """Serialize a mailbox to JMAP format."""
+        full_email = f"{mailbox.local_part}@{mailbox.domain.name}"
+
+        data = {
+            "id": str(mailbox.id),
+            "name": full_email,
+            "role": None,  # Could map to is_identity or other logic
+            "sortOrder": 0,
+            "totalEmails": models.Message.objects.filter(
+                thread__accesses__mailbox=mailbox
+            ).count(),
+            "unreadEmails": models.Message.objects.filter(
+                thread__accesses__mailbox=mailbox, is_unread=True
+            ).count(),
+            "totalThreads": models.Thread.objects.filter(
+                accesses__mailbox=mailbox
+            ).count(),
+            "unreadThreads": models.Thread.objects.filter(
+                accesses__mailbox=mailbox, has_unread=True
+            ).count(),
+            "myRights": {
+                "mayReadItems": True,
+                "mayAddItems": True,
+                "mayRemoveItems": True,
+                "maySetSeen": True,
+                "maySetKeywords": True,
+                "mayCreateChild": False,
+                "mayRename": False,
+                "mayDelete": False,
+                "maySubmit": True,
+            },
+            "isSubscribed": True,
+        }
+
+        if properties:
+            return {k: v for k, v in data.items() if k in properties or k == "id"}
+        return data
+
+
+# ---------- Email Methods ----------
+
+
+@MethodRegistry.register("Email/query")
+class EmailQuery(BaseMethod):
+    """Query emails with filters and sorting."""
+
+    def execute(self, args: dict) -> dict:
+        account_id = self._get_account_id(args)
+        filter_obj = args.get("filter", {})
+        sort = args.get("sort", [{"property": "receivedAt", "isAscending": False}])
+        collapse_threads = args.get("collapseThreads", False)
+        position = args.get("position", 0)
+        limit = args.get("limit", 50)
+
+        # Base queryset - messages user has access to
+        messages = models.Message.objects.filter(
+            thread__accesses__mailbox__accesses__user=self.context.user
+        )
+
+        # Apply filters
+        if in_mailbox := self.resolve_value(filter_obj.get("inMailbox")):
+            messages = messages.filter(thread__accesses__mailbox_id=in_mailbox)
+
+        if after := filter_obj.get("after"):
+            # Parse ISO date string
+            if isinstance(after, str):
+                after = datetime.fromisoformat(after.replace("Z", "+00:00"))
+            messages = messages.filter(created_at__gte=after)
+
+        if before := filter_obj.get("before"):
+            if isinstance(before, str):
+                before = datetime.fromisoformat(before.replace("Z", "+00:00"))
+            messages = messages.filter(created_at__lt=before)
+
+        # Apply sorting
+        order_by = self._build_order_by(sort)
+        messages = messages.order_by(*order_by)
+
+        # Collapse threads if requested (return only latest email per thread)
+        if collapse_threads:
+            # For each thread, get the latest message that matches the filters
+            # We need to get the max id from the filtered set, grouped by thread
+
+            # Get the ID of the latest filtered message per thread
+            latest_filtered_per_thread = (
+                messages.filter(thread_id=OuterRef("thread_id"))
+                .order_by("-created_at")
+                .values("id")[:1]
+            )
+
+            # Keep only messages that are the latest filtered message in their thread
+            messages = messages.filter(
+                id=Subquery(latest_filtered_per_thread)
+            ).distinct()
+
+        # Get total before pagination
+        total = messages.count()
+
+        # Apply pagination
+        message_ids = list(
+            messages.values_list("id", flat=True)[position : position + limit]
+        )
+
+        return {
+            "accountId": account_id,
+            "queryState": self._get_state(),
+            "canCalculateChanges": False,
+            "position": position,
+            "ids": [str(mid) for mid in message_ids],
+            "total": total,
+        }
+
+    def _build_order_by(self, sort: list) -> list:
+        """Build Django order_by from JMAP sort."""
+        property_map = {
+            "receivedAt": "created_at",
+            "sentAt": "sent_at",
+            "size": "blob__size",
+            "subject": "subject",
+        }
+
+        order_by = []
+        for sort_item in sort:
+            prop = sort_item.get("property", "receivedAt")
+            is_ascending = sort_item.get("isAscending", False)
+
+            django_field = property_map.get(prop, "created_at")
+            if not is_ascending:
+                django_field = f"-{django_field}"
+            order_by.append(django_field)
+
+        return order_by or ["-created_at"]
+
+
+@MethodRegistry.register("Email/get")
+class EmailGet(BaseMethod):
+    """Get email details by IDs."""
+
+    def execute(self, args: dict) -> dict:
+        account_id = self._get_account_id(args)
+        ids = self.resolve_value(args.get("ids"))
+        properties = args.get("properties")
+
+        if ids is None:
+            raise InvalidArgumentsError("ids is required for Email/get")
+
+        messages = (
+            models.Message.objects.filter(id__in=ids)
+            .filter(thread__accesses__mailbox__accesses__user=self.context.user)
+            .select_related("sender", "thread", "blob")
+            .prefetch_related("recipients__contact")
+        )
+
+        result_list = []
+        for message in messages:
+            email_data = self._serialize_email(message, properties)
+            result_list.append(email_data)
+
+        found_ids = {str(m.id) for m in messages}
+        not_found = [mid for mid in ids if mid not in found_ids]
+
+        return {
+            "accountId": account_id,
+            "state": self._get_state(),
+            "list": result_list,
+            "notFound": not_found,
+        }
+
+    def _serialize_email(
+        self, message: models.Message, properties: list | None
+    ) -> dict:
+        """Serialize a message to JMAP Email format."""
+        # Get mailbox IDs for this message's thread
+        mailbox_ids = list(
+            models.ThreadAccess.objects.filter(thread=message.thread).values_list(
+                "mailbox_id", flat=True
+            )
+        )
+
+        # Build keywords from flags
+        keywords = {}
+        if not message.is_unread:
+            keywords["$seen"] = True
+        if message.is_starred:
+            keywords["$flagged"] = True
+        if message.is_draft:
+            keywords["$draft"] = True
+
+        # Get recipients by type
+        recipients_to = []
+        recipients_cc = []
+        recipients_bcc = []
+        for recipient in message.recipients.all():
+            addr = {"name": recipient.contact.name, "email": recipient.contact.email}
+            if recipient.type == models.MessageRecipientTypeChoices.TO:
+                recipients_to.append(addr)
+            elif recipient.type == models.MessageRecipientTypeChoices.CC:
+                recipients_cc.append(addr)
+            elif recipient.type == models.MessageRecipientTypeChoices.BCC:
+                recipients_bcc.append(addr)
+
+        # Get preview from thread snippet or generate one
+        preview = message.thread.snippet[:256] if message.thread.snippet else ""
+
+        data = {
+            "id": str(message.id),
+            "blobId": str(message.blob.id) if message.blob else None,
+            "threadId": str(message.thread.id),
+            "mailboxIds": {str(mid): True for mid in mailbox_ids},
+            "keywords": keywords,
+            "size": message.blob.size if message.blob else 0,
+            "receivedAt": (
+                message.created_at.isoformat() if message.created_at else None
+            ),
+            "sentAt": message.sent_at.isoformat() if message.sent_at else None,
+            "from": (
+                [{"name": message.sender.name, "email": message.sender.email}]
+                if message.sender
+                else []
+            ),
+            "to": recipients_to,
+            "cc": recipients_cc,
+            "bcc": recipients_bcc,
+            "subject": message.subject,
+            "preview": preview,
+            "hasAttachment": message.has_attachments,
+        }
+
+        if properties:
+            return {k: v for k, v in data.items() if k in properties or k == "id"}
+        return data
+
+
+# ---------- Thread Methods ----------
+
+
+@MethodRegistry.register("Thread/get")
+class ThreadGet(BaseMethod):
+    """Get thread details by IDs."""
+
+    def execute(self, args: dict) -> dict:
+        account_id = self._get_account_id(args)
+        ids = self.resolve_value(args.get("ids"))
+
+        if ids is None:
+            raise InvalidArgumentsError("ids is required for Thread/get")
+
+        threads = models.Thread.objects.filter(
+            id__in=ids, accesses__mailbox__accesses__user=self.context.user
+        ).prefetch_related("messages")
+
+        result_list = []
+        for thread in threads:
+            thread_data = {
+                "id": str(thread.id),
+                "emailIds": [str(m.id) for m in thread.messages.order_by("created_at")],
+            }
+            result_list.append(thread_data)
+
+        found_ids = {str(t.id) for t in threads}
+        not_found = [tid for tid in ids if tid not in found_ids]
+
+        return {
+            "accountId": account_id,
+            "state": self._get_state(),
+            "list": result_list,
+            "notFound": not_found,
+        }
+
+
+# ---------- Email/set Method ----------
+
+
+@MethodRegistry.register("Email/set")
+class EmailSetMethod(BaseMethod):
+    """Create, update, and destroy emails."""
+
+    def execute(self, args: dict) -> dict:
+        account_id = self._get_account_id(args)
+        create = args.get("create", {})
+        update = args.get("update", {})
+        destroy = args.get("destroy", [])
+
+        created = {}
+        not_created = {}
+        updated = {}
+        not_updated = {}
+        destroyed = []
+        not_destroyed = {}
+
+        # --- create ---
+        for creation_id, email_data in create.items():
+            try:
+                created[creation_id] = self._create_email(email_data)
+            except Exception as e:
+                not_created[creation_id] = {
+                    "type": "invalidArguments",
+                    "description": str(e),
+                }
+
+        # --- update ---
+        for email_id, patch in update.items():
+            try:
+                self._update_email(email_id, patch)
+                updated[email_id] = None
+            except Exception as e:
+                not_updated[email_id] = {
+                    "type": "invalidArguments",
+                    "description": str(e),
+                }
+
+        # --- destroy ---
+        for email_id in destroy:
+            try:
+                self._destroy_email(email_id)
+                destroyed.append(email_id)
+            except Exception as e:
+                not_destroyed[email_id] = {
+                    "type": "invalidArguments",
+                    "description": str(e),
+                }
+
+        return {
+            "accountId": account_id,
+            "oldState": self._get_state(),
+            "newState": self._get_state(),
+            "created": created or None,
+            "notCreated": not_created or None,
+            "updated": updated or None,
+            "notUpdated": not_updated or None,
+            "destroyed": destroyed or None,
+            "notDestroyed": not_destroyed or None,
+        }
+
+    def _create_email(self, email_data: dict) -> dict:
+        """Create a draft email from JMAP Email data."""
+        mailbox_ids = email_data.get("mailboxIds", {})
+        if not mailbox_ids:
+            raise InvalidArgumentsError("mailboxIds is required")
+
+        # Get the first mailbox the user has access to
+        mailbox_id = next(iter(mailbox_ids))
+        try:
+            mailbox = models.Mailbox.objects.get(
+                id=mailbox_id, accesses__user=self.context.user
+            )
+        except models.Mailbox.DoesNotExist:
+            raise InvalidArgumentsError(f"Mailbox not found: {mailbox_id}")
+
+        # Extract fields
+        subject = email_data.get("subject", "")
+        to_list = email_data.get("to", [])
+        cc_list = email_data.get("cc", [])
+        bcc_list = email_data.get("bcc", [])
+
+        to_emails = [addr["email"] for addr in to_list if "email" in addr]
+        cc_emails = [addr["email"] for addr in cc_list if "email" in addr]
+        bcc_emails = [addr["email"] for addr in bcc_list if "email" in addr]
+
+        # Extract body content from JMAP bodyValues / textBody / htmlBody
+        text_body = ""
+        html_body = ""
+        body_values = email_data.get("bodyValues", {})
+        for part in email_data.get("textBody", []):
+            part_id = part.get("partId")
+            if part_id and part_id in body_values:
+                text_body = body_values[part_id].get("value", "")
+        for part in email_data.get("htmlBody", []):
+            part_id = part.get("partId")
+            if part_id and part_id in body_values:
+                html_body = body_values[part_id].get("value", "")
+
+        # Store body as JSON in draft_blob so EmailSubmission can read it back
+        draft_body = json.dumps(
+            {"format": "jmap", "textBody": text_body, "htmlBody": html_body}
+        )
+
+        message = create_draft(
+            mailbox=mailbox,
+            subject=subject,
+            draft_body=draft_body,
+            to_emails=to_emails,
+            cc_emails=cc_emails,
+            bcc_emails=bcc_emails,
+        )
+
+        return {
+            "id": str(message.id),
+            "blobId": str(message.draft_blob.id) if message.draft_blob else None,
+            "threadId": str(message.thread_id),
+            "size": message.draft_blob.size if message.draft_blob else 0,
+        }
+
+    def _update_email(self, email_id: str, patch: dict) -> None:
+        """Update email keywords/flags via JMAP patch syntax."""
+        message = models.Message.objects.filter(
+            id=email_id,
+            thread__accesses__mailbox__accesses__user=self.context.user,
+        ).first()
+        if not message:
+            raise InvalidArgumentsError(f"Email not found: {email_id}")
+
+        updated_fields = []
+
+        # Handle full keywords replacement
+        if "keywords" in patch:
+            keywords = patch["keywords"]
+            is_seen = keywords.get("$seen", False)
+            is_flagged = keywords.get("$flagged", False)
+            is_draft = keywords.get("$draft", False)
+
+            message.is_unread = not is_seen
+            message.is_starred = is_flagged
+            message.is_draft = is_draft
+            updated_fields.extend(["is_unread", "is_starred", "is_draft"])
+
+        # Handle individual keyword patches (e.g. "keywords/$seen": True)
+        for key, value in patch.items():
+            if key == "keywords/$seen":
+                message.is_unread = not value
+                if "is_unread" not in updated_fields:
+                    updated_fields.append("is_unread")
+            elif key == "keywords/$flagged":
+                message.is_starred = value
+                if "is_starred" not in updated_fields:
+                    updated_fields.append("is_starred")
+            elif key == "keywords/$draft":
+                message.is_draft = value
+                if "is_draft" not in updated_fields:
+                    updated_fields.append("is_draft")
+
+        if updated_fields:
+            message.save(update_fields=updated_fields + ["updated_at"])
+
+    def _destroy_email(self, email_id: str) -> None:
+        """Trash a message (sets is_trashed=True)."""
+        message = models.Message.objects.filter(
+            id=email_id,
+            thread__accesses__mailbox__accesses__user=self.context.user,
+        ).first()
+        if not message:
+            raise InvalidArgumentsError(f"Email not found: {email_id}")
+
+        message.is_trashed = True
+        message.trashed_at = timezone.now()
+        message.save(update_fields=["is_trashed", "trashed_at", "updated_at"])
+
+
+# ---------- EmailSubmission/set Method ----------
+
+
+@MethodRegistry.register("EmailSubmission/set")
+class EmailSubmissionSetMethod(BaseMethod):
+    """Submit emails for delivery."""
+
+    def execute(self, args: dict) -> dict:
+        account_id = self._get_account_id(args)
+        create = args.get("create", {})
+        on_success_update = args.get("onSuccessUpdateEmail", {})
+
+        created = {}
+        not_created = {}
+
+        for creation_id, submission_data in create.items():
+            try:
+                result = self._create_submission(submission_data)
+                created[creation_id] = result
+
+                # Handle onSuccessUpdateEmail
+                if on_success_update:
+                    self._apply_on_success(
+                        result["emailId"], creation_id, on_success_update
+                    )
+            except Exception as e:
+                logger.exception(
+                    "Failed to create email submission %s: %s", creation_id, e
+                )
+                not_created[creation_id] = {
+                    "type": "invalidArguments",
+                    "description": str(e),
+                }
+
+        return {
+            "accountId": account_id,
+            "oldState": self._get_state(),
+            "newState": self._get_state(),
+            "created": created or None,
+            "notCreated": not_created or None,
+        }
+
+    def _create_submission(self, submission_data: dict) -> dict:
+        """Submit a draft email for delivery."""
+        email_id = submission_data.get("emailId")
+        identity_id = submission_data.get("identityId")
+
+        if not email_id:
+            raise InvalidArgumentsError("emailId is required")
+        if not identity_id:
+            raise InvalidArgumentsError("identityId is required")
+
+        # Validate identity (mailbox)
+        try:
+            mailbox = models.Mailbox.objects.get(
+                id=identity_id, accesses__user=self.context.user
+            )
+        except models.Mailbox.DoesNotExist:
+            raise InvalidArgumentsError(f"Identity not found: {identity_id}")
+
+        # Get the draft message
+        message = models.Message.objects.filter(
+            id=email_id,
+            thread__accesses__mailbox__accesses__user=self.context.user,
+        ).select_related("thread", "sender", "draft_blob", "signature").first()
+        if not message:
+            raise InvalidArgumentsError(f"Email not found: {email_id}")
+        if not message.is_draft:
+            raise InvalidArgumentsError("Email is not a draft")
+
+        # Read body from draft_blob
+        text_body = ""
+        html_body = ""
+        if message.draft_blob:
+            try:
+                blob_content = message.draft_blob.get_content()
+                body_data = json.loads(blob_content)
+                if body_data.get("format") == "jmap":
+                    text_body = body_data.get("textBody", "")
+                    html_body = body_data.get("htmlBody", "")
+                else:
+                    # Legacy format - treat as plain text
+                    text_body = blob_content.decode("utf-8")
+            except (json.JSONDecodeError, UnicodeDecodeError):
+                text_body = message.draft_blob.get_content().decode(
+                    "utf-8", errors="replace"
+                )
+
+        # Prepare and queue the message for sending
+        prepare_outbound_message(
+            mailbox_sender=mailbox,
+            message=message,
+            text_body=text_body,
+            html_body=html_body,
+            user=self.context.user,
+        )
+
+        # Queue async send task
+        send_message_task.delay(str(message.id))
+
+        return {
+            "id": str(message.id),
+            "emailId": str(message.id),
+            "threadId": str(message.thread_id),
+            "undoStatus": "final",
+        }
+
+    def _apply_on_success(
+        self, email_id: str, creation_id: str, on_success_update: dict
+    ) -> None:
+        """Apply onSuccessUpdateEmail patches as an implicit Email/set response."""
+        # Build the update map, replacing #emailSubmission/foo references with the email_id
+        update_map = {}
+        for ref_key, patch in on_success_update.items():
+            # ref_key is like "#emailSubmission/creation_id"
+            update_map[email_id] = patch
+
+        # Execute the implicit Email/set
+        implicit_args = {
+            "accountId": str(self.context.user.id),
+            "update": update_map,
+        }
+        handler = EmailSetMethod(self.context)
+        result = handler.execute(implicit_args)
+
+        # Add to implicit responses
+        call_id = self.context.current_call_id
+        self.context.implicit_responses.append(
+            ["Email/set", result, call_id]
+        )
+
+
+# ---------- Identity Methods ----------
+
+
+@MethodRegistry.register("Identity/get")
+class IdentityGetMethod(BaseMethod):
+    """Get sending identities (user's mailboxes)."""
+
+    def execute(self, args: dict) -> dict:
+        account_id = self._get_account_id(args)
+        ids = args.get("ids")
+
+        mailboxes = models.Mailbox.objects.filter(
+            accesses__user=self.context.user
+        ).select_related("domain")
+
+        if ids is not None:
+            mailboxes = mailboxes.filter(id__in=ids)
+
+        result_list = []
+        for mailbox in mailboxes:
+            full_email = f"{mailbox.local_part}@{mailbox.domain.name}"
+            result_list.append({
+                "id": str(mailbox.id),
+                "name": full_email,
+                "email": full_email,
+                "replyTo": None,
+                "mayDelete": False,
+            })
+
+        found_ids = {str(m.id) for m in mailboxes}
+        not_found = [mid for mid in (ids or []) if mid not in found_ids]
+
+        return {
+            "accountId": account_id,
+            "state": self._get_state(),
+            "list": result_list,
+            "notFound": not_found,
+        }
diff --git a/src/backend/core/api/jmap/views.py b/src/backend/core/api/jmap/views.py
new file mode 100644
index 000000000..a9b96812f
--- /dev/null
+++ b/src/backend/core/api/jmap/views.py
@@ -0,0 +1,262 @@
+"""JMAP API views."""
+
+from datetime import datetime
+from datetime import timezone as dt_timezone
+from urllib.parse import urlencode, urlparse
+
+from django.conf import settings
+from django.http import HttpResponse, HttpResponseRedirect
+from django.views import View
+
+from rest_framework.parsers import JSONParser
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from core import models
+from core.api.permissions import IsAuthenticated
+
+from .errors import (
+    JMAPError,
+)
+from .methods import JMAPContext, MethodRegistry, resolve_args
+
+# JMAP capabilities we support
+JMAP_CAPABILITIES = {
+    "urn:ietf:params:jmap:core": {
+        "maxSizeUpload": 50000000,
+        "maxConcurrentUpload": 4,
+        "maxSizeRequest": 10000000,
+        "maxConcurrentRequests": 4,
+        "maxCallsInRequest": 16,
+        "maxObjectsInGet": 500,
+        "maxObjectsInSet": 500,
+        "collationAlgorithms": ["i;ascii-casemap", "i;octet"],
+    },
+    "urn:ietf:params:jmap:mail": {
+        "maxMailboxesPerEmail": None,
+        "maxMailboxDepth": None,
+        "maxSizeMailboxName": 255,
+        "maxSizeAttachmentsPerEmail": 50000000,
+        "emailQuerySortOptions": ["receivedAt", "sentAt", "size", "subject"],
+        "mayCreateTopLevelMailbox": True,
+    },
+    "urn:ietf:params:jmap:submission": {
+        "maxDelayedSend": 0,
+        "submissionExtensions": {},
+    },
+}
+
+
+class JMAPSessionView(APIView):
+    """
+    JMAP Session Resource.
+
+    GET request returns the session object with capabilities, accounts, and URLs.
+    """
+
+    permission_classes = [IsAuthenticated]
+
+    def get(self, request):
+        """Return the JMAP session object."""
+        user = request.user
+
+        # Get user's accessible mailboxes for account info
+        mailboxes = models.Mailbox.objects.filter(accesses__user=user).select_related(
+            "domain"
+        )
+
+        # Use the first mailbox email as the account name, or user email
+        primary_email = None
+        for mailbox in mailboxes[:1]:
+            primary_email = f"{mailbox.local_part}@{mailbox.domain.name}"
+            break
+
+        if not primary_email:
+            primary_email = user.email or str(user.id)
+
+        # Build account
+        account_id = str(user.id)
+        accounts = {
+            account_id: {
+                "name": primary_email,
+                "isPersonal": True,
+                "isReadOnly": False,
+                "accountCapabilities": {
+                    "urn:ietf:params:jmap:mail": {},
+                },
+            }
+        }
+
+        # Build the base URL for API endpoints
+        base_url = request.build_absolute_uri(f"/api/{settings.API_VERSION}/jmap/")
+
+        session = {
+            "capabilities": JMAP_CAPABILITIES,
+            "accounts": accounts,
+            "primaryAccounts": {
+                "urn:ietf:params:jmap:mail": account_id,
+            },
+            "username": primary_email,
+            "apiUrl": base_url,
+            "downloadUrl": f"{base_url}download/{{accountId}}/{{blobId}}/{{name}}",
+            "uploadUrl": f"{base_url}upload/{{accountId}}/",
+            "eventSourceUrl": f"{base_url}eventsource/?types={{types}}&closeafter={{closeafter}}&ping={{ping}}",
+            "state": datetime.now(dt_timezone.utc).isoformat(),
+        }
+
+        return Response(session)
+
+
+class JMAPOIDCLoginView(View):
+    """
+    OIDC login helper for the JMAP demo webmail.
+
+    Initiates the existing backend OIDC flow and redirects back to
+    the demo with the access token in the URL fragment.
+
+    GET /api/v1.0/jmap/oidc-login?redirect_uri=<demo_url>
+    """
+
+    def get(self, request):
+        redirect_uri = request.GET.get("redirect_uri", "")
+
+        # Store redirect_uri in session on first visit
+        if redirect_uri:
+            # Validate redirect_uri to prevent open redirects
+            parsed = urlparse(redirect_uri)
+            if not parsed.scheme or not parsed.netloc:
+                return HttpResponse("Invalid redirect_uri", status=400)
+            request.session["jmap_oidc_redirect_uri"] = redirect_uri
+
+        if not request.user.is_authenticated:
+            # Redirect to the existing OIDC login flow,
+            # with next= pointing back here after authentication
+            next_url = f"/api/{settings.API_VERSION}/jmap/oidc-login"
+            authenticate_url = (
+                f"/api/{settings.API_VERSION}/authenticate/?{urlencode({'next': next_url})}"
+            )
+            return HttpResponseRedirect(authenticate_url)
+
+        # User is authenticated — retrieve the access token from session
+        stored_redirect = request.session.pop("jmap_oidc_redirect_uri", "")
+        access_token = request.session.get("oidc_access_token", "")
+
+        if not stored_redirect:
+            return HttpResponse(
+                "Missing redirect_uri. Start the flow with ?redirect_uri=<url>",
+                status=400,
+            )
+
+        if not access_token:
+            return HttpResponse(
+                "No OIDC access token in session. "
+                "Ensure OIDC_STORE_ACCESS_TOKEN=True is set.",
+                status=500,
+            )
+
+        # Redirect back to the demo with the token in the URL fragment
+        return HttpResponseRedirect(f"{stored_redirect}#access_token={access_token}")
+
+
+class JMAPAPIView(APIView):
+    """
+    JMAP API endpoint for method calls.
+
+    POST request accepts JSON body with method calls and returns responses.
+    """
+
+    permission_classes = [IsAuthenticated]
+    parser_classes = [JSONParser]
+
+    def post(self, request):
+        """Process JMAP method calls."""
+        data = request.data
+
+        # Validate request structure
+        if not isinstance(data, dict):
+            return Response(
+                {"type": "urn:ietf:params:jmap:error:notRequest"},
+                status=400,
+            )
+
+        using = data.get("using", [])
+        method_calls = data.get("methodCalls", [])
+
+        if not isinstance(using, list) or not isinstance(method_calls, list):
+            return Response(
+                {"type": "urn:ietf:params:jmap:error:notRequest"},
+                status=400,
+            )
+
+        # Validate capabilities
+        for capability in using:
+            if capability not in JMAP_CAPABILITIES:
+                return Response(
+                    {
+                        "type": "urn:ietf:params:jmap:error:unknownCapability",
+                        "unknownCapabilities": [capability],
+                    },
+                    status=400,
+                )
+
+        # Process method calls
+        method_responses = []
+        results_by_call_id = {}
+        context = JMAPContext(request.user, results_by_call_id)
+
+        for call in method_calls:
+            if not isinstance(call, list) or len(call) != 3:
+                method_responses.append(
+                    [
+                        "error",
+                        {
+                            "type": "invalidArguments",
+                            "description": "Method call must be [name, args, callId]",
+                        },
+                        call[2] if len(call) > 2 else "unknown",
+                    ]
+                )
+                continue
+
+            method_name, args, call_id = call
+
+            try:
+                # Resolve back-references in arguments
+                resolved_args = resolve_args(args, context)
+
+                # Set current call_id for implicit responses
+                context.current_call_id = call_id
+                context.implicit_responses = []
+
+                # Get and execute the method handler
+                handler_class = MethodRegistry.get_handler(method_name)
+                handler = handler_class(context)
+                result = handler.execute(resolved_args)
+
+                # Store result for back-references
+                results_by_call_id[call_id] = result
+
+                method_responses.append([method_name, result, call_id])
+
+                # Append any implicit responses (e.g. from onSuccessUpdateEmail)
+                for implicit in context.implicit_responses:
+                    method_responses.append(implicit)
+
+            except JMAPError as e:
+                method_responses.append(e.to_response(call_id))
+            except Exception as e:
+                # Catch any unexpected errors
+                method_responses.append(
+                    [
+                        "error",
+                        {"type": "serverFail", "description": str(e)},
+                        call_id,
+                    ]
+                )
+
+        response = {
+            "methodResponses": method_responses,
+            "sessionState": datetime.now(dt_timezone.utc).isoformat(),
+        }
+
+        return Response(response)
diff --git a/src/backend/core/tests/api/jmap/__init__.py b/src/backend/core/tests/api/jmap/__init__.py
new file mode 100644
index 000000000..7ee58fe21
--- /dev/null
+++ b/src/backend/core/tests/api/jmap/__init__.py
@@ -0,0 +1 @@
+"""JMAP API tests."""
diff --git a/src/backend/core/tests/api/jmap/conftest.py b/src/backend/core/tests/api/jmap/conftest.py
new file mode 100644
index 000000000..57025debd
--- /dev/null
+++ b/src/backend/core/tests/api/jmap/conftest.py
@@ -0,0 +1,328 @@
+"""Fixtures for JMAP API tests."""
+
+from datetime import timedelta
+
+from django.conf import settings
+from django.utils import timezone
+
+import pytest
+from jmapc import Ref
+from rest_framework.test import APIClient
+
+from core import enums, factories
+
+
+@pytest.fixture
+def api_client():
+    """Provide an instance of the API client for tests."""
+    return APIClient()
+
+
+@pytest.fixture
+def user():
+    """Create a test user."""
+    return factories.UserFactory()
+
+
+@pytest.fixture
+def mailbox(user):
+    """Create a mailbox with user access."""
+    return factories.MailboxFactory(users_read=[user])
+
+
+@pytest.fixture
+def mailbox_with_threads(user):
+    """Create a mailbox with multiple threads and messages."""
+    mailbox = factories.MailboxFactory(users_read=[user])
+
+    # Create 5 threads with messages
+    for i in range(5):
+        thread = factories.ThreadFactory(subject=f"Thread {i}")
+        factories.ThreadAccessFactory(
+            mailbox=mailbox,
+            thread=thread,
+            role=enums.ThreadAccessRoleChoices.EDITOR,
+        )
+
+        # Create a contact for the sender in this mailbox
+        sender = factories.ContactFactory(mailbox=mailbox)
+
+        # Create 1-3 messages per thread
+        for j in range(1, (i % 3) + 2):
+            factories.MessageFactory(
+                thread=thread,
+                subject=f"Message {j} in Thread {i}",
+                sender=sender,
+                created_at=timezone.now() - timedelta(days=i, hours=j),
+            )
+
+        thread.update_stats()
+
+    return mailbox
+
+
+class JMAPTestClient:
+    """
+    A test client that adapts DRF's APIClient to work with jmapc method objects.
+
+    This allows tests to use jmapc's MailboxQuery, EmailGet, etc. classes
+    while still using Django's test client for actual HTTP requests.
+    """
+
+    def __init__(self, api_client: APIClient, user, account_id: str):
+        self.api_client = api_client
+        self.user = user
+        self.account_id = account_id
+        self.api_url = f"/api/{settings.API_VERSION}/jmap/"
+        self.session_url = f"/api/{settings.API_VERSION}/jmap/session"
+
+        # Authenticate the client
+        self.api_client.force_authenticate(user=user)
+
+    def get_session(self):
+        """Get the JMAP session."""
+        response = self.api_client.get(self.session_url)
+        assert response.status_code == 200, f"Session request failed: {response.data}"
+        return response.data
+
+    def request(self, methods, raise_errors: bool = True):
+        """
+        Execute JMAP method calls.
+
+        Args:
+            methods: A single jmapc Method object or a list of Method objects.
+            raise_errors: If True, raise an exception on JMAP errors.
+
+        Returns:
+            List of response tuples or single response if one method was passed.
+        """
+        # Normalize to list
+        if not isinstance(methods, (list, tuple)):
+            methods = [methods]
+            single = True
+        else:
+            single = False
+
+        # Build the JMAP request
+        method_calls = []
+        using = {"urn:ietf:params:jmap:core"}
+
+        # First pass: generate all call_ids for Ref resolution
+        self._call_ids = []
+        for i, method in enumerate(methods):
+            method_name = self._get_method_name(method)
+            call_id = f"{i}.{method_name}"
+            self._call_ids.append(call_id)
+
+        # Second pass: serialize methods with Ref resolution
+        for i, method in enumerate(methods):
+            method_name = self._get_method_name(method)
+            args = self._serialize_method(method, call_index=i)
+            call_id = self._call_ids[i]
+            method_calls.append([method_name, args, call_id])
+
+            # Add required capabilities
+            if hasattr(method, "using"):
+                using.update(method.using)
+            if (
+                "Mail" in method_name
+                or "Email" in method_name
+                or "Thread" in method_name
+            ):
+                using.add("urn:ietf:params:jmap:mail")
+            if "Submission" in method_name or "Identity" in method_name:
+                using.add("urn:ietf:params:jmap:submission")
+
+        request_data = {
+            "using": list(using),
+            "methodCalls": method_calls,
+        }
+
+        # Make the request
+        response = self.api_client.post(self.api_url, request_data, format="json")
+        assert response.status_code == 200, f"JMAP request failed: {response.data}"
+
+        # Parse responses
+        method_responses = response.data.get("methodResponses", [])
+        results = []
+
+        for resp in method_responses:
+            method_name, result, call_id = resp
+
+            if method_name == "error" and raise_errors:
+                raise JMAPTestError(
+                    f"JMAP error: {result.get('type')}: {result.get('description')}"
+                )
+
+            results.append(JMAPTestResponse(method_name, result, call_id))
+
+        if single and len(results) == 1:
+            return results[0]
+        return results
+
+    def _get_method_name(self, method) -> str:
+        """Get the JMAP method name from a jmapc method object."""
+        if hasattr(method, "jmap_method_name"):
+            return method.jmap_method_name
+        # Fallback: construct from class name
+        class_name = method.__class__.__name__
+        # Convert CamelCase to JMAP format (e.g., MailboxQuery -> Mailbox/query)
+        for noun in ["Mailbox", "EmailSubmission", "Email", "Thread", "Identity"]:
+            if class_name.startswith(noun):
+                verb = class_name[len(noun) :].lower()
+                return f"{noun}/{verb}"
+        return class_name
+
+    def _serialize_method(self, method, call_index: int = 0) -> dict:
+        """Serialize a jmapc method object to a dict of arguments."""
+        # Always use manual serialization to handle Ref objects properly
+        # (jmapc's to_dict() tries to resolve Refs which requires context)
+        data = {}
+
+        # Get dataclass fields if available
+        if hasattr(method, "__dataclass_fields__"):
+            for field_name, _field_info in method.__dataclass_fields__.items():
+                if field_name.startswith("_"):
+                    continue
+                value = getattr(method, field_name, None)
+                if value is not None:
+                    # Convert snake_case to camelCase
+                    camel_name = self._to_camel_case(field_name)
+                    data[camel_name] = value
+        else:
+            # Fallback: serialize all non-private attributes
+            for field_name in dir(method):
+                if field_name.startswith("_"):
+                    continue
+                if field_name in ("using", "jmap_method_name", "to_dict", "to_json"):
+                    continue
+                value = getattr(method, field_name)
+                if callable(value):
+                    continue
+                if value is not None:
+                    camel_name = self._to_camel_case(field_name)
+                    data[camel_name] = value
+
+        # Process Ref objects in the data and convert to JMAP back-reference format
+        data = self._process_refs(data, call_index)
+
+        # Always add accountId
+        if "accountId" not in data:
+            data["accountId"] = self.account_id
+
+        return data
+
+    def _process_refs(self, data: dict, call_index: int) -> dict:
+        """Process Ref objects and convert them to JMAP back-reference format."""
+        result = {}
+        for key, value in data.items():
+            if isinstance(value, Ref):
+                # Convert Ref to JMAP ResultReference format
+                # Ref.method can be -1 (previous), a string (method name), or int (index)
+                ref_method = getattr(value, "method", -1)
+                if ref_method == -1:
+                    # Reference the previous method
+                    ref_index = call_index - 1
+                    ref_call_id = self._call_ids[ref_index] if ref_index >= 0 else "0"
+                    ref_method_name = (
+                        ref_call_id.split(".", 1)[1]
+                        if "." in ref_call_id
+                        else ref_call_id
+                    )
+                elif isinstance(ref_method, int):
+                    ref_call_id = self._call_ids[ref_method]
+                    ref_method_name = (
+                        ref_call_id.split(".", 1)[1]
+                        if "." in ref_call_id
+                        else ref_call_id
+                    )
+                else:
+                    ref_method_name = ref_method
+                    # Find the call_id for this method
+                    ref_call_id = None
+                    for cid in self._call_ids:
+                        if ref_method_name in cid:
+                            ref_call_id = cid
+                            break
+                    if not ref_call_id:
+                        ref_call_id = f"0.{ref_method_name}"
+
+                # Use #key format for JMAP back-reference
+                result[f"#{key}"] = {
+                    "resultOf": ref_call_id,
+                    "name": ref_method_name,
+                    "path": value.path,
+                }
+            elif isinstance(value, dict):
+                # Check if this is already a serialized Ref (from to_dict())
+                if "__ref" in value and value.get("__ref") == "Ref":
+                    # This is a serialized Ref object
+                    ref_method = value.get("method", -1)
+                    path = value.get("path", "/")
+                    if ref_method == -1:
+                        ref_index = call_index - 1
+                        ref_call_id = (
+                            self._call_ids[ref_index] if ref_index >= 0 else "0"
+                        )
+                        ref_method_name = (
+                            ref_call_id.split(".", 1)[1]
+                            if "." in ref_call_id
+                            else ref_call_id
+                        )
+                    else:
+                        ref_method_name = str(ref_method)
+                        ref_call_id = f"0.{ref_method_name}"
+                    result[f"#{key}"] = {
+                        "resultOf": ref_call_id,
+                        "name": ref_method_name,
+                        "path": path,
+                    }
+                else:
+                    result[key] = self._process_refs(value, call_index)
+            else:
+                result[key] = value
+        return result
+
+    def _to_camel_case(self, snake_str: str) -> str:
+        """Convert snake_case to camelCase."""
+        components = snake_str.split("_")
+        return components[0] + "".join(x.title() for x in components[1:])
+
+
+class JMAPTestResponse:
+    """A response from a JMAP method call."""
+
+    def __init__(self, method_name: str, data: dict, call_id: str):
+        self.method_name = method_name
+        self.data = data
+        self.call_id = call_id
+
+    @property
+    def ids(self):
+        """Get the 'ids' field from the response (for query results)."""
+        return self.data.get("ids", [])
+
+    @property
+    def list(self):
+        """Get the 'list' field from the response (for get results)."""
+        return self.data.get("list", [])
+
+    @property
+    def not_found(self):
+        """Get the 'notFound' field from the response."""
+        return self.data.get("notFound", [])
+
+    def __repr__(self):
+        return f"JMAPTestResponse({self.method_name}, {self.call_id})"
+
+
+class JMAPTestError(Exception):
+    """Exception raised when a JMAP method call fails."""
+
+    pass
+
+
+@pytest.fixture
+def jmap_client(api_client, user):
+    """Create a JMAP test client for the user."""
+    return JMAPTestClient(api_client, user, str(user.id))
diff --git a/src/backend/core/tests/api/jmap/test_jmap_methods.py b/src/backend/core/tests/api/jmap/test_jmap_methods.py
new file mode 100644
index 000000000..370115b6e
--- /dev/null
+++ b/src/backend/core/tests/api/jmap/test_jmap_methods.py
@@ -0,0 +1,640 @@
+"""Tests for JMAP methods using jmapc library."""
+
+import json
+from datetime import timedelta
+from unittest.mock import patch
+
+from django.utils import timezone
+
+import pytest
+from jmapc import Ref
+from jmapc.methods import (
+    EmailGet,
+    EmailQuery,
+    MailboxGet,
+    MailboxQuery,
+    ThreadGet,
+)
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from core import enums, factories, models
+from core.tests.api.jmap.conftest import JMAPTestError
+
+pytestmark = pytest.mark.django_db
+
+
+# ---------- Helper method classes for testing new JMAP methods ----------
+
+
+@dataclass
+class EmailSet:
+    """Test helper for Email/set."""
+
+    jmap_method_name: str = field(default="Email/set", init=False, repr=False)
+    create: dict[str, Any] | None = None
+    update: dict[str, Any] | None = None
+    destroy: list[str] | None = None
+
+
+@dataclass
+class EmailSubmissionSet:
+    """Test helper for EmailSubmission/set."""
+
+    jmap_method_name: str = field(
+        default="EmailSubmission/set", init=False, repr=False
+    )
+    create: dict[str, Any] | None = None
+    on_success_update_email: dict[str, Any] | None = None
+
+
+@dataclass
+class IdentityGet:
+    """Test helper for Identity/get."""
+
+    jmap_method_name: str = field(default="Identity/get", init=False, repr=False)
+    ids: list[str] | None = None
+
+
+class TestMailboxQuery:
+    """Tests for Mailbox/query method."""
+
+    def test_mailbox_query_returns_accessible_mailboxes(self, jmap_client, mailbox):
+        """Test that Mailbox/query returns mailboxes the user has access to."""
+        result = jmap_client.request(MailboxQuery())
+
+        assert str(mailbox.id) in result.ids
+
+    def test_mailbox_query_filters_by_name(self, jmap_client, user):
+        """Test that Mailbox/query can filter by name."""
+        import uuid
+
+        # Create mailboxes with specific names using unique domain
+        domain_name = f"filter-{uuid.uuid4().hex[:8]}.com"
+        domain = factories.MailDomainFactory(name=domain_name)
+        mailbox1 = factories.MailboxFactory(
+            local_part="inbox",
+            domain=domain,
+            users_read=[user],
+        )
+        factories.MailboxFactory(
+            local_part="other",
+            domain=domain,
+            users_read=[user],
+        )
+
+        result = jmap_client.request(
+            MailboxQuery(filter={"name": f"inbox@{domain_name}"})
+        )
+
+        assert str(mailbox1.id) in result.ids
+        assert len(result.ids) == 1
+
+    def test_mailbox_query_excludes_inaccessible_mailboxes(self, jmap_client, user):
+        """Test that Mailbox/query excludes mailboxes without access."""
+        # Create a mailbox the user can access
+        accessible = factories.MailboxFactory(users_read=[user])
+        # Create a mailbox the user cannot access
+        inaccessible = factories.MailboxFactory()
+
+        result = jmap_client.request(MailboxQuery())
+
+        assert str(accessible.id) in result.ids
+        assert str(inaccessible.id) not in result.ids
+
+
+class TestMailboxGet:
+    """Tests for Mailbox/get method."""
+
+    def test_mailbox_get_returns_mailbox_details(self, jmap_client, mailbox):
+        """Test that Mailbox/get returns mailbox details."""
+        result = jmap_client.request(MailboxGet(ids=[str(mailbox.id)]))
+
+        assert len(result.list) == 1
+        mailbox_data = result.list[0]
+        assert mailbox_data["id"] == str(mailbox.id)
+        assert "name" in mailbox_data
+        assert "totalEmails" in mailbox_data
+        assert "unreadEmails" in mailbox_data
+
+    def test_mailbox_get_with_back_reference(self, jmap_client, user):
+        """Test that Mailbox/get works with back-references from Mailbox/query."""
+        mailbox = factories.MailboxFactory(
+            local_part="test", domain__name="backref-example.com", users_read=[user]
+        )
+
+        results = jmap_client.request(
+            [
+                MailboxQuery(filter={"name": "test@backref-example.com"}),
+                MailboxGet(ids=Ref("/ids")),  # References ids from previous method
+            ]
+        )
+
+        assert str(mailbox.id) in results[0].ids
+        assert len(results[1].list) == 1
+        assert results[1].list[0]["id"] == str(mailbox.id)
+
+    def test_mailbox_get_returns_not_found(self, jmap_client):
+        """Test that Mailbox/get returns notFound for non-existent IDs."""
+        fake_id = "00000000-0000-0000-0000-000000000000"
+
+        result = jmap_client.request(MailboxGet(ids=[fake_id]))
+
+        assert len(result.list) == 0
+        assert fake_id in result.not_found
+
+
+class TestEmailQuery:
+    """Tests for Email/query method."""
+
+    def test_email_query_returns_emails(self, jmap_client, mailbox_with_threads):
+        """Test that Email/query returns emails the user has access to."""
+        result = jmap_client.request(EmailQuery())
+
+        assert len(result.ids) > 0
+
+    def test_email_query_filters_by_mailbox(self, jmap_client, user):
+        """Test that Email/query filters by inMailbox."""
+        # Create two mailboxes with messages
+        mailbox1 = factories.MailboxFactory(users_read=[user])
+        mailbox2 = factories.MailboxFactory(users_read=[user])
+
+        thread1 = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox1, thread=thread1, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender1 = factories.ContactFactory(mailbox=mailbox1)
+        msg1 = factories.MessageFactory(thread=thread1, sender=sender1)
+
+        thread2 = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox2, thread=thread2, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender2 = factories.ContactFactory(mailbox=mailbox2)
+        factories.MessageFactory(thread=thread2, sender=sender2)
+
+        result = jmap_client.request(EmailQuery(filter={"inMailbox": str(mailbox1.id)}))
+
+        assert str(msg1.id) in result.ids
+
+    def test_email_query_with_collapse_threads(self, jmap_client, user):
+        """Test that Email/query with collapseThreads returns one email per thread."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+        thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox, thread=thread, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(mailbox=mailbox)
+
+        # Create 3 messages in the same thread
+        for i in range(3):
+            factories.MessageFactory(
+                thread=thread,
+                sender=sender,
+                created_at=timezone.now() - timedelta(hours=i),
+            )
+
+        result = jmap_client.request(EmailQuery(collapse_threads=True))
+
+        # Should only return 1 email (the latest) since all are in the same thread
+        thread_ids_seen = set()
+        for email_id in result.ids:
+            # Get the email to check its threadId
+            email_result = jmap_client.request(EmailGet(ids=[email_id]))
+            thread_id = email_result.list[0]["threadId"]
+            assert thread_id not in thread_ids_seen, "Same thread appeared twice"
+            thread_ids_seen.add(thread_id)
+
+    def test_email_query_sorts_by_received_at(self, jmap_client, user):
+        """Test that Email/query sorts by receivedAt."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+        thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox, thread=thread, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(mailbox=mailbox)
+
+        # Create messages at different times
+        msg_old = factories.MessageFactory(
+            thread=thread, sender=sender, created_at=timezone.now() - timedelta(days=2)
+        )
+        msg_new = factories.MessageFactory(
+            thread=thread, sender=sender, created_at=timezone.now()
+        )
+
+        result = jmap_client.request(
+            EmailQuery(sort=[{"property": "receivedAt", "isAscending": False}])
+        )
+
+        # Newest should be first
+        assert result.ids.index(str(msg_new.id)) < result.ids.index(str(msg_old.id))
+
+
+class TestEmailGet:
+    """Tests for Email/get method."""
+
+    def test_email_get_returns_email_details(self, jmap_client, user):
+        """Test that Email/get returns email details."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+        thread = factories.ThreadFactory(subject="Test Subject")
+        factories.ThreadAccessFactory(
+            mailbox=mailbox, thread=thread, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(
+            name="Sender", email="sender@test.com", mailbox=mailbox
+        )
+        message = factories.MessageFactory(
+            thread=thread,
+            subject="Test Email",
+            sender=sender,
+        )
+
+        result = jmap_client.request(EmailGet(ids=[str(message.id)]))
+
+        assert len(result.list) == 1
+        email = result.list[0]
+        assert email["id"] == str(message.id)
+        assert email["threadId"] == str(thread.id)
+        assert email["subject"] == "Test Email"
+        assert email["from"][0]["email"] == "sender@test.com"
+
+    def test_email_get_with_properties(self, jmap_client, user):
+        """Test that Email/get respects the properties parameter."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+        thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox, thread=thread, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(mailbox=mailbox)
+        message = factories.MessageFactory(thread=thread, sender=sender)
+
+        result = jmap_client.request(
+            EmailGet(ids=[str(message.id)], properties=["threadId", "subject"])
+        )
+
+        email = result.list[0]
+        assert "id" in email  # id is always included
+        assert "threadId" in email
+        assert "subject" in email
+        assert "from" not in email  # not requested
+
+    def test_email_get_includes_keywords(self, jmap_client, user):
+        """Test that Email/get includes keywords (flags)."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+        thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox, thread=thread, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(mailbox=mailbox)
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender,
+            is_unread=False,  # $seen
+            is_starred=True,  # $flagged
+            is_draft=True,  # $draft
+        )
+
+        result = jmap_client.request(EmailGet(ids=[str(message.id)]))
+
+        keywords = result.list[0]["keywords"]
+        assert keywords.get("$seen") is True
+        assert keywords.get("$flagged") is True
+        assert keywords.get("$draft") is True
+
+
+class TestThreadGet:
+    """Tests for Thread/get method."""
+
+    def test_thread_get_returns_thread_with_email_ids(self, jmap_client, user):
+        """Test that Thread/get returns thread with emailIds."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+        thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox, thread=thread, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(mailbox=mailbox)
+
+        # Create multiple messages in thread
+        msg1 = factories.MessageFactory(
+            thread=thread, sender=sender, created_at=timezone.now() - timedelta(hours=2)
+        )
+        msg2 = factories.MessageFactory(
+            thread=thread, sender=sender, created_at=timezone.now() - timedelta(hours=1)
+        )
+        msg3 = factories.MessageFactory(
+            thread=thread, sender=sender, created_at=timezone.now()
+        )
+
+        result = jmap_client.request(ThreadGet(ids=[str(thread.id)]))
+
+        assert len(result.list) == 1
+        thread_data = result.list[0]
+        assert thread_data["id"] == str(thread.id)
+        assert len(thread_data["emailIds"]) == 3
+
+        # Check order (oldest to newest)
+        assert thread_data["emailIds"] == [str(msg1.id), str(msg2.id), str(msg3.id)]
+
+    def test_thread_get_returns_not_found(self, jmap_client):
+        """Test that Thread/get returns notFound for non-existent IDs."""
+        fake_id = "00000000-0000-0000-0000-000000000000"
+
+        result = jmap_client.request(ThreadGet(ids=[fake_id]))
+
+        assert len(result.list) == 0
+        assert fake_id in result.not_found
+
+
+class TestEmailSet:
+    """Tests for Email/set method."""
+
+    def test_create_draft(self, jmap_client, user):
+        """Test creating a draft email via Email/set."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+
+        result = jmap_client.request(
+            EmailSet(
+                create={
+                    "draft1": {
+                        "mailboxIds": {str(mailbox.id): True},
+                        "subject": "Test Draft",
+                        "from": [
+                            {
+                                "name": "Sender",
+                                "email": f"{mailbox.local_part}@{mailbox.domain.name}",
+                            }
+                        ],
+                        "to": [{"name": "Recipient", "email": "to@example.com"}],
+                        "bodyValues": {"1": {"value": "Hello, world!"}},
+                        "textBody": [{"partId": "1", "type": "text/plain"}],
+                        "keywords": {"$draft": True},
+                    }
+                }
+            )
+        )
+
+        assert result.data["created"] is not None
+        created = result.data["created"]["draft1"]
+        assert "id" in created
+        assert "threadId" in created
+
+        # Verify the message was actually created
+        message = models.Message.objects.get(id=created["id"])
+        assert message.is_draft is True
+        assert message.subject == "Test Draft"
+
+    def test_create_draft_with_html_body(self, jmap_client, user):
+        """Test creating a draft with HTML body."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+
+        result = jmap_client.request(
+            EmailSet(
+                create={
+                    "draft1": {
+                        "mailboxIds": {str(mailbox.id): True},
+                        "subject": "HTML Draft",
+                        "to": [{"name": "Recipient", "email": "to@example.com"}],
+                        "bodyValues": {
+                            "t1": {"value": "Plain text body"},
+                            "h1": {"value": "<p>HTML body</p>"},
+                        },
+                        "textBody": [{"partId": "t1", "type": "text/plain"}],
+                        "htmlBody": [{"partId": "h1", "type": "text/html"}],
+                    }
+                }
+            )
+        )
+
+        created = result.data["created"]["draft1"]
+        message = models.Message.objects.get(id=created["id"])
+        assert message.is_draft is True
+
+        # Verify body stored in draft_blob
+        blob_content = json.loads(message.draft_blob.get_content())
+        assert blob_content["format"] == "jmap"
+        assert blob_content["textBody"] == "Plain text body"
+        assert blob_content["htmlBody"] == "<p>HTML body</p>"
+
+    def test_create_draft_missing_mailbox(self, jmap_client, user):
+        """Test creating a draft with a nonexistent mailbox returns error."""
+        fake_id = "00000000-0000-0000-0000-000000000000"
+
+        result = jmap_client.request(
+            EmailSet(
+                create={
+                    "draft1": {
+                        "mailboxIds": {fake_id: True},
+                        "subject": "Test",
+                    }
+                }
+            ),
+            raise_errors=False,
+        )
+
+        assert result.data["notCreated"] is not None
+        assert "draft1" in result.data["notCreated"]
+
+    def test_update_keywords(self, jmap_client, user):
+        """Test updating email keywords via Email/set."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+        thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox, thread=thread, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(mailbox=mailbox)
+        message = factories.MessageFactory(
+            thread=thread,
+            sender=sender,
+            is_unread=True,
+            is_starred=False,
+        )
+
+        result = jmap_client.request(
+            EmailSet(
+                update={
+                    str(message.id): {
+                        "keywords/$seen": True,
+                        "keywords/$flagged": True,
+                    }
+                }
+            )
+        )
+
+        assert result.data["updated"] is not None
+        assert str(message.id) in result.data["updated"]
+
+        message.refresh_from_db()
+        assert message.is_unread is False
+        assert message.is_starred is True
+
+    def test_destroy_email(self, jmap_client, user):
+        """Test destroying (trashing) an email via Email/set."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+        thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox, thread=thread, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(mailbox=mailbox)
+        message = factories.MessageFactory(thread=thread, sender=sender)
+
+        result = jmap_client.request(
+            EmailSet(destroy=[str(message.id)])
+        )
+
+        assert result.data["destroyed"] is not None
+        assert str(message.id) in result.data["destroyed"]
+
+        message.refresh_from_db()
+        assert message.is_trashed is True
+
+
+class TestEmailSubmission:
+    """Tests for EmailSubmission/set method."""
+
+    @patch("core.api.jmap.methods.send_message_task")
+    def test_submit_draft(self, mock_send_task, jmap_client, user):
+        """Test submitting a draft for delivery."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+
+        # First create a draft via Email/set
+        create_result = jmap_client.request(
+            EmailSet(
+                create={
+                    "draft1": {
+                        "mailboxIds": {str(mailbox.id): True},
+                        "subject": "Send Test",
+                        "to": [{"name": "Recipient", "email": "to@example.com"}],
+                        "bodyValues": {"1": {"value": "Message body"}},
+                        "textBody": [{"partId": "1", "type": "text/plain"}],
+                    }
+                }
+            )
+        )
+        email_id = create_result.data["created"]["draft1"]["id"]
+
+        # Submit it
+        result = jmap_client.request(
+            EmailSubmissionSet(
+                create={
+                    "sub1": {
+                        "emailId": email_id,
+                        "identityId": str(mailbox.id),
+                    }
+                }
+            )
+        )
+
+        assert result.data["created"] is not None
+        submission = result.data["created"]["sub1"]
+        assert submission["emailId"] == email_id
+        assert submission["undoStatus"] == "final"
+
+        # Verify the message is no longer a draft
+        message = models.Message.objects.get(id=email_id)
+        assert message.is_draft is False
+
+        # Verify send_message_task was queued
+        mock_send_task.delay.assert_called_once_with(email_id)
+
+    def test_submit_non_draft_fails(self, jmap_client, user):
+        """Test that submitting a non-draft email fails."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+        thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=mailbox, thread=thread, role=enums.ThreadAccessRoleChoices.EDITOR
+        )
+        sender = factories.ContactFactory(mailbox=mailbox)
+        message = factories.MessageFactory(
+            thread=thread, sender=sender, is_draft=False
+        )
+
+        result = jmap_client.request(
+            EmailSubmissionSet(
+                create={
+                    "sub1": {
+                        "emailId": str(message.id),
+                        "identityId": str(mailbox.id),
+                    }
+                }
+            ),
+            raise_errors=False,
+        )
+
+        assert result.data["notCreated"] is not None
+        assert "sub1" in result.data["notCreated"]
+
+    def test_submit_invalid_identity_fails(self, jmap_client, user):
+        """Test that submitting with an invalid identity fails."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+
+        # Create a draft first
+        create_result = jmap_client.request(
+            EmailSet(
+                create={
+                    "draft1": {
+                        "mailboxIds": {str(mailbox.id): True},
+                        "subject": "Test",
+                        "to": [{"name": "R", "email": "r@example.com"}],
+                        "bodyValues": {"1": {"value": "body"}},
+                        "textBody": [{"partId": "1", "type": "text/plain"}],
+                    }
+                }
+            )
+        )
+        email_id = create_result.data["created"]["draft1"]["id"]
+
+        fake_identity = "00000000-0000-0000-0000-000000000000"
+        result = jmap_client.request(
+            EmailSubmissionSet(
+                create={
+                    "sub1": {
+                        "emailId": email_id,
+                        "identityId": fake_identity,
+                    }
+                }
+            ),
+            raise_errors=False,
+        )
+
+        assert result.data["notCreated"] is not None
+        assert "sub1" in result.data["notCreated"]
+
+
+class TestIdentityGet:
+    """Tests for Identity/get method."""
+
+    def test_list_identities(self, jmap_client, user):
+        """Test listing all identities."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+
+        result = jmap_client.request(IdentityGet())
+
+        assert len(result.list) >= 1
+        identity_ids = [i["id"] for i in result.list]
+        assert str(mailbox.id) in identity_ids
+
+        # Check identity fields
+        identity = next(i for i in result.list if i["id"] == str(mailbox.id))
+        expected_email = f"{mailbox.local_part}@{mailbox.domain.name}"
+        assert identity["email"] == expected_email
+        assert identity["name"] == expected_email
+        assert identity["mayDelete"] is False
+
+    def test_filter_by_id(self, jmap_client, user):
+        """Test getting a specific identity by ID."""
+        mailbox = factories.MailboxFactory(users_read=[user])
+
+        result = jmap_client.request(IdentityGet(ids=[str(mailbox.id)]))
+
+        assert len(result.list) == 1
+        assert result.list[0]["id"] == str(mailbox.id)
+
+    def test_identity_not_found(self, jmap_client, user):
+        """Test that nonexistent identity IDs appear in notFound."""
+        fake_id = "00000000-0000-0000-0000-000000000000"
+
+        result = jmap_client.request(IdentityGet(ids=[fake_id]))
+
+        assert len(result.list) == 0
+        assert fake_id in result.not_found
diff --git a/src/backend/core/tests/api/jmap/test_jmap_session.py b/src/backend/core/tests/api/jmap/test_jmap_session.py
new file mode 100644
index 000000000..9277ad29a
--- /dev/null
+++ b/src/backend/core/tests/api/jmap/test_jmap_session.py
@@ -0,0 +1,81 @@
+"""Tests for the JMAP session endpoint."""
+
+from django.conf import settings
+
+import pytest
+
+from core import factories
+
+pytestmark = pytest.mark.django_db
+
+
+class TestJMAPSession:
+    """Tests for GET /api/v1.0/jmap/session."""
+
+    def test_session_returns_capabilities(self, api_client, user):
+        """Test that the session endpoint returns JMAP capabilities."""
+        api_client.force_authenticate(user=user)
+
+        response = api_client.get(f"/api/{settings.API_VERSION}/jmap/session")
+
+        assert response.status_code == 200
+        assert "capabilities" in response.data
+        assert "urn:ietf:params:jmap:core" in response.data["capabilities"]
+        assert "urn:ietf:params:jmap:mail" in response.data["capabilities"]
+
+    def test_session_returns_account(self, api_client, user, mailbox):
+        """Test that the session includes the user's account."""
+        api_client.force_authenticate(user=user)
+
+        response = api_client.get(f"/api/{settings.API_VERSION}/jmap/session")
+
+        assert response.status_code == 200
+        assert "accounts" in response.data
+        assert str(user.id) in response.data["accounts"]
+
+        account = response.data["accounts"][str(user.id)]
+        assert account["isPersonal"] is True
+        assert "urn:ietf:params:jmap:mail" in account["accountCapabilities"]
+
+    def test_session_returns_primary_account(self, api_client, user, mailbox):
+        """Test that the session includes primary accounts."""
+        api_client.force_authenticate(user=user)
+
+        response = api_client.get(f"/api/{settings.API_VERSION}/jmap/session")
+
+        assert response.status_code == 200
+        assert "primaryAccounts" in response.data
+        assert response.data["primaryAccounts"]["urn:ietf:params:jmap:mail"] == str(
+            user.id
+        )
+
+    def test_session_returns_api_url(self, api_client, user):
+        """Test that the session includes the API URL."""
+        api_client.force_authenticate(user=user)
+
+        response = api_client.get(f"/api/{settings.API_VERSION}/jmap/session")
+
+        assert response.status_code == 200
+        assert "apiUrl" in response.data
+        assert f"/api/{settings.API_VERSION}/jmap/" in response.data["apiUrl"]
+
+    def test_session_requires_authentication(self, api_client):
+        """Test that the session endpoint requires authentication."""
+        response = api_client.get(f"/api/{settings.API_VERSION}/jmap/session")
+
+        assert response.status_code == 401
+
+    def test_session_uses_mailbox_email_as_name(self, api_client, user):
+        """Test that account name uses the mailbox email address."""
+        factories.MailboxFactory(
+            local_part="john.doe",
+            domain__name="example.com",
+            users_read=[user],
+        )
+        api_client.force_authenticate(user=user)
+
+        response = api_client.get(f"/api/{settings.API_VERSION}/jmap/session")
+
+        assert response.status_code == 200
+        account = response.data["accounts"][str(user.id)]
+        assert account["name"] == "john.doe@example.com"
diff --git a/src/backend/core/tests/api/jmap/test_jmap_workflow.py b/src/backend/core/tests/api/jmap/test_jmap_workflow.py
new file mode 100644
index 000000000..4e496f6cd
--- /dev/null
+++ b/src/backend/core/tests/api/jmap/test_jmap_workflow.py
@@ -0,0 +1,345 @@
+"""End-to-end workflow tests for JMAP API using jmapc library.
+
+These tests replicate the workflow from jmapc's recent_threads.py example.
+"""
+
+import uuid
+from datetime import timedelta
+from unittest.mock import patch
+
+from django.utils import timezone
+
+import pytest
+from jmapc import Ref
+from jmapc.methods import (
+    EmailGet,
+    EmailQuery,
+    MailboxGet,
+    MailboxQuery,
+    ThreadGet,
+)
+
+from core import enums, factories, models
+from core.tests.api.jmap.test_jmap_methods import (
+    EmailSet,
+    EmailSubmissionSet,
+    IdentityGet,
+)
+
+pytestmark = pytest.mark.django_db
+
+
+class TestRecentThreadsWorkflow:
+    """Tests for the 'read recent threads' workflow from jmapc example."""
+
+    def test_recent_threads_full_workflow(self, jmap_client, user):
+        """
+        Test the complete 'recent threads' workflow:
+        1. Find the inbox mailbox
+        2. Query recent emails with collapseThreads
+        3. Get thread details for each email
+        """
+        # Setup: Create a mailbox with threads and messages
+        # Use unique domain name to avoid conflicts
+        domain_name = f"workflow-{uuid.uuid4().hex[:8]}.com"
+        mailbox = factories.MailboxFactory(
+            local_part="inbox",
+            domain__name=domain_name,
+            users_read=[user],
+        )
+
+        # Create 3 threads with multiple messages
+        threads_data = []
+        for i in range(3):
+            thread = factories.ThreadFactory(subject=f"Thread {i}")
+            factories.ThreadAccessFactory(
+                mailbox=mailbox,
+                thread=thread,
+                role=enums.ThreadAccessRoleChoices.EDITOR,
+            )
+            sender = factories.ContactFactory(mailbox=mailbox)
+
+            messages = []
+            for j in range(2):  # 2 messages per thread
+                msg = factories.MessageFactory(
+                    thread=thread,
+                    subject=f"Message {j} in Thread {i}",
+                    sender=sender,
+                    created_at=timezone.now() - timedelta(days=i, hours=j),
+                )
+                messages.append(msg)
+
+            thread.update_stats()
+            threads_data.append({"thread": thread, "messages": messages})
+
+        # Step 1: Find the inbox mailbox
+        results = jmap_client.request(
+            [
+                MailboxQuery(filter={"name": "inbox"}),
+                MailboxGet(ids=Ref("/ids")),
+            ]
+        )
+
+        assert len(results[0].ids) == 1
+        assert str(mailbox.id) in results[0].ids
+        inbox = results[1].list[0]
+        assert inbox["name"] == f"inbox@{domain_name}"
+
+        # Step 2: Query recent emails with collapseThreads
+        results = jmap_client.request(
+            [
+                EmailQuery(
+                    filter={"inMailbox": inbox["id"]},
+                    sort=[{"property": "receivedAt", "isAscending": False}],
+                    collapse_threads=True,
+                    limit=5,
+                ),
+                EmailGet(
+                    ids=Ref("/ids"),
+                    properties=["threadId", "subject", "from", "receivedAt"],
+                ),
+            ]
+        )
+
+        # Should get one email per thread (3 threads, but collapseThreads)
+        query_result = results[0]
+        assert len(query_result.ids) == 3
+
+        emails = results[1].list
+        assert len(emails) == 3
+
+        # Each email should have the required properties
+        for email in emails:
+            assert "threadId" in email
+            assert "subject" in email
+            assert "from" in email
+            assert "receivedAt" in email
+
+        # Step 3: Get thread details
+        thread_ids = [email["threadId"] for email in emails]
+        results = jmap_client.request(ThreadGet(ids=thread_ids))
+
+        assert len(results.list) == 3
+        for thread in results.list:
+            assert "id" in thread
+            assert "emailIds" in thread
+            assert len(thread["emailIds"]) == 2  # Each thread has 2 messages
+
+    def test_workflow_with_date_filter(self, jmap_client, user):
+        """Test the workflow with a date filter (like 7 days in jmapc example)."""
+        domain_name = f"datefilter-{uuid.uuid4().hex[:8]}.com"
+        mailbox = factories.MailboxFactory(
+            local_part="inbox",
+            domain__name=domain_name,
+            users_read=[user],
+        )
+
+        sender = factories.ContactFactory(mailbox=mailbox)
+
+        # Create recent thread (within 7 days)
+        recent_thread = factories.ThreadFactory(subject="Recent")
+        factories.ThreadAccessFactory(
+            mailbox=mailbox,
+            thread=recent_thread,
+            role=enums.ThreadAccessRoleChoices.EDITOR,
+        )
+        recent_msg = factories.MessageFactory(
+            thread=recent_thread,
+            sender=sender,
+        )
+        # Update created_at directly in DB (auto_now_add prevents setting it on create)
+        from core.models import Message
+
+        Message.objects.filter(id=recent_msg.id).update(
+            created_at=timezone.now() - timedelta(days=2)
+        )
+        recent_msg.refresh_from_db()
+
+        # Create old thread (older than 7 days)
+        old_thread = factories.ThreadFactory(subject="Old")
+        factories.ThreadAccessFactory(
+            mailbox=mailbox,
+            thread=old_thread,
+            role=enums.ThreadAccessRoleChoices.EDITOR,
+        )
+        old_msg = factories.MessageFactory(
+            thread=old_thread,
+            sender=sender,
+        )
+        Message.objects.filter(id=old_msg.id).update(
+            created_at=timezone.now() - timedelta(days=14)
+        )
+
+        # Query emails from last 7 days
+        seven_days_ago = (timezone.now() - timedelta(days=7)).isoformat()
+
+        result = jmap_client.request(
+            EmailQuery(
+                filter={
+                    "inMailbox": str(mailbox.id),
+                    "after": seven_days_ago,
+                },
+                collapse_threads=True,
+            )
+        )
+
+        # Should only get the recent email
+        assert len(result.ids) == 1, (
+            f"Expected 1 message, got {len(result.ids)}: {result.ids}"
+        )
+        assert str(recent_msg.id) in result.ids
+
+    def test_workflow_empty_mailbox(self, jmap_client, user):
+        """Test the workflow with an empty mailbox."""
+        domain_name = f"empty-{uuid.uuid4().hex[:8]}.com"
+        mailbox = factories.MailboxFactory(
+            local_part="empty",
+            domain__name=domain_name,
+            users_read=[user],
+        )
+
+        # Query emails in empty mailbox
+        result = jmap_client.request(
+            EmailQuery(
+                filter={"inMailbox": str(mailbox.id)},
+                collapse_threads=True,
+            )
+        )
+
+        assert len(result.ids) == 0
+
+    def test_workflow_multiple_mailboxes(self, jmap_client, user):
+        """Test the workflow with multiple mailboxes."""
+        # Create two mailboxes on the same domain
+        domain_name = f"multi-{uuid.uuid4().hex[:8]}.com"
+        domain = factories.MailDomainFactory(name=domain_name)
+        inbox = factories.MailboxFactory(
+            local_part="inbox",
+            domain=domain,
+            users_read=[user],
+        )
+        sent = factories.MailboxFactory(
+            local_part="sent",
+            domain=domain,
+            users_read=[user],
+        )
+
+        # Add messages to each
+        inbox_sender = factories.ContactFactory(mailbox=inbox)
+        inbox_thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=inbox,
+            thread=inbox_thread,
+            role=enums.ThreadAccessRoleChoices.EDITOR,
+        )
+        factories.MessageFactory(thread=inbox_thread, sender=inbox_sender)
+
+        sent_sender = factories.ContactFactory(mailbox=sent)
+        sent_thread = factories.ThreadFactory()
+        factories.ThreadAccessFactory(
+            mailbox=sent,
+            thread=sent_thread,
+            role=enums.ThreadAccessRoleChoices.EDITOR,
+        )
+        factories.MessageFactory(thread=sent_thread, sender=sent_sender)
+
+        # Query all mailboxes
+        result = jmap_client.request(MailboxQuery())
+
+        # Should find both mailboxes
+        assert len(result.ids) == 2
+        assert str(inbox.id) in result.ids
+        assert str(sent.id) in result.ids
+
+        # Query emails in inbox only
+        inbox_result = jmap_client.request(
+            EmailQuery(filter={"inMailbox": str(inbox.id)})
+        )
+
+        # Should only get inbox emails
+        assert len(inbox_result.ids) == 1
+
+
+class TestSendWorkflow:
+    """Tests for the full message sending workflow."""
+
+    @patch("core.api.jmap.methods.send_message_task")
+    def test_full_send_workflow(self, mock_send_task, jmap_client, user):
+        """
+        Test the complete send workflow:
+        1. Identity/get - discover sending identities
+        2. Email/set create - create a draft
+        3. EmailSubmission/set create - submit for delivery
+        4. Verify message is no longer a draft
+        """
+        mailbox = factories.MailboxFactory(users_read=[user])
+
+        # Step 1: Get identities
+        identity_result = jmap_client.request(IdentityGet())
+
+        assert len(identity_result.list) >= 1
+        identity = next(
+            i for i in identity_result.list if i["id"] == str(mailbox.id)
+        )
+        identity_id = identity["id"]
+        sender_email = identity["email"]
+
+        # Step 2: Create a draft
+        create_result = jmap_client.request(
+            EmailSet(
+                create={
+                    "draft1": {
+                        "mailboxIds": {identity_id: True},
+                        "subject": "Workflow Test Email",
+                        "from": [{"name": "Me", "email": sender_email}],
+                        "to": [
+                            {"name": "Recipient", "email": "recipient@example.com"}
+                        ],
+                        "bodyValues": {
+                            "text": {"value": "Hello from the workflow test!"},
+                            "html": {
+                                "value": "<p>Hello from the workflow test!</p>"
+                            },
+                        },
+                        "textBody": [{"partId": "text", "type": "text/plain"}],
+                        "htmlBody": [{"partId": "html", "type": "text/html"}],
+                        "keywords": {"$draft": True},
+                    }
+                }
+            )
+        )
+
+        assert create_result.data["created"] is not None
+        draft = create_result.data["created"]["draft1"]
+        email_id = draft["id"]
+        thread_id = draft["threadId"]
+
+        # Verify draft was created correctly
+        message = models.Message.objects.get(id=email_id)
+        assert message.is_draft is True
+        assert message.subject == "Workflow Test Email"
+
+        # Step 3: Submit for delivery
+        submit_result = jmap_client.request(
+            EmailSubmissionSet(
+                create={
+                    "sub1": {
+                        "emailId": email_id,
+                        "identityId": identity_id,
+                    }
+                }
+            )
+        )
+
+        assert submit_result.data["created"] is not None
+        submission = submit_result.data["created"]["sub1"]
+        assert submission["emailId"] == email_id
+        assert submission["threadId"] == thread_id
+        assert submission["undoStatus"] == "final"
+
+        # Step 4: Verify message is no longer a draft
+        message.refresh_from_db()
+        assert message.is_draft is False
+
+        # Verify send task was queued
+        mock_send_task.delay.assert_called_once_with(email_id)
diff --git a/src/backend/core/urls.py b/src/backend/core/urls.py
index 892079039..e1e999239 100644
--- a/src/backend/core/urls.py
+++ b/src/backend/core/urls.py
@@ -5,6 +5,7 @@
 
 from rest_framework.routers import DefaultRouter
 
+from core.api.jmap.views import JMAPAPIView, JMAPOIDCLoginView, JMAPSessionView
 from core.api.viewsets.blob import BlobViewSet
 from core.api.viewsets.config import ConfigView
 from core.api.viewsets.contacts import ContactViewSet
@@ -224,6 +225,22 @@
         InboundMTAViewSet.as_view({"post": "deliver"}),
         name="mta-inbound-email",
     ),
+    # JMAP endpoints
+    path(
+        f"api/{settings.API_VERSION}/jmap/session",
+        JMAPSessionView.as_view(),
+        name="jmap-session",
+    ),
+    path(
+        f"api/{settings.API_VERSION}/jmap/",
+        JMAPAPIView.as_view(),
+        name="jmap-api",
+    ),
+    path(
+        f"api/{settings.API_VERSION}/jmap/oidc-login",
+        JMAPOIDCLoginView.as_view(),
+        name="jmap-oidc-login",
+    ),
 ]
 
 if settings.DRIVE_CONFIG.get("base_url"):
diff --git a/src/backend/messages/urls.py b/src/backend/messages/urls.py
index 4493ed1dd..93a71235b 100644
--- a/src/backend/messages/urls.py
+++ b/src/backend/messages/urls.py
@@ -11,6 +11,7 @@
 from django.http import HttpResponse
 from django.urls import include, path, re_path
 
+from core.api.jmap.views import JMAPSessionView
 from drf_spectacular.views import (
     SpectacularJSONAPIView,
     SpectacularRedocView,
@@ -33,6 +34,11 @@ def heartbeat(request):
 urlpatterns = [
     path(settings.ADMIN_URL, admin.site.urls),
     path("", include("core.urls")),
+    path(
+        ".well-known/jmap",
+        JMAPSessionView.as_view(),
+        name="well-known-jmap",
+    ),
     path(
         "__heartbeat__/",
         heartbeat,
diff --git a/src/backend/poetry.lock b/src/backend/poetry.lock
index f62576621..835e852d1 100644
--- a/src/backend/poetry.lock
+++ b/src/backend/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 2.2.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 2.1.4 and should not be changed by hand.
 
 [[package]]
 name = "aiofiles"
@@ -189,6 +189,117 @@ urllib3 = {version = ">=1.25.4,<2.2.0 || >2.2.0,<3", markers = "python_version >
 [package.extras]
 crt = ["awscrt (==0.27.6)"]
 
+[[package]]
+name = "brotli"
+version = "1.2.0"
+description = "Python bindings for the Brotli compression library"
+optional = true
+python-versions = "*"
+groups = ["main"]
+markers = "extra == \"dev\""
+files = [
+    {file = "brotli-1.2.0-cp27-cp27m-macosx_10_9_x86_64.whl", hash = "sha256:99cfa69813d79492f0e5d52a20fd18395bc82e671d5d40bd5a91d13e75e468e8"},
+    {file = "brotli-1.2.0-cp27-cp27m-manylinux1_i686.whl", hash = "sha256:3ebe801e0f4e56d17cd386ca6600573e3706ce1845376307f5d2cbd32149b69a"},
+    {file = "brotli-1.2.0-cp27-cp27m-manylinux1_x86_64.whl", hash = "sha256:a387225a67f619bf16bd504c37655930f910eb03675730fc2ad69d3d8b5e7e92"},
+    {file = "brotli-1.2.0-cp27-cp27m-win32.whl", hash = "sha256:b908d1a7b28bc72dfb743be0d4d3f8931f8309f810af66c906ae6cd4127c93cb"},
+    {file = "brotli-1.2.0-cp27-cp27m-win_amd64.whl", hash = "sha256:d206a36b4140fbb5373bf1eb73fb9de589bb06afd0d22376de23c5e91d0ab35f"},
+    {file = "brotli-1.2.0-cp27-cp27mu-manylinux1_i686.whl", hash = "sha256:7e9053f5fb4e0dfab89243079b3e217f2aea4085e4d58c5c06115fc34823707f"},
+    {file = "brotli-1.2.0-cp27-cp27mu-manylinux1_x86_64.whl", hash = "sha256:4735a10f738cb5516905a121f32b24ce196ab82cfc1e4ba2e3ad1b371085fd46"},
+    {file = "brotli-1.2.0-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:3b90b767916ac44e93a8e28ce6adf8d551e43affb512f2377c732d486ac6514e"},
+    {file = "brotli-1.2.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:6be67c19e0b0c56365c6a76e393b932fb0e78b3b56b711d180dd7013cb1fd984"},
+    {file = "brotli-1.2.0-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0bbd5b5ccd157ae7913750476d48099aaf507a79841c0d04a9db4415b14842de"},
+    {file = "brotli-1.2.0-cp310-cp310-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:3f3c908bcc404c90c77d5a073e55271a0a498f4e0756e48127c35d91cf155947"},
+    {file = "brotli-1.2.0-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:1b557b29782a643420e08d75aea889462a4a8796e9a6cf5621ab05a3f7da8ef2"},
+    {file = "brotli-1.2.0-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:81da1b229b1889f25adadc929aeb9dbc4e922bd18561b65b08dd9343cfccca84"},
+    {file = "brotli-1.2.0-cp310-cp310-musllinux_1_2_ppc64le.whl", hash = "sha256:ff09cd8c5eec3b9d02d2408db41be150d8891c5566addce57513bf546e3d6c6d"},
+    {file = "brotli-1.2.0-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:a1778532b978d2536e79c05dac2d8cd857f6c55cd0c95ace5b03740824e0e2f1"},
+    {file = "brotli-1.2.0-cp310-cp310-win32.whl", hash = "sha256:b232029d100d393ae3c603c8ffd7e3fe6f798c5e28ddca5feabb8e8fdb732997"},
+    {file = "brotli-1.2.0-cp310-cp310-win_amd64.whl", hash = "sha256:ef87b8ab2704da227e83a246356a2b179ef826f550f794b2c52cddb4efbd0196"},
+    {file = "brotli-1.2.0-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:15b33fe93cedc4caaff8a0bd1eb7e3dab1c61bb22a0bf5bdfdfd97cd7da79744"},
+    {file = "brotli-1.2.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:898be2be399c221d2671d29eed26b6b2713a02c2119168ed914e7d00ceadb56f"},
+    {file = "brotli-1.2.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:350c8348f0e76fff0a0fd6c26755d2653863279d086d3aa2c290a6a7251135dd"},
+    {file = "brotli-1.2.0-cp311-cp311-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:2e1ad3fda65ae0d93fec742a128d72e145c9c7a99ee2fcd667785d99eb25a7fe"},
+    {file = "brotli-1.2.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:40d918bce2b427a0c4ba189df7a006ac0c7277c180aee4617d99e9ccaaf59e6a"},
+    {file = "brotli-1.2.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:2a7f1d03727130fc875448b65b127a9ec5d06d19d0148e7554384229706f9d1b"},
+    {file = "brotli-1.2.0-cp311-cp311-musllinux_1_2_ppc64le.whl", hash = "sha256:9c79f57faa25d97900bfb119480806d783fba83cd09ee0b33c17623935b05fa3"},
+    {file = "brotli-1.2.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:844a8ceb8483fefafc412f85c14f2aae2fb69567bf2a0de53cdb88b73e7c43ae"},
+    {file = "brotli-1.2.0-cp311-cp311-win32.whl", hash = "sha256:aa47441fa3026543513139cb8926a92a8e305ee9c71a6209ef7a97d91640ea03"},
+    {file = "brotli-1.2.0-cp311-cp311-win_amd64.whl", hash = "sha256:022426c9e99fd65d9475dce5c195526f04bb8be8907607e27e747893f6ee3e24"},
+    {file = "brotli-1.2.0-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:35d382625778834a7f3061b15423919aa03e4f5da34ac8e02c074e4b75ab4f84"},
+    {file = "brotli-1.2.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:7a61c06b334bd99bc5ae84f1eeb36bfe01400264b3c352f968c6e30a10f9d08b"},
+    {file = "brotli-1.2.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:acec55bb7c90f1dfc476126f9711a8e81c9af7fb617409a9ee2953115343f08d"},
+    {file = "brotli-1.2.0-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:260d3692396e1895c5034f204f0db022c056f9e2ac841593a4cf9426e2a3faca"},
+    {file = "brotli-1.2.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:072e7624b1fc4d601036ab3f4f27942ef772887e876beff0301d261210bca97f"},
+    {file = "brotli-1.2.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:adedc4a67e15327dfdd04884873c6d5a01d3e3b6f61406f99b1ed4865a2f6d28"},
+    {file = "brotli-1.2.0-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:7a47ce5c2288702e09dc22a44d0ee6152f2c7eda97b3c8482d826a1f3cfc7da7"},
+    {file = "brotli-1.2.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:af43b8711a8264bb4e7d6d9a6d004c3a2019c04c01127a868709ec29962b6036"},
+    {file = "brotli-1.2.0-cp312-cp312-win32.whl", hash = "sha256:e99befa0b48f3cd293dafeacdd0d191804d105d279e0b387a32054c1180f3161"},
+    {file = "brotli-1.2.0-cp312-cp312-win_amd64.whl", hash = "sha256:b35c13ce241abdd44cb8ca70683f20c0c079728a36a996297adb5334adfc1c44"},
+    {file = "brotli-1.2.0-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:9e5825ba2c9998375530504578fd4d5d1059d09621a02065d1b6bfc41a8e05ab"},
+    {file = "brotli-1.2.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:0cf8c3b8ba93d496b2fae778039e2f5ecc7cff99df84df337ca31d8f2252896c"},
+    {file = "brotli-1.2.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c8565e3cdc1808b1a34714b553b262c5de5fbda202285782173ec137fd13709f"},
+    {file = "brotli-1.2.0-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:26e8d3ecb0ee458a9804f47f21b74845cc823fd1bb19f02272be70774f56e2a6"},
+    {file = "brotli-1.2.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:67a91c5187e1eec76a61625c77a6c8c785650f5b576ca732bd33ef58b0dff49c"},
+    {file = "brotli-1.2.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:4ecdb3b6dc36e6d6e14d3a1bdc6c1057c8cbf80db04031d566eb6080ce283a48"},
+    {file = "brotli-1.2.0-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:3e1b35d56856f3ed326b140d3c6d9db91740f22e14b06e840fe4bb1923439a18"},
+    {file = "brotli-1.2.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:54a50a9dad16b32136b2241ddea9e4df159b41247b2ce6aac0b3276a66a8f1e5"},
+    {file = "brotli-1.2.0-cp313-cp313-win32.whl", hash = "sha256:1b1d6a4efedd53671c793be6dd760fcf2107da3a52331ad9ea429edf0902f27a"},
+    {file = "brotli-1.2.0-cp313-cp313-win_amd64.whl", hash = "sha256:b63daa43d82f0cdabf98dee215b375b4058cce72871fd07934f179885aad16e8"},
+    {file = "brotli-1.2.0-cp314-cp314-macosx_10_15_universal2.whl", hash = "sha256:6c12dad5cd04530323e723787ff762bac749a7b256a5bece32b2243dd5c27b21"},
+    {file = "brotli-1.2.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:3219bd9e69868e57183316ee19c84e03e8f8b5a1d1f2667e1aa8c2f91cb061ac"},
+    {file = "brotli-1.2.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:963a08f3bebd8b75ac57661045402da15991468a621f014be54e50f53a58d19e"},
+    {file = "brotli-1.2.0-cp314-cp314-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:9322b9f8656782414b37e6af884146869d46ab85158201d82bab9abbcb971dc7"},
+    {file = "brotli-1.2.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:cf9cba6f5b78a2071ec6fb1e7bd39acf35071d90a81231d67e92d637776a6a63"},
+    {file = "brotli-1.2.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7547369c4392b47d30a3467fe8c3330b4f2e0f7730e45e3103d7d636678a808b"},
+    {file = "brotli-1.2.0-cp314-cp314-musllinux_1_2_ppc64le.whl", hash = "sha256:fc1530af5c3c275b8524f2e24841cbe2599d74462455e9bae5109e9ff42e9361"},
+    {file = "brotli-1.2.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:d2d085ded05278d1c7f65560aae97b3160aeb2ea2c0b3e26204856beccb60888"},
+    {file = "brotli-1.2.0-cp314-cp314-win32.whl", hash = "sha256:832c115a020e463c2f67664560449a7bea26b0c1fdd690352addad6d0a08714d"},
+    {file = "brotli-1.2.0-cp314-cp314-win_amd64.whl", hash = "sha256:e7c0af964e0b4e3412a0ebf341ea26ec767fa0b4cf81abb5e897c9338b5ad6a3"},
+    {file = "brotli-1.2.0-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:82676c2781ecf0ab23833796062786db04648b7aae8be139f6b8065e5e7b1518"},
+    {file = "brotli-1.2.0-cp36-cp36m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:c16ab1ef7bb55651f5836e8e62db1f711d55b82ea08c3b8083ff037157171a69"},
+    {file = "brotli-1.2.0-cp36-cp36m-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:e85190da223337a6b7431d92c799fca3e2982abd44e7b8dec69938dcc81c8e9e"},
+    {file = "brotli-1.2.0-cp36-cp36m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_12_i686.manylinux2010_i686.whl", hash = "sha256:d8c05b1dfb61af28ef37624385b0029df902ca896a639881f594060b30ffc9a7"},
+    {file = "brotli-1.2.0-cp36-cp36m-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:465a0d012b3d3e4f1d6146ea019b5c11e3e87f03d1676da1cc3833462e672fb0"},
+    {file = "brotli-1.2.0-cp36-cp36m-musllinux_1_2_aarch64.whl", hash = "sha256:96fbe82a58cdb2f872fa5d87dedc8477a12993626c446de794ea025bbda625ea"},
+    {file = "brotli-1.2.0-cp36-cp36m-musllinux_1_2_i686.whl", hash = "sha256:1b71754d5b6eda54d16fbbed7fce2d8bc6c052a1b91a35c320247946ee103502"},
+    {file = "brotli-1.2.0-cp36-cp36m-musllinux_1_2_ppc64le.whl", hash = "sha256:66c02c187ad250513c2f4fce973ef402d22f80e0adce734ee4e4efd657b6cb64"},
+    {file = "brotli-1.2.0-cp36-cp36m-musllinux_1_2_x86_64.whl", hash = "sha256:ba76177fd318ab7b3b9bf6522be5e84c2ae798754b6cc028665490f6e66b5533"},
+    {file = "brotli-1.2.0-cp36-cp36m-win32.whl", hash = "sha256:c1702888c9f3383cc2f09eb3e88b8babf5965a54afb79649458ec7c3c7a63e96"},
+    {file = "brotli-1.2.0-cp36-cp36m-win_amd64.whl", hash = "sha256:f8d635cafbbb0c61327f942df2e3f474dde1cff16c3cd0580564774eaba1ee13"},
+    {file = "brotli-1.2.0-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:e80a28f2b150774844c8b454dd288be90d76ba6109670fe33d7ff54d96eb5cb8"},
+    {file = "brotli-1.2.0-cp37-cp37m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:50b1b799f45da91292ffaa21a473ab3a3054fa78560e8ff67082a185274431c8"},
+    {file = "brotli-1.2.0-cp37-cp37m-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:29b7e6716ee4ea0c59e3b241f682204105f7da084d6254ec61886508efeb43bc"},
+    {file = "brotli-1.2.0-cp37-cp37m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_12_i686.manylinux2010_i686.whl", hash = "sha256:640fe199048f24c474ec6f3eae67c48d286de12911110437a36a87d7c89573a6"},
+    {file = "brotli-1.2.0-cp37-cp37m-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:92edab1e2fd6cd5ca605f57d4545b6599ced5dea0fd90b2bcdf8b247a12bd190"},
+    {file = "brotli-1.2.0-cp37-cp37m-musllinux_1_2_aarch64.whl", hash = "sha256:7274942e69b17f9cef76691bcf38f2b2d4c8a5f5dba6ec10958363dcb3308a0a"},
+    {file = "brotli-1.2.0-cp37-cp37m-musllinux_1_2_i686.whl", hash = "sha256:a56ef534b66a749759ebd091c19c03ef81eb8cd96f0d1d16b59127eaf1b97a12"},
+    {file = "brotli-1.2.0-cp37-cp37m-musllinux_1_2_ppc64le.whl", hash = "sha256:5732eff8973dd995549a18ecbd8acd692ac611c5c0bb3f59fa3541ae27b33be3"},
+    {file = "brotli-1.2.0-cp37-cp37m-musllinux_1_2_x86_64.whl", hash = "sha256:598e88c736f63a0efec8363f9eb34e5b5536b7b6b1821e401afcb501d881f59a"},
+    {file = "brotli-1.2.0-cp37-cp37m-win32.whl", hash = "sha256:7ad8cec81f34edf44a1c6a7edf28e7b7806dfb8886e371d95dcf789ccd4e4982"},
+    {file = "brotli-1.2.0-cp37-cp37m-win_amd64.whl", hash = "sha256:865cedc7c7c303df5fad14a57bc5db1d4f4f9b2b4d0a7523ddd206f00c121a16"},
+    {file = "brotli-1.2.0-cp38-cp38-macosx_10_9_universal2.whl", hash = "sha256:ac27a70bda257ae3f380ec8310b0a06680236bea547756c277b5dfe55a2452a8"},
+    {file = "brotli-1.2.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:e813da3d2d865e9793ef681d3a6b66fa4b7c19244a45b817d0cceda67e615990"},
+    {file = "brotli-1.2.0-cp38-cp38-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9fe11467c42c133f38d42289d0861b6b4f9da31e8087ca2c0d7ebb4543625526"},
+    {file = "brotli-1.2.0-cp38-cp38-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:c0d6770111d1879881432f81c369de5cde6e9467be7c682a983747ec800544e2"},
+    {file = "brotli-1.2.0-cp38-cp38-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:eda5a6d042c698e28bda2507a89b16555b9aa954ef1d750e1c20473481aff675"},
+    {file = "brotli-1.2.0-cp38-cp38-musllinux_1_2_aarch64.whl", hash = "sha256:3173e1e57cebb6d1de186e46b5680afbd82fd4301d7b2465beebe83ed317066d"},
+    {file = "brotli-1.2.0-cp38-cp38-musllinux_1_2_ppc64le.whl", hash = "sha256:71a66c1c9be66595d628467401d5976158c97888c2c9379c034e1e2312c5b4f5"},
+    {file = "brotli-1.2.0-cp38-cp38-musllinux_1_2_x86_64.whl", hash = "sha256:1e68cdf321ad05797ee41d1d09169e09d40fdf51a725bb148bff892ce04583d7"},
+    {file = "brotli-1.2.0-cp38-cp38-win32.whl", hash = "sha256:f16dace5e4d3596eaeb8af334b4d2c820d34b8278da633ce4a00020b2eac981c"},
+    {file = "brotli-1.2.0-cp38-cp38-win_amd64.whl", hash = "sha256:14ef29fc5f310d34fc7696426071067462c9292ed98b5ff5a27ac70a200e5470"},
+    {file = "brotli-1.2.0-cp39-cp39-macosx_10_9_universal2.whl", hash = "sha256:8d4f47f284bdd28629481c97b5f29ad67544fa258d9091a6ed1fda47c7347cd1"},
+    {file = "brotli-1.2.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:2881416badd2a88a7a14d981c103a52a23a276a553a8aacc1346c2ff47c8dc17"},
+    {file = "brotli-1.2.0-cp39-cp39-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:2d39b54b968f4b49b5e845758e202b1035f948b0561ff5e6385e855c96625971"},
+    {file = "brotli-1.2.0-cp39-cp39-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:95db242754c21a88a79e01504912e537808504465974ebb92931cfca2510469e"},
+    {file = "brotli-1.2.0-cp39-cp39-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:bba6e7e6cfe1e6cb6eb0b7c2736a6059461de1fa2c0ad26cf845de6c078d16c8"},
+    {file = "brotli-1.2.0-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:88ef7d55b7bcf3331572634c3fd0ed327d237ceb9be6066810d39020a3ebac7a"},
+    {file = "brotli-1.2.0-cp39-cp39-musllinux_1_2_ppc64le.whl", hash = "sha256:7fa18d65a213abcfbb2f6cafbb4c58863a8bd6f2103d65203c520ac117d1944b"},
+    {file = "brotli-1.2.0-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:09ac247501d1909e9ee47d309be760c89c990defbb2e0240845c892ea5ff0de4"},
+    {file = "brotli-1.2.0-cp39-cp39-win32.whl", hash = "sha256:c25332657dee6052ca470626f18349fc1fe8855a56218e19bd7a8c6ad4952c49"},
+    {file = "brotli-1.2.0-cp39-cp39-win_amd64.whl", hash = "sha256:1ce223652fd4ed3eb2b7f78fbea31c52314baecfac68db44037bb4167062a937"},
+    {file = "brotli-1.2.0.tar.gz", hash = "sha256:e310f77e41941c13340a95976fe66a8a95b01e783d430eeaf7a2f87e0a57dd0a"},
+]
+
 [[package]]
 name = "cachecontrol"
 version = "0.14.3"
@@ -749,6 +860,23 @@ json-validation = ["jsonschema[format] (>=4.18,<5.0)"]
 validation = ["jsonschema[format] (>=4.18,<5.0)", "lxml (>=4,<6)"]
 xml-validation = ["lxml (>=4,<6)"]
 
+[[package]]
+name = "dataclasses-json"
+version = "0.6.7"
+description = "Easily serialize dataclasses to and from JSON."
+optional = true
+python-versions = "<4.0,>=3.7"
+groups = ["main"]
+markers = "extra == \"dev\""
+files = [
+    {file = "dataclasses_json-0.6.7-py3-none-any.whl", hash = "sha256:0dbf33f26c8d5305befd61b39d2b3414e8a407bedc2834dea9b8d642666fb40a"},
+    {file = "dataclasses_json-0.6.7.tar.gz", hash = "sha256:b6b3e528266ea45b9535223bc53ca645f5208833c29229e847b3f26a1cc55fc0"},
+]
+
+[package.dependencies]
+marshmallow = ">=3.18.0,<4.0.0"
+typing-inspect = ">=0.4.0,<1"
+
 [[package]]
 name = "defusedxml"
 version = "0.7.1"
@@ -1580,6 +1708,26 @@ files = [
     {file = "jiter-0.10.0.tar.gz", hash = "sha256:07a7142c38aacc85194391108dc91b5b57093c978a9932bd86a36862759d9500"},
 ]
 
+[[package]]
+name = "jmapc"
+version = "0.2.23"
+description = "JMAP client library for Python"
+optional = true
+python-versions = "<4.0,>=3.9"
+groups = ["main"]
+markers = "extra == \"dev\""
+files = [
+    {file = "jmapc-0.2.23-py3-none-any.whl", hash = "sha256:7bc429764b7edc4b377d3c05c2e9aeabaf725c4ad3a529463aa962473e6dfd7d"},
+    {file = "jmapc-0.2.23.tar.gz", hash = "sha256:c1e8a54f922dca1dbb9aca2fb541fd0c3acc7a582c773aa3a440177f48d67542"},
+]
+
+[package.dependencies]
+brotli = ">=1.0.9"
+dataclasses-json = "*"
+python-dateutil = "*"
+requests = "*"
+sseclient = "*"
+
 [[package]]
 name = "jmespath"
 version = "1.0.1"
@@ -1773,6 +1921,27 @@ profiling = ["gprof2dot"]
 rtd = ["ipykernel", "jupyter_sphinx", "mdit-py-plugins (>=0.5.0)", "myst-parser", "pyyaml", "sphinx", "sphinx-book-theme (>=1.0,<2.0)", "sphinx-copybutton", "sphinx-design"]
 testing = ["coverage", "pytest", "pytest-cov", "pytest-regressions", "requests"]
 
+[[package]]
+name = "marshmallow"
+version = "3.26.2"
+description = "A lightweight library for converting complex datatypes to and from native Python datatypes."
+optional = true
+python-versions = ">=3.9"
+groups = ["main"]
+markers = "extra == \"dev\""
+files = [
+    {file = "marshmallow-3.26.2-py3-none-any.whl", hash = "sha256:013fa8a3c4c276c24d26d84ce934dc964e2aa794345a0f8c7e5a7191482c8a73"},
+    {file = "marshmallow-3.26.2.tar.gz", hash = "sha256:bbe2adb5a03e6e3571b573f42527c6fe926e17467833660bebd11593ab8dfd57"},
+]
+
+[package.dependencies]
+packaging = ">=17.0"
+
+[package.extras]
+dev = ["marshmallow[tests]", "pre-commit (>=3.5,<5.0)", "tox"]
+docs = ["autodocsumm (==0.2.14)", "furo (==2024.8.6)", "sphinx (==8.1.3)", "sphinx-copybutton (==0.5.2)", "sphinx-issues (==5.0.0)", "sphinxext-opengraph (==0.9.1)"]
+tests = ["pytest", "simplejson"]
+
 [[package]]
 name = "mccabe"
 version = "0.7.0"
@@ -1887,6 +2056,19 @@ files = [
     {file = "msgpack-1.1.1.tar.gz", hash = "sha256:77b79ce34a2bdab2594f490c8e80dd62a02d650b91a75159a63ec413b8d104cd"},
 ]
 
+[[package]]
+name = "mypy-extensions"
+version = "1.1.0"
+description = "Type system extensions for programs checked with the mypy type checker."
+optional = true
+python-versions = ">=3.8"
+groups = ["main"]
+markers = "extra == \"dev\""
+files = [
+    {file = "mypy_extensions-1.1.0-py3-none-any.whl", hash = "sha256:1be4cccdb0f2482337c4743e60421de3a356cd97508abadd57d47403e94f5505"},
+    {file = "mypy_extensions-1.1.0.tar.gz", hash = "sha256:52e68efc3284861e772bbcd66823fde5ae21fd2fdb51c62a211403730b916558"},
+]
+
 [[package]]
 name = "nested-multipart-parser"
 version = "1.5.0"
@@ -3465,6 +3647,22 @@ files = [
 dev = ["build", "hatch"]
 doc = ["sphinx"]
 
+[[package]]
+name = "sseclient"
+version = "0.0.27"
+description = "Python client library for reading Server Sent Event streams."
+optional = true
+python-versions = "*"
+groups = ["main"]
+markers = "extra == \"dev\""
+files = [
+    {file = "sseclient-0.0.27.tar.gz", hash = "sha256:b2fe534dcb33b1d3faad13d60c5a7c718e28f85987f2a034ecf5ec279918c11c"},
+]
+
+[package.dependencies]
+requests = ">=2.9"
+six = "*"
+
 [[package]]
 name = "tld"
 version = "0.13.1"
@@ -3560,6 +3758,23 @@ files = [
     {file = "typing_extensions-4.15.0.tar.gz", hash = "sha256:0cea48d173cc12fa28ecabc3b837ea3cf6f38c6d1136f85cbaaf598984861466"},
 ]
 
+[[package]]
+name = "typing-inspect"
+version = "0.9.0"
+description = "Runtime inspection utilities for typing module."
+optional = true
+python-versions = "*"
+groups = ["main"]
+markers = "extra == \"dev\""
+files = [
+    {file = "typing_inspect-0.9.0-py3-none-any.whl", hash = "sha256:9ee6fc59062311ef8547596ab6b955e1b8aa46242d854bfc78f4f6b0eff35f9f"},
+    {file = "typing_inspect-0.9.0.tar.gz", hash = "sha256:b23fc42ff6f6ef6954e4852c1fb512cdd18dbea03134f91f856a95ccc9461f78"},
+]
+
+[package.dependencies]
+mypy-extensions = ">=0.3.0"
+typing-extensions = ">=3.7.4"
+
 [[package]]
 name = "typing-inspection"
 version = "0.4.1"
@@ -3691,9 +3906,9 @@ files = [
 brotli = ["brotli"]
 
 [extras]
-dev = ["django-extensions", "drf-spectacular-sidecar", "flower", "pip-audit", "pipdeptree", "pylint", "pylint-django", "pytest", "pytest-cov", "pytest-django", "pytest-icdiff", "pytest-repeat", "pytest-xdist", "responses", "ruff"]
+dev = ["django-extensions", "drf-spectacular-sidecar", "flower", "jmapc", "pip-audit", "pipdeptree", "pylint", "pylint-django", "pytest", "pytest-cov", "pytest-django", "pytest-icdiff", "pytest-repeat", "pytest-xdist", "responses", "ruff"]
 
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.13,<4.0"
-content-hash = "63c5b110585e0b4d0f0d109ca27c64ec4b7f884dc5132f37c7d38d7836d53f15"
+content-hash = "52d9e0d446b5809462dfba130086e5c41f86f2ce7324f5a2526793e4643dbb76"
diff --git a/src/backend/pyproject.toml b/src/backend/pyproject.toml
index 9085d5573..fb3169260 100644
--- a/src/backend/pyproject.toml
+++ b/src/backend/pyproject.toml
@@ -79,6 +79,7 @@ dev = [
     "django-extensions==3.2.3",
     "drf-spectacular-sidecar==2024.12.1",
     "flower==2.0.1",
+    "jmapc==0.2.23",
     "pip-audit==2.9.0",
     "pipdeptree==2.28.0",
     "pylint-django==2.6.1",