bulk_delete_emails

Instructions

Implementation Reference

• src/productivity_mcp/server.py:826-846 (handler)

  The handler function for bulk_delete_emails. Decorated with @mcp.tool() and @_logged (which applies rate-limiting). Iterates over message_ids calling provider.delete_message() on each, returning counts of successes and failures.

  @mcp.tool()
  @_logged
  def bulk_delete_emails(
      message_ids: list[str],
      account: str | None = None,
      folder: str = "INBOX",
  ) -> dict[str, Any]:
      """Delete many emails in one call. IMAP expunges; Gmail moves to Trash.
      Rate-limited."""
      provider = _email(account)
      ok, failed = 0, 0
      errors: list[str] = []
      for mid in message_ids:
          try:
              provider.delete_message(mid, folder=folder)
              ok += 1
          except Exception as exc:
              failed += 1
              if len(errors) < 5:
                  errors.append(f"{mid}: {exc}")
      return {"ok": ok, "failed": failed, "errors": errors}

• src/productivity_mcp/server.py:826-828 (registration)

  Registration via @mcp.tool() decorator on FastMCP instance.

  @mcp.tool()
  @_logged
  def bulk_delete_emails(

• src/productivity_mcp/util/rate_limit.py:26-61 (helper)

  Rate-limit configuration: 10 calls per 60-second window for bulk_delete_emails.

      "bulk_delete_emails": (10, 60.0),
      "bulk_move_emails": (10, 60.0),
      "create_event": (30, 60.0),
      "update_event": (30, 60.0),
  }
  
  
  class RateLimiter:
      def __init__(self, limits: dict[str, tuple[int, float]] | None = None) -> None:
          self._limits = dict(_DEFAULT_LIMITS)
          if limits:
              self._limits.update(limits)
          self._hits: dict[str, deque[float]] = {}
          self._lock = threading.Lock()
  
      def check(self, tool: str) -> None:
          limit = self._limits.get(tool)
          if limit is None:
              return
          max_calls, window = limit
          now = time.time()
          with self._lock:
              bucket = self._hits.setdefault(tool, deque())
              cutoff = now - window
              while bucket and bucket[0] < cutoff:
                  bucket.popleft()
              if len(bucket) >= max_calls:
                  wait = window - (now - bucket[0])
                  raise RuntimeError(
                      f"Rate limit exceeded for '{tool}': {max_calls} calls per "
                      f"{int(window)}s. Retry in {int(wait) + 1}s."
                  )
              bucket.append(now)
  
  
  limiter = RateLimiter()

• src/productivity_mcp/providers/gmail_email.py:141-143 (helper)

  Gmail implementation of delete_message — moves message to Trash (recoverable) via Gmail API.

  def delete_message(self, message_id: str, folder: str = "INBOX") -> None:
      # Move to Trash (recoverable). Use `delete` for permanent removal.
      self._svc().users().messages().trash(userId="me", id=message_id).execute()

• src/productivity_mcp/providers/imap_email.py:158-162 (helper)

  IMAP implementation of delete_message — marks message as \Deleted and expunges immediately.

  def delete_message(self, message_id: str, folder: str = "INBOX") -> None:
      with self._lock:
          conn = self._select(folder, readonly=False)
          conn.uid("STORE", message_id, "+FLAGS", "(\\Deleted)")
          conn.expunge()