feat(gmail): add get_gmail_label_stats tool for unread/total message counts

gmail/gmail_tools.py

@@ -2573,6 +2573,63 @@ async def list_gmail_labels(service, user_google_email: str) -> str:
     return "\n".join(lines)
 
 
+@server.tool()
+@handle_http_errors("get_gmail_label_stats", is_read_only=True, service_type="gmail")
+@require_google_service("gmail", "gmail_read")
+async def get_gmail_label_stats(
+    service,
+    user_google_email: str,
+    label_id: Annotated[
+        str,
+        Field(
+            description="The Gmail label ID to get stats for (e.g. 'INBOX', 'UNREAD', "
+            "'STARRED', 'SENT', 'DRAFT', 'SPAM', 'TRASH', or a custom label ID). "
+            "Defaults to 'INBOX'."
+        ),
+    ] = "INBOX",
+) -> str:
+    """
+    Returns message and thread counts for a Gmail label.
+
+    Provides messagesTotal, messagesUnread, threadsTotal, and threadsUnread
+    in a single API call — much faster than paginating through search results
+    to count unread messages.
+
+    Args:
+        user_google_email (str): The user's Google email address. Required.
+        label_id (str): The Gmail label ID. Defaults to 'INBOX'.
+
+    Returns:
+        str: Formatted label statistics including message and thread counts.
+    """
+    logger.info(
+        f"[get_gmail_label_stats] Invoked. Email: '{user_google_email}', "
+        f"label_id: '{label_id}'"
+    )
+
+    response = await asyncio.to_thread(
+        service.users().labels().get(userId="me", id=label_id).execute
+    )
+
+    name = response.get("name", label_id)
+    label_type = response.get("type", "unknown")
+    messages_total = response.get("messagesTotal", 0)
+    messages_unread = response.get("messagesUnread", 0)
+    threads_total = response.get("threadsTotal", 0)
+    threads_unread = response.get("threadsUnread", 0)
+
+    lines = [
+        f"📊 Label stats for '{name}' (ID: {label_id}, type: {label_type}):",
+        "",
+        f"  Messages total:  {messages_total}",
+        f"  Messages unread: {messages_unread}",
+        f"  Threads total:   {threads_total}",
+        f"  Threads unread:  {threads_unread}",
+    ]
+
+    return "\n".join(lines)
+
+
 @server.tool()
 @handle_http_errors("manage_gmail_label", service_type="gmail")
 @require_google_service("gmail", GMAIL_LABELS_SCOPE)

tests/gmail/test_get_gmail_label_stats.py

@@ -0,0 +1,122 @@
+import os
+import sys
+from unittest.mock import Mock
+
+import pytest
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
+
+from gmail.gmail_tools import get_gmail_label_stats
+
+
+def _unwrap(tool):
+    """Unwrap FunctionTool + decorators to the original async function."""
+    fn = tool.fn if hasattr(tool, "fn") else tool
+    while hasattr(fn, "__wrapped__"):
+        fn = fn.__wrapped__
+    return fn
+
+
+def _mock_service(label_response: dict) -> Mock:
+    """Build a mock Gmail service that returns the given label response."""
+    service = Mock()
+    service.users.return_value.labels.return_value.get.return_value.execute.return_value = (
+        label_response
+    )
+    return service
+
+
+@pytest.mark.asyncio
+async def test_inbox_default():
+    """Default label_id='INBOX' returns formatted stats."""
+    response = {
+        "id": "INBOX",
+        "name": "INBOX",
+        "type": "system",
+        "messagesTotal": 1234,
+        "messagesUnread": 56,
+        "threadsTotal": 800,
+        "threadsUnread": 30,
+    }
+    service = _mock_service(response)
+    fn = _unwrap(get_gmail_label_stats)
+
+    result = await fn(service, user_google_email="test@example.com")
+
+    assert "INBOX" in result
+    assert "1234" in result
+    assert "56" in result
+    assert "800" in result
+    assert "30" in result
+    service.users().labels().get.assert_called_once_with(userId="me", id="INBOX")
+
+
+@pytest.mark.asyncio
+async def test_custom_label_id():
+    """Passing a custom label_id calls the API with that ID."""
+    response = {
+        "id": "Label_42",
+        "name": "Projects",
+        "type": "user",
+        "messagesTotal": 10,
+        "messagesUnread": 3,
+        "threadsTotal": 7,
+        "threadsUnread": 2,
+    }
+    service = _mock_service(response)
+    fn = _unwrap(get_gmail_label_stats)
+
+    result = await fn(
+        service, user_google_email="test@example.com", label_id="Label_42"
+    )
+
+    assert "Projects" in result
+    assert "Label_42" in result
+    assert "user" in result
+    assert "10" in result
+    assert "3" in result
+    service.users().labels().get.assert_called_once_with(userId="me", id="Label_42")
+
+
+@pytest.mark.asyncio
+async def test_zero_counts():
+    """Labels with zero messages/threads are handled gracefully."""
+    response = {
+        "id": "TRASH",
+        "name": "TRASH",
+        "type": "system",
+        "messagesTotal": 0,
+        "messagesUnread": 0,
+        "threadsTotal": 0,
+        "threadsUnread": 0,
+    }
+    service = _mock_service(response)
+    fn = _unwrap(get_gmail_label_stats)
+
+    result = await fn(service, user_google_email="test@example.com", label_id="TRASH")
+
+    assert "TRASH" in result
+    assert "Messages total:  0" in result
+    assert "Threads total:   0" in result
+
+
+@pytest.mark.asyncio
+async def test_missing_optional_fields():
+    """When API response omits count fields, defaults to 0."""
+    response = {
+        "id": "STARRED",
+        "name": "STARRED",
+        "type": "system",
+    }
+    service = _mock_service(response)
+    fn = _unwrap(get_gmail_label_stats)
+
+    result = await fn(
+        service, user_google_email="test@example.com", label_id="STARRED"
+    )
+
+    assert "STARRED" in result
+    assert "Messages total:  0" in result
+    assert "Messages unread: 0" in result
+    assert "Threads total:   0" in result
+    assert "Threads unread:  0" in result