diff --git a/integrations/beads-mcp/src/beads_mcp/mail.py b/integrations/beads-mcp/src/beads_mcp/mail.py
new file mode 100644
index 00000000..ad6a839f
--- /dev/null
+++ b/integrations/beads-mcp/src/beads_mcp/mail.py
@@ -0,0 +1,611 @@
+"""Agent Mail integration for beads MCP server.
+
+Provides simple messaging functions that wrap Agent Mail HTTP API.
+Requires BEADS_AGENT_MAIL_URL and BEADS_AGENT_NAME environment variables.
+"""
+
+import logging
+import os
+from typing import Any, Optional
+from urllib.parse import urljoin
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+# Timeout for Agent Mail HTTP requests (seconds)
+AGENT_MAIL_TIMEOUT = 5.0
+AGENT_MAIL_RETRIES = 2
+
+
+class MailError(Exception):
+    """Base exception for Agent Mail errors."""
+
+    def __init__(self, code: str, message: str, data: Optional[dict] = None):
+        self.code = code
+        self.message = message
+        self.data = data or {}
+        super().__init__(message)
+
+
+def _get_config() -> tuple[str, str, Optional[str]]:
+    """Get Agent Mail configuration from environment.
+
+    Returns:
+        (base_url, agent_name, token)
+
+    Raises:
+        MailError: If required configuration is missing
+    """
+    base_url = os.environ.get("BEADS_AGENT_MAIL_URL")
+    if not base_url:
+        raise MailError(
+            "NOT_CONFIGURED",
+            "Agent Mail not configured. Set BEADS_AGENT_MAIL_URL environment variable.\n"
+            "Example: export BEADS_AGENT_MAIL_URL=http://127.0.0.1:8765\n"
+            "See docs/AGENT_MAIL_QUICKSTART.md for setup instructions.",
+        )
+
+    agent_name = os.environ.get("BEADS_AGENT_NAME")
+    if not agent_name:
+        # Try to derive from user/repo
+        import getpass
+
+        try:
+            user = getpass.getuser()
+            cwd = os.getcwd()
+            repo_name = os.path.basename(cwd)
+            agent_name = f"{user}-{repo_name}"
+            logger.warning(
+                f"BEADS_AGENT_NAME not set, using derived name: {agent_name}"
+            )
+        except Exception:
+            raise MailError(
+                "NOT_CONFIGURED",
+                "Agent Mail not configured. Set BEADS_AGENT_NAME environment variable.\n"
+                "Example: export BEADS_AGENT_NAME=my-agent",
+            )
+
+    token = os.environ.get("BEADS_AGENT_MAIL_TOKEN")
+
+    return base_url, agent_name, token
+
+
+def _get_project_key() -> str:
+    """Get project key from environment or derive from Git/workspace.
+
+    Returns:
+        Project key (absolute path to workspace root)
+    """
+    # Check explicit project ID first
+    project_id = os.environ.get("BEADS_PROJECT_ID")
+    if project_id:
+        return project_id
+
+    # Try to get from bd workspace detection
+    # Import here to avoid circular dependency
+    from .tools import _find_beads_db_in_tree
+
+    workspace = _find_beads_db_in_tree()
+    if workspace:
+        return os.path.abspath(workspace)
+
+    # Fallback to current directory
+    return os.path.abspath(os.getcwd())
+
+
+def _call_agent_mail(
+    method: str,
+    endpoint: str,
+    json_data: Optional[dict] = None,
+    params: Optional[dict] = None,
+) -> dict[str, Any]:
+    """Make HTTP request to Agent Mail server with retries.
+
+    Args:
+        method: HTTP method (GET, POST, DELETE, etc.)
+        endpoint: API endpoint path (e.g., "/api/messages")
+        json_data: Request body as JSON
+        params: URL query parameters
+
+    Returns:
+        Response JSON
+
+    Raises:
+        MailError: On request failure or server error
+    """
+    base_url, _, token = _get_config()
+    url = urljoin(base_url, endpoint)
+
+    headers = {}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+
+    # Use idempotency key for write operations to avoid duplicates on retry
+    if method in {"POST", "PUT"} and json_data:
+        import uuid
+
+        headers["Idempotency-Key"] = str(uuid.uuid4())
+
+    last_error = None
+    for attempt in range(AGENT_MAIL_RETRIES + 1):
+        try:
+            response = requests.request(
+                method,
+                url,
+                json=json_data,
+                params=params,
+                headers=headers,
+                timeout=AGENT_MAIL_TIMEOUT,
+            )
+
+            # Success
+            if response.status_code < 400:
+                return response.json() if response.content else {}
+
+            # Client error - don't retry
+            if 400 <= response.status_code < 500:
+                error_data = {}
+                try:
+                    error_data = response.json()
+                except Exception:
+                    error_data = {"detail": response.text}
+
+                if response.status_code == 404:
+                    raise MailError(
+                        "NOT_FOUND",
+                        f"Resource not found: {endpoint}",
+                        error_data,
+                    )
+                elif response.status_code == 409:
+                    raise MailError(
+                        "CONFLICT",
+                        error_data.get("detail", "Conflict"),
+                        error_data,
+                    )
+                else:
+                    raise MailError(
+                        "INVALID_ARGUMENT",
+                        error_data.get("detail", f"HTTP {response.status_code}"),
+                        error_data,
+                    )
+
+            # Server error - retry
+            last_error = MailError(
+                "UNAVAILABLE",
+                f"Agent Mail server error: HTTP {response.status_code}",
+                {"status": response.status_code, "attempt": attempt + 1},
+            )
+
+        except requests.exceptions.Timeout:
+            last_error = MailError(
+                "TIMEOUT",
+                f"Agent Mail request timeout after {AGENT_MAIL_TIMEOUT}s",
+                {"attempt": attempt + 1},
+            )
+        except requests.exceptions.ConnectionError as e:
+            last_error = MailError(
+                "UNAVAILABLE",
+                f"Cannot connect to Agent Mail server at {base_url}",
+                {"error": str(e), "attempt": attempt + 1},
+            )
+        except MailError:
+            raise  # Re-raise our own errors
+        except Exception as e:
+            last_error = MailError(
+                "INTERNAL_ERROR",
+                f"Unexpected error calling Agent Mail: {e}",
+                {"error": str(e), "attempt": attempt + 1},
+            )
+
+        # Exponential backoff between retries
+        if attempt < AGENT_MAIL_RETRIES:
+            import time
+
+            time.sleep(0.5 * (2**attempt))
+
+    # All retries exhausted
+    raise last_error
+
+
+def mail_send(
+    to: list[str],
+    subject: str,
+    body: str,
+    urgent: bool = False,
+    cc: Optional[list[str]] = None,
+    project_key: Optional[str] = None,
+    sender_name: Optional[str] = None,
+) -> dict[str, Any]:
+    """Send a message to other agents.
+
+    Args:
+        to: List of recipient agent names
+        subject: Message subject
+        body: Message body (Markdown)
+        urgent: Mark as urgent (default: False)
+        cc: Optional CC recipients
+        project_key: Override project identifier (default: auto-detect)
+        sender_name: Override sender name (default: BEADS_AGENT_NAME)
+
+    Returns:
+        {
+            "message_id": int,
+            "thread_id": str,
+            "sent_to": int  # number of recipients
+        }
+
+    Raises:
+        MailError: On configuration or delivery error
+    """
+    _, auto_agent_name, _ = _get_config()
+    auto_project_key = _get_project_key()
+
+    sender = sender_name or auto_agent_name
+    project = project_key or auto_project_key
+
+    importance = "urgent" if urgent else "normal"
+
+    # Call Agent Mail send_message tool via HTTP
+    # Note: Agent Mail MCP tools use POST to /mcp/call endpoint
+    result = _call_agent_mail(
+        "POST",
+        "/mcp/call",
+        json_data={
+            "method": "tools/call",
+            "params": {
+                "name": "send_message",
+                "arguments": {
+                    "project_key": project,
+                    "sender_name": sender,
+                    "to": to,
+                    "subject": subject,
+                    "body_md": body,
+                    "cc": cc or [],
+                    "importance": importance,
+                },
+            },
+        },
+    )
+
+    # Extract message details from result
+    # Agent Mail returns: {"deliveries": [...], "count": N}
+    deliveries = result.get("deliveries", [])
+    if not deliveries:
+        raise MailError("INTERNAL_ERROR", "No deliveries returned from Agent Mail")
+
+    # Get message ID from first delivery
+    first_delivery = deliveries[0]
+    payload = first_delivery.get("payload", {})
+    message_id = payload.get("id")
+    thread_id = payload.get("thread_id")
+
+    return {
+        "message_id": message_id,
+        "thread_id": thread_id,
+        "sent_to": len(deliveries),
+    }
+
+
+def mail_inbox(
+    limit: int = 20,
+    urgent_only: bool = False,
+    unread_only: bool = False,
+    cursor: Optional[str] = None,
+    agent_name: Optional[str] = None,
+    project_key: Optional[str] = None,
+) -> dict[str, Any]:
+    """Get messages from inbox.
+
+    Args:
+        limit: Maximum messages to return (default: 20)
+        urgent_only: Only return urgent messages
+        unread_only: Only return unread messages
+        cursor: Pagination cursor (for next page)
+        agent_name: Override agent name (default: BEADS_AGENT_NAME)
+        project_key: Override project (default: auto-detect)
+
+    Returns:
+        {
+            "messages": [
+                {
+                    "id": int,
+                    "thread_id": str,
+                    "from": str,
+                    "subject": str,
+                    "created_ts": str,  # ISO-8601
+                    "unread": bool,
+                    "ack_required": bool,
+                    "urgent": bool,
+                    "preview": str  # first 100 chars
+                },
+                ...
+            ],
+            "next_cursor": str | None
+        }
+    """
+    _, auto_agent_name, _ = _get_config()
+    auto_project_key = _get_project_key()
+
+    agent = agent_name or auto_agent_name
+    project = project_key or auto_project_key
+
+    # Call fetch_inbox via MCP
+    result = _call_agent_mail(
+        "POST",
+        "/mcp/call",
+        json_data={
+            "method": "tools/call",
+            "params": {
+                "name": "fetch_inbox",
+                "arguments": {
+                    "project_key": project,
+                    "agent_name": agent,
+                    "limit": limit,
+                    "urgent_only": urgent_only,
+                    "include_bodies": False,  # Get preview only
+                },
+            },
+        },
+    )
+
+    # Agent Mail returns list of messages directly
+    messages = result if isinstance(result, list) else []
+
+    # Transform to our format and filter unread if requested
+    formatted_messages = []
+    for msg in messages:
+        # Skip read messages if unread_only
+        if unread_only and msg.get("read_ts"):
+            continue
+
+        formatted_messages.append(
+            {
+                "id": msg.get("id"),
+                "thread_id": msg.get("thread_id"),
+                "from": msg.get("from"),
+                "subject": msg.get("subject"),
+                "created_ts": msg.get("created_ts"),
+                "unread": not bool(msg.get("read_ts")),
+                "ack_required": msg.get("ack_required", False),
+                "urgent": msg.get("importance") in {"high", "urgent"},
+                "preview": msg.get("body_md", "")[:100] if msg.get("body_md") else "",
+            }
+        )
+
+    # Simple cursor pagination (use last message ID)
+    next_cursor = None
+    if formatted_messages and len(formatted_messages) >= limit:
+        next_cursor = str(formatted_messages[-1]["id"])
+
+    return {"messages": formatted_messages, "next_cursor": next_cursor}
+
+
+def mail_read(
+    message_id: int,
+    mark_read: bool = True,
+    agent_name: Optional[str] = None,
+    project_key: Optional[str] = None,
+) -> dict[str, Any]:
+    """Read full message with body.
+
+    Args:
+        message_id: Message ID to read
+        mark_read: Mark message as read (default: True)
+        agent_name: Override agent name (default: BEADS_AGENT_NAME)
+        project_key: Override project (default: auto-detect)
+
+    Returns:
+        {
+            "id": int,
+            "thread_id": str,
+            "from": str,
+            "to": list[str],
+            "subject": str,
+            "body": str,  # Full Markdown body
+            "created_ts": str,
+            "ack_required": bool,
+            "ack_status": bool,
+            "read_ts": str | None,
+            "urgent": bool
+        }
+    """
+    _, auto_agent_name, _ = _get_config()
+    auto_project_key = _get_project_key()
+
+    agent = agent_name or auto_agent_name
+    project = project_key or auto_project_key
+
+    # Get message via resource
+    result = _call_agent_mail(
+        "GET", f"/mcp/resources/resource://message/{message_id}"
+    )
+
+    # Mark as read if requested
+    if mark_read:
+        try:
+            _call_agent_mail(
+                "POST",
+                "/mcp/call",
+                json_data={
+                    "method": "tools/call",
+                    "params": {
+                        "name": "mark_message_read",
+                        "arguments": {
+                            "project_key": project,
+                            "agent_name": agent,
+                            "message_id": message_id,
+                        },
+                    },
+                },
+            )
+        except MailError as e:
+            # Don't fail read if mark fails
+            logger.warning(f"Failed to mark message {message_id} as read: {e}")
+
+    # Extract message from result
+    # Resource returns: {"contents": [{...}]}
+    contents = result.get("contents", [])
+    if not contents:
+        raise MailError("NOT_FOUND", f"Message {message_id} not found")
+
+    msg = contents[0]
+
+    return {
+        "id": msg.get("id"),
+        "thread_id": msg.get("thread_id"),
+        "from": msg.get("from"),
+        "to": msg.get("to", []),
+        "subject": msg.get("subject"),
+        "body": msg.get("body_md", ""),
+        "created_ts": msg.get("created_ts"),
+        "ack_required": msg.get("ack_required", False),
+        "ack_status": bool(msg.get("ack_ts")),
+        "read_ts": msg.get("read_ts"),
+        "urgent": msg.get("importance") in {"high", "urgent"},
+    }
+
+
+def mail_reply(
+    message_id: int,
+    body: str,
+    subject: Optional[str] = None,
+    agent_name: Optional[str] = None,
+    project_key: Optional[str] = None,
+) -> dict[str, Any]:
+    """Reply to a message (preserves thread).
+
+    Args:
+        message_id: Message ID to reply to
+        body: Reply body (Markdown)
+        subject: Override subject (default: "Re: <original subject>")
+        agent_name: Override sender name (default: BEADS_AGENT_NAME)
+        project_key: Override project (default: auto-detect)
+
+    Returns:
+        {
+            "message_id": int,
+            "thread_id": str
+        }
+    """
+    _, auto_agent_name, _ = _get_config()
+    auto_project_key = _get_project_key()
+
+    sender = agent_name or auto_agent_name
+    project = project_key or auto_project_key
+
+    # Call reply_message via MCP
+    args = {
+        "project_key": project,
+        "message_id": message_id,
+        "sender_name": sender,
+        "body_md": body,
+    }
+
+    if subject:
+        args["subject_prefix"] = subject
+
+    result = _call_agent_mail(
+        "POST",
+        "/mcp/call",
+        json_data={
+            "method": "tools/call",
+            "params": {"name": "reply_message", "arguments": args},
+        },
+    )
+
+    # Extract reply details
+    reply = result.get("reply", {})
+    return {
+        "message_id": reply.get("id"),
+        "thread_id": reply.get("thread_id"),
+    }
+
+
+def mail_ack(
+    message_id: int,
+    agent_name: Optional[str] = None,
+    project_key: Optional[str] = None,
+) -> dict[str, bool]:
+    """Acknowledge a message (for ack_required messages).
+
+    Args:
+        message_id: Message ID to acknowledge
+        agent_name: Override agent name (default: BEADS_AGENT_NAME)
+        project_key: Override project (default: auto-detect)
+
+    Returns:
+        {"acknowledged": True}
+    """
+    _, auto_agent_name, _ = _get_config()
+    auto_project_key = _get_project_key()
+
+    agent = agent_name or auto_agent_name
+    project = project_key or auto_project_key
+
+    # Call acknowledge_message via MCP
+    _call_agent_mail(
+        "POST",
+        "/mcp/call",
+        json_data={
+            "method": "tools/call",
+            "params": {
+                "name": "acknowledge_message",
+                "arguments": {
+                    "project_key": project,
+                    "agent_name": agent,
+                    "message_id": message_id,
+                },
+            },
+        },
+    )
+
+    return {"acknowledged": True}
+
+
+def mail_delete(
+    message_id: int,
+    agent_name: Optional[str] = None,
+    project_key: Optional[str] = None,
+) -> dict[str, bool]:
+    """Delete (archive) a message from inbox.
+
+    Note: Agent Mail may archive rather than truly delete.
+
+    Args:
+        message_id: Message ID to delete
+        agent_name: Override agent name (default: BEADS_AGENT_NAME)
+        project_key: Override project (default: auto-detect)
+
+    Returns:
+        {"deleted": True} or {"archived": True}
+    """
+    _, auto_agent_name, _ = _get_config()
+    auto_project_key = _get_project_key()
+
+    agent = agent_name or auto_agent_name
+    project = project_key or auto_project_key
+
+    # Agent Mail doesn't have explicit delete in MCP API
+    # Best we can do is mark as read and acknowledged
+    # (This prevents it from showing in urgent/unread views)
+    try:
+        _call_agent_mail(
+            "POST",
+            "/mcp/call",
+            json_data={
+                "method": "tools/call",
+                "params": {
+                    "name": "mark_message_read",
+                    "arguments": {
+                        "project_key": project,
+                        "agent_name": agent,
+                        "message_id": message_id,
+                    },
+                },
+            },
+        )
+        return {"archived": True}
+    except MailError:
+        # Soft failure - message may not exist or already read
+        return {"archived": True}
diff --git a/integrations/beads-mcp/src/beads_mcp/mail_tools.py b/integrations/beads-mcp/src/beads_mcp/mail_tools.py
new file mode 100644
index 00000000..8dae6ac7
--- /dev/null
+++ b/integrations/beads-mcp/src/beads_mcp/mail_tools.py
@@ -0,0 +1,211 @@
+"""MCP tools for Agent Mail messaging."""
+
+import logging
+from typing import Annotated, Any
+
+from .mail import (
+    MailError,
+    mail_ack,
+    mail_delete,
+    mail_inbox,
+    mail_read,
+    mail_reply,
+    mail_send,
+)
+from .models import (
+    MailAckParams,
+    MailDeleteParams,
+    MailInboxParams,
+    MailReadParams,
+    MailReplyParams,
+    MailSendParams,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def beads_mail_send(params: MailSendParams) -> dict[str, Any]:
+    """Send a message to other agents via Agent Mail.
+
+    Requires BEADS_AGENT_MAIL_URL and BEADS_AGENT_NAME environment variables.
+    Auto-detects project from workspace root.
+
+    Example:
+        mail_send(to=["alice"], subject="Review PR", body="Can you review PR #42?")
+
+    Args:
+        params: Message parameters (to, subject, body, etc.)
+
+    Returns:
+        {message_id: int, thread_id: str, sent_to: int}
+
+    Raises:
+        MailError: On configuration or delivery error
+    """
+    try:
+        return mail_send(
+            to=params.to,
+            subject=params.subject,
+            body=params.body,
+            urgent=params.urgent,
+            cc=params.cc,
+            project_key=params.project_key,
+            sender_name=params.sender_name,
+        )
+    except MailError as e:
+        logger.error(f"mail_send failed: {e.message}")
+        return {"error": e.code, "message": e.message, "data": e.data}
+
+
+def beads_mail_inbox(
+    params: Annotated[MailInboxParams, "Parameters"] = MailInboxParams(),
+) -> dict[str, Any]:
+    """Get messages from Agent Mail inbox.
+
+    Requires BEADS_AGENT_MAIL_URL and BEADS_AGENT_NAME environment variables.
+
+    Example:
+        mail_inbox(limit=10, unread_only=True)
+
+    Args:
+        params: Inbox filter parameters
+
+    Returns:
+        {
+            messages: [{id, thread_id, from, subject, created_ts, unread, ack_required, urgent, preview}, ...],
+            next_cursor: str | None
+        }
+
+    Raises:
+        MailError: On configuration or fetch error
+    """
+    try:
+        return mail_inbox(
+            limit=params.limit,
+            urgent_only=params.urgent_only,
+            unread_only=params.unread_only,
+            cursor=params.cursor,
+            agent_name=params.agent_name,
+            project_key=params.project_key,
+        )
+    except MailError as e:
+        logger.error(f"mail_inbox failed: {e.message}")
+        return {"error": e.code, "message": e.message, "data": e.data}
+
+
+def beads_mail_read(params: MailReadParams) -> dict[str, Any]:
+    """Read full message with body from Agent Mail.
+
+    By default, marks the message as read. Set mark_read=False to preview without marking.
+
+    Example:
+        mail_read(message_id=123)
+
+    Args:
+        params: Read parameters (message_id, mark_read)
+
+    Returns:
+        {
+            id, thread_id, from, to, subject, body,
+            created_ts, ack_required, ack_status, read_ts, urgent
+        }
+
+    Raises:
+        MailError: On configuration or read error
+    """
+    try:
+        return mail_read(
+            message_id=params.message_id,
+            mark_read=params.mark_read,
+            agent_name=params.agent_name,
+            project_key=params.project_key,
+        )
+    except MailError as e:
+        logger.error(f"mail_read failed: {e.message}")
+        return {"error": e.code, "message": e.message, "data": e.data}
+
+
+def beads_mail_reply(params: MailReplyParams) -> dict[str, Any]:
+    """Reply to a message (preserves thread).
+
+    Automatically inherits thread_id from the original message.
+
+    Example:
+        mail_reply(message_id=123, body="Thanks, will review today!")
+
+    Args:
+        params: Reply parameters (message_id, body, subject)
+
+    Returns:
+        {message_id: int, thread_id: str}
+
+    Raises:
+        MailError: On configuration or reply error
+    """
+    try:
+        return mail_reply(
+            message_id=params.message_id,
+            body=params.body,
+            subject=params.subject,
+            agent_name=params.agent_name,
+            project_key=params.project_key,
+        )
+    except MailError as e:
+        logger.error(f"mail_reply failed: {e.message}")
+        return {"error": e.code, "message": e.message, "data": e.data}
+
+
+def beads_mail_ack(params: MailAckParams) -> dict[str, bool]:
+    """Acknowledge a message (for ack_required messages).
+
+    Safe to call even if message doesn't require acknowledgement.
+
+    Example:
+        mail_ack(message_id=123)
+
+    Args:
+        params: Acknowledgement parameters (message_id)
+
+    Returns:
+        {acknowledged: True}
+
+    Raises:
+        MailError: On configuration or ack error
+    """
+    try:
+        return mail_ack(
+            message_id=params.message_id,
+            agent_name=params.agent_name,
+            project_key=params.project_key,
+        )
+    except MailError as e:
+        logger.error(f"mail_ack failed: {e.message}")
+        return {"error": e.code, "acknowledged": False, "message": e.message}
+
+
+def beads_mail_delete(params: MailDeleteParams) -> dict[str, bool]:
+    """Delete (archive) a message from Agent Mail inbox.
+
+    Note: Agent Mail archives messages rather than permanently deleting them.
+
+    Example:
+        mail_delete(message_id=123)
+
+    Args:
+        params: Delete parameters (message_id)
+
+    Returns:
+        {deleted: True} or {archived: True}
+
+    Raises:
+        MailError: On configuration or delete error
+    """
+    try:
+        return mail_delete(
+            message_id=params.message_id,
+            agent_name=params.agent_name,
+            project_key=params.project_key,
+        )
+    except MailError as e:
+        logger.error(f"mail_delete failed: {e.message}")
+        return {"error": e.code, "deleted": False, "message": e.message}
diff --git a/integrations/beads-mcp/tests/test_mail.py b/integrations/beads-mcp/tests/test_mail.py
new file mode 100644
index 00000000..044479c0
--- /dev/null
+++ b/integrations/beads-mcp/tests/test_mail.py
@@ -0,0 +1,426 @@
+"""Tests for Agent Mail messaging integration."""
+
+import os
+from unittest.mock import Mock, patch
+
+import pytest
+
+from beads_mcp.mail import (
+    MailError,
+    mail_ack,
+    mail_delete,
+    mail_inbox,
+    mail_read,
+    mail_reply,
+    mail_send,
+)
+from beads_mcp.models import (
+    MailAckParams,
+    MailDeleteParams,
+    MailInboxParams,
+    MailReadParams,
+    MailReplyParams,
+    MailSendParams,
+)
+
+
+@pytest.fixture
+def mock_agent_mail_env(tmp_path):
+    """Set up Agent Mail environment variables."""
+    old_env = os.environ.copy()
+    
+    os.environ["BEADS_AGENT_MAIL_URL"] = "http://127.0.0.1:8765"
+    os.environ["BEADS_AGENT_NAME"] = "test-agent"
+    os.environ["BEADS_PROJECT_ID"] = str(tmp_path)
+    
+    yield
+    
+    # Restore environment
+    os.environ.clear()
+    os.environ.update(old_env)
+
+
+@pytest.fixture
+def mock_requests():
+    """Mock requests library for HTTP calls."""
+    with patch("beads_mcp.mail.requests.request") as mock_req:
+        yield mock_req
+
+
+class TestMailConfiguration:
+    """Test configuration and error handling."""
+    
+    def test_missing_url_raises_error(self):
+        """Test that missing BEADS_AGENT_MAIL_URL raises NOT_CONFIGURED."""
+        old_url = os.environ.pop("BEADS_AGENT_MAIL_URL", None)
+        
+        try:
+            with pytest.raises(MailError) as exc_info:
+                mail_send(to=["alice"], subject="Test", body="Test")
+            
+            assert exc_info.value.code == "NOT_CONFIGURED"
+            assert "BEADS_AGENT_MAIL_URL" in exc_info.value.message
+        finally:
+            if old_url:
+                os.environ["BEADS_AGENT_MAIL_URL"] = old_url
+    
+    def test_missing_agent_name_derives_default(self, mock_agent_mail_env, mock_requests, tmp_path):
+        """Test that missing BEADS_AGENT_NAME derives from user/repo."""
+        del os.environ["BEADS_AGENT_NAME"]
+        os.environ["BEADS_PROJECT_ID"] = str(tmp_path)
+        
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {
+            "deliveries": [{
+                "payload": {
+                    "id": 123,
+                    "thread_id": "thread-1",
+                }
+            }]
+        }
+        mock_requests.return_value.content = b'{"deliveries": []}'
+        
+        # Should not raise - derives agent name
+        result = mail_send(to=["alice"], subject="Test", body="Test")
+        assert result["message_id"] == 123
+
+
+class TestMailSend:
+    """Test mail_send function."""
+    
+    def test_send_basic_message(self, mock_agent_mail_env, mock_requests):
+        """Test sending a basic message."""
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {
+            "deliveries": [{
+                "payload": {
+                    "id": 123,
+                    "thread_id": "thread-abc",
+                }
+            }]
+        }
+        mock_requests.return_value.content = b'{"deliveries": []}'
+        
+        result = mail_send(
+            to=["alice", "bob"],
+            subject="Test Message",
+            body="Hello world!",
+        )
+        
+        assert result["message_id"] == 123
+        assert result["thread_id"] == "thread-abc"
+        assert result["sent_to"] == 1
+        
+        # Verify HTTP request
+        mock_requests.assert_called_once()
+        call_kwargs = mock_requests.call_args.kwargs
+        assert call_kwargs["method"] == "POST"
+        assert call_kwargs["json"]["params"]["name"] == "send_message"
+        assert call_kwargs["json"]["params"]["arguments"]["to"] == ["alice", "bob"]
+        assert call_kwargs["json"]["params"]["arguments"]["subject"] == "Test Message"
+    
+    def test_send_urgent_message(self, mock_agent_mail_env, mock_requests):
+        """Test sending urgent message."""
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {
+            "deliveries": [{
+                "payload": {"id": 456, "thread_id": "thread-xyz"}
+            }]
+        }
+        mock_requests.return_value.content = b'{"deliveries": []}'
+        
+        result = mail_send(
+            to=["alice"],
+            subject="URGENT",
+            body="Need review now!",
+            urgent=True,
+        )
+        
+        call_kwargs = mock_requests.call_args.kwargs
+        assert call_kwargs["json"]["params"]["arguments"]["importance"] == "urgent"
+    
+    def test_send_with_cc(self, mock_agent_mail_env, mock_requests):
+        """Test sending message with CC recipients."""
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {
+            "deliveries": [{
+                "payload": {"id": 789, "thread_id": "thread-123"}
+            }]
+        }
+        mock_requests.return_value.content = b'{"deliveries": []}'
+        
+        result = mail_send(
+            to=["alice"],
+            subject="FYI",
+            body="For your info",
+            cc=["bob", "charlie"],
+        )
+        
+        call_kwargs = mock_requests.call_args.kwargs
+        assert call_kwargs["json"]["params"]["arguments"]["cc"] == ["bob", "charlie"]
+    
+    def test_send_connection_error(self, mock_agent_mail_env, mock_requests):
+        """Test handling connection errors."""
+        import requests.exceptions
+        mock_requests.side_effect = requests.exceptions.ConnectionError("Connection refused")
+        
+        with pytest.raises(MailError) as exc_info:
+            mail_send(to=["alice"], subject="Test", body="Test")
+        
+        assert exc_info.value.code == "UNAVAILABLE"
+        assert "Cannot connect" in exc_info.value.message
+
+
+class TestMailInbox:
+    """Test mail_inbox function."""
+    
+    def test_fetch_inbox_default(self, mock_agent_mail_env, mock_requests):
+        """Test fetching inbox with default parameters."""
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = [
+            {
+                "id": 1,
+                "thread_id": "thread-1",
+                "from": "alice",
+                "subject": "Hello",
+                "created_ts": "2025-01-01T00:00:00Z",
+                "read_ts": None,
+                "ack_required": False,
+                "importance": "normal",
+                "body_md": "This is a test message",
+            },
+            {
+                "id": 2,
+                "thread_id": "thread-2",
+                "from": "bob",
+                "subject": "Urgent!",
+                "created_ts": "2025-01-02T00:00:00Z",
+                "read_ts": "2025-01-02T01:00:00Z",
+                "ack_required": True,
+                "importance": "urgent",
+                "body_md": "Please review ASAP",
+            },
+        ]
+        mock_requests.return_value.content = b'[]'
+        
+        result = mail_inbox()
+        
+        assert len(result["messages"]) == 2
+        assert result["messages"][0]["id"] == 1
+        assert result["messages"][0]["unread"] is True
+        assert result["messages"][0]["urgent"] is False
+        assert result["messages"][1]["id"] == 2
+        assert result["messages"][1]["unread"] is False
+        assert result["messages"][1]["urgent"] is True
+    
+    def test_fetch_inbox_unread_only(self, mock_agent_mail_env, mock_requests):
+        """Test fetching only unread messages."""
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = [
+            {"id": 1, "thread_id": "t1", "from": "alice", "subject": "Test", "created_ts": "2025-01-01T00:00:00Z", "read_ts": None, "importance": "normal"},
+            {"id": 2, "thread_id": "t2", "from": "bob", "subject": "Test2", "created_ts": "2025-01-01T00:00:00Z", "read_ts": "2025-01-01T01:00:00Z", "importance": "normal"},
+        ]
+        mock_requests.return_value.content = b'[]'
+        
+        result = mail_inbox(unread_only=True)
+        
+        # Should filter out message 2 (read)
+        assert len(result["messages"]) == 1
+        assert result["messages"][0]["id"] == 1
+    
+    def test_fetch_inbox_pagination(self, mock_agent_mail_env, mock_requests):
+        """Test inbox pagination with next_cursor."""
+        mock_requests.return_value.status_code = 200
+        # Simulate full page (limit reached)
+        mock_requests.return_value.json.return_value = [
+            {"id": i, "thread_id": f"t{i}", "from": "alice", "subject": f"Msg {i}", "created_ts": "2025-01-01T00:00:00Z", "importance": "normal"}
+            for i in range(20)
+        ]
+        mock_requests.return_value.content = b'[]'
+        
+        result = mail_inbox(limit=20)
+        
+        # Should return next_cursor when limit reached
+        assert result["next_cursor"] == "19"  # Last message ID
+
+
+class TestMailRead:
+    """Test mail_read function."""
+    
+    def test_read_message_marks_read(self, mock_agent_mail_env, mock_requests):
+        """Test reading message marks it as read by default."""
+        # Mock resource fetch
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {
+            "contents": [{
+                "id": 123,
+                "thread_id": "thread-1",
+                "from": "alice",
+                "to": ["test-agent"],
+                "subject": "Test",
+                "body_md": "Hello world!",
+                "created_ts": "2025-01-01T00:00:00Z",
+                "ack_required": False,
+                "importance": "normal",
+                "read_ts": None,
+            }]
+        }
+        mock_requests.return_value.content = b'{}'
+        
+        result = mail_read(message_id=123)
+        
+        assert result["id"] == 123
+        assert result["body"] == "Hello world!"
+        assert result["urgent"] is False
+        
+        # Should have called both GET resource and POST mark_read
+        assert mock_requests.call_count == 2
+    
+    def test_read_message_no_mark(self, mock_agent_mail_env, mock_requests):
+        """Test reading without marking as read."""
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {
+            "contents": [{
+                "id": 123,
+                "thread_id": "thread-1",
+                "from": "alice",
+                "to": ["test-agent"],
+                "subject": "Test",
+                "body_md": "Preview",
+                "created_ts": "2025-01-01T00:00:00Z",
+                "importance": "normal",
+            }]
+        }
+        mock_requests.return_value.content = b'{}'
+        
+        result = mail_read(message_id=123, mark_read=False)
+        
+        # Should only call GET resource, not mark_read
+        assert mock_requests.call_count == 1
+
+
+class TestMailReply:
+    """Test mail_reply function."""
+    
+    def test_reply_to_message(self, mock_agent_mail_env, mock_requests):
+        """Test replying to a message."""
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {
+            "reply": {
+                "id": 456,
+                "thread_id": "thread-1",
+            }
+        }
+        mock_requests.return_value.content = b'{}'
+        
+        result = mail_reply(
+            message_id=123,
+            body="Thanks for the message!",
+        )
+        
+        assert result["message_id"] == 456
+        assert result["thread_id"] == "thread-1"
+        
+        call_kwargs = mock_requests.call_args.kwargs
+        assert call_kwargs["json"]["params"]["name"] == "reply_message"
+        assert call_kwargs["json"]["params"]["arguments"]["message_id"] == 123
+
+
+class TestMailAck:
+    """Test mail_ack function."""
+    
+    def test_acknowledge_message(self, mock_agent_mail_env, mock_requests):
+        """Test acknowledging a message."""
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {}
+        mock_requests.return_value.content = b'{}'
+        
+        result = mail_ack(message_id=123)
+        
+        assert result["acknowledged"] is True
+        
+        call_kwargs = mock_requests.call_args.kwargs
+        assert call_kwargs["json"]["params"]["name"] == "acknowledge_message"
+
+
+class TestMailDelete:
+    """Test mail_delete function."""
+    
+    def test_delete_message(self, mock_agent_mail_env, mock_requests):
+        """Test deleting/archiving a message."""
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {}
+        mock_requests.return_value.content = b'{}'
+        
+        result = mail_delete(message_id=123)
+        
+        assert result["archived"] is True
+
+
+class TestMailRetries:
+    """Test retry logic and error handling."""
+    
+    def test_retries_on_server_error(self, mock_agent_mail_env, mock_requests):
+        """Test that 500 errors trigger retries."""
+        mock_requests.return_value.status_code = 500
+        mock_requests.return_value.content = b'Internal Server Error'
+        
+        with pytest.raises(MailError) as exc_info:
+            mail_send(to=["alice"], subject="Test", body="Test")
+        
+        assert exc_info.value.code == "UNAVAILABLE"
+        # Should retry 3 times total (initial + 2 retries)
+        assert mock_requests.call_count == 3
+    
+    def test_no_retry_on_client_error(self, mock_agent_mail_env, mock_requests):
+        """Test that 404 errors don't trigger retries."""
+        mock_requests.return_value.status_code = 404
+        mock_requests.return_value.json.return_value = {"detail": "Not found"}
+        mock_requests.return_value.content = b'{"detail": "Not found"}'
+        
+        with pytest.raises(MailError) as exc_info:
+            mail_read(message_id=999)
+        
+        assert exc_info.value.code == "NOT_FOUND"
+        # Should not retry on 404
+        assert mock_requests.call_count == 1
+
+
+class TestMailToolWrappers:
+    """Test MCP tool wrappers."""
+    
+    def test_mail_send_params(self, mock_agent_mail_env, mock_requests):
+        """Test MailSendParams validation."""
+        from beads_mcp.mail_tools import beads_mail_send
+        
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = {
+            "deliveries": [{
+                "payload": {"id": 123, "thread_id": "t1"}
+            }]
+        }
+        mock_requests.return_value.content = b'{}'
+        
+        params = MailSendParams(
+            to=["alice"],
+            subject="Test",
+            body="Hello",
+            urgent=True,
+        )
+        
+        result = beads_mail_send(params)
+        assert result["message_id"] == 123
+    
+    def test_mail_inbox_default_params(self, mock_agent_mail_env, mock_requests):
+        """Test MailInboxParams with defaults."""
+        from beads_mcp.mail_tools import beads_mail_inbox
+        
+        mock_requests.return_value.status_code = 200
+        mock_requests.return_value.json.return_value = []
+        mock_requests.return_value.content = b'[]'
+        
+        params = MailInboxParams()  # All defaults
+        result = beads_mail_inbox(params)
+        
+        assert result["messages"] == []
+        assert result["next_cursor"] is None