Google Workspace MCP Server - Control Gmail, Calendar, Docs, Sheets, Slides, Chat, Forms & Drive

Instructions

Retrieves the content of multiple Gmail messages in a single batch request.
Supports up to 100 messages per request using Google's batch API.

Args:
    message_ids (List[str]): List of Gmail message IDs to retrieve (max 100).
    user_google_email (str): The user's Google email address. Required.
    format (Literal["full", "metadata"]): Message format. "full" includes body, "metadata" only headers.

Returns:
    str: A formatted list of message contents with separators.

Implementation Reference

• gmail/gmail_tools.py:460-639 (handler)

  Main handler function implementing get_gmail_messages_content_batch tool. Uses Gmail API batch requests with fallback to sequential fetches for robustness against SSL errors. Extracts headers, bodies (text/HTML), and generates web links.

  @server.tool()
  @handle_http_errors("get_gmail_messages_content_batch", is_read_only=True, service_type="gmail")
  @require_google_service("gmail", "gmail_read")
  async def get_gmail_messages_content_batch(
      service,
      message_ids: List[str],
      user_google_email: str,
      format: Literal["full", "metadata"] = "full",
  ) -> str:
      """
      Retrieves the content of multiple Gmail messages in a single batch request.
      Supports up to 25 messages per batch to prevent SSL connection exhaustion.
  
      Args:
          message_ids (List[str]): List of Gmail message IDs to retrieve (max 25 per batch).
          user_google_email (str): The user's Google email address. Required.
          format (Literal["full", "metadata"]): Message format. "full" includes body, "metadata" only headers.
  
      Returns:
          str: A formatted list of message contents including subject, sender, recipients (To, Cc), and body (if full format).
      """
      logger.info(
          f"[get_gmail_messages_content_batch] Invoked. Message count: {len(message_ids)}, Email: '{user_google_email}'"
      )
  
      if not message_ids:
          raise Exception("No message IDs provided")
  
      output_messages = []
  
      # Process in smaller chunks to prevent SSL connection exhaustion
      for chunk_start in range(0, len(message_ids), GMAIL_BATCH_SIZE):
          chunk_ids = message_ids[chunk_start : chunk_start + GMAIL_BATCH_SIZE]
          results: Dict[str, Dict] = {}
  
          def _batch_callback(request_id, response, exception):
              """Callback for batch requests"""
              results[request_id] = {"data": response, "error": exception}
  
          # Try to use batch API
          try:
              batch = service.new_batch_http_request(callback=_batch_callback)
  
              for mid in chunk_ids:
                  if format == "metadata":
                      req = (
                          service.users()
                          .messages()
                          .get(
                              userId="me",
                              id=mid,
                              format="metadata",
                              metadataHeaders=["Subject", "From", "To", "Cc"],
                          )
                      )
                  else:
                      req = (
                          service.users()
                          .messages()
                          .get(userId="me", id=mid, format="full")
                      )
                  batch.add(req, request_id=mid)
  
              # Execute batch request
              await asyncio.to_thread(batch.execute)
  
          except Exception as batch_error:
              # Fallback to sequential processing instead of parallel to prevent SSL exhaustion
              logger.warning(
                  f"[get_gmail_messages_content_batch] Batch API failed, falling back to sequential processing: {batch_error}"
              )
  
              async def fetch_message_with_retry(mid: str, max_retries: int = 3):
                  """Fetch a single message with exponential backoff retry for SSL errors"""
                  for attempt in range(max_retries):
                      try:
                          if format == "metadata":
                              msg = await asyncio.to_thread(
                                  service.users()
                                  .messages()
                                  .get(
                                      userId="me",
                                      id=mid,
                                      format="metadata",
                                      metadataHeaders=["Subject", "From", "To", "Cc"],
                                  )
                                  .execute
                              )
                          else:
                              msg = await asyncio.to_thread(
                                  service.users()
                                  .messages()
                                  .get(userId="me", id=mid, format="full")
                                  .execute
                              )
                          return mid, msg, None
                      except ssl.SSLError as ssl_error:
                          if attempt < max_retries - 1:
                              # Exponential backoff: 1s, 2s, 4s
                              delay = 2 ** attempt
                              logger.warning(
                                  f"[get_gmail_messages_content_batch] SSL error for message {mid} on attempt {attempt + 1}: {ssl_error}. Retrying in {delay}s..."
                              )
                              await asyncio.sleep(delay)
                          else:
                              logger.error(
                                  f"[get_gmail_messages_content_batch] SSL error for message {mid} on final attempt: {ssl_error}"
                              )
                              return mid, None, ssl_error
                      except Exception as e:
                          return mid, None, e
  
              # Process messages sequentially with small delays to prevent connection exhaustion
              for mid in chunk_ids:
                  mid_result, msg_data, error = await fetch_message_with_retry(mid)
                  results[mid_result] = {"data": msg_data, "error": error}
                  # Brief delay between requests to allow connection cleanup
                  await asyncio.sleep(GMAIL_REQUEST_DELAY)
  
          # Process results for this chunk
          for mid in chunk_ids:
              entry = results.get(mid, {"data": None, "error": "No result"})
  
              if entry["error"]:
                  output_messages.append(f"⚠️ Message {mid}: {entry['error']}\n")
              else:
                  message = entry["data"]
                  if not message:
                      output_messages.append(f"⚠️ Message {mid}: No data returned\n")
                      continue
  
                  # Extract content based on format
                  payload = message.get("payload", {})
  
                  if format == "metadata":
                      headers = _extract_headers(payload, ["Subject", "From", "To", "Cc"])
                      subject = headers.get("Subject", "(no subject)")
                      sender = headers.get("From", "(unknown sender)")
                      to = headers.get("To", "")
                      cc = headers.get("Cc", "")
  
                      msg_output = f"Message ID: {mid}\nSubject: {subject}\nFrom: {sender}\n"
                      if to:
                          msg_output += f"To: {to}\n"
                      if cc:
                          msg_output += f"Cc: {cc}\n"
                      msg_output += f"Web Link: {_generate_gmail_web_url(mid)}\n"
  
                      output_messages.append(msg_output)
                  else:
                      # Full format - extract body too
                      headers = _extract_headers(payload, ["Subject", "From", "To", "Cc"])
                      subject = headers.get("Subject", "(no subject)")
                      sender = headers.get("From", "(unknown sender)")
                      to = headers.get("To", "")
                      cc = headers.get("Cc", "")
  
                      # Extract both text and HTML bodies using enhanced helper function
                      bodies = _extract_message_bodies(payload)
                      text_body = bodies.get("text", "")
                      html_body = bodies.get("html", "")
  
                      # Format body content with HTML fallback
                      body_data = _format_body_content(text_body, html_body)
  
                      msg_output = f"Message ID: {mid}\nSubject: {subject}\nFrom: {sender}\n"
                      if to:
                          msg_output += f"To: {to}\n"
                      if cc:
                          msg_output += f"Cc: {cc}\n"
                      msg_output += f"Web Link: {_generate_gmail_web_url(mid)}\n\n{body_data}\n"
  
                      output_messages.append(msg_output)
  
      # Combine all messages with separators
      final_output = f"Retrieved {len(message_ids)} messages:\n\n"
      final_output += "\n---\n\n".join(output_messages)
  
      return final_output

• gmail/gmail_tools.py:50-99 (helper)

  Helper to extract plain text and HTML bodies from Gmail message payloads by traversing MIME parts.

  def _extract_message_bodies(payload):
      """
      Helper function to extract both plain text and HTML bodies from a Gmail message payload.
  
      Args:
          payload (dict): The message payload from Gmail API
  
      Returns:
          dict: Dictionary with 'text' and 'html' keys containing body content
      """
      text_body = ""
      html_body = ""
      parts = [payload] if "parts" not in payload else payload.get("parts", [])
  
      part_queue = list(parts)  # Use a queue for BFS traversal of parts
      while part_queue:
          part = part_queue.pop(0)
          mime_type = part.get("mimeType", "")
          body_data = part.get("body", {}).get("data")
  
          if body_data:
              try:
                  decoded_data = base64.urlsafe_b64decode(body_data).decode("utf-8", errors="ignore")
                  if mime_type == "text/plain" and not text_body:
                      text_body = decoded_data
                  elif mime_type == "text/html" and not html_body:
                      html_body = decoded_data
              except Exception as e:
                  logger.warning(f"Failed to decode body part: {e}")
  
          # Add sub-parts to queue for multipart messages
          if mime_type.startswith("multipart/") and "parts" in part:
              part_queue.extend(part.get("parts", []))
  
      # Check the main payload if it has body data directly
      if payload.get("body", {}).get("data"):
          try:
              decoded_data = base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", errors="ignore")
              mime_type = payload.get("mimeType", "")
              if mime_type == "text/plain" and not text_body:
                  text_body = decoded_data
              elif mime_type == "text/html" and not html_body:
                  html_body = decoded_data
          except Exception as e:
              logger.warning(f"Failed to decode main payload body: {e}")
  
      return {
          "text": text_body,
          "html": html_body
      }

• gmail/gmail_tools.py:102-122 (helper)

  Helper to format message body content, preferring plain text over HTML with truncation for large HTML.

  def _format_body_content(text_body: str, html_body: str) -> str:
      """
      Helper function to format message body content with HTML fallback and truncation.
  
      Args:
          text_body: Plain text body content
          html_body: HTML body content
  
      Returns:
          Formatted body content string
      """
      if text_body.strip():
          return text_body
      elif html_body.strip():
          # Truncate very large HTML to keep responses manageable
          if len(html_body) > HTML_BODY_TRUNCATE_LIMIT:
              html_body = html_body[:HTML_BODY_TRUNCATE_LIMIT] + "\n\n[HTML content truncated...]"
          return f"[HTML Content Converted]\n{html_body}"
      else:
          return "[No readable content found]"

• gmail/gmail_tools.py:157-173 (helper)

  Helper to extract specific headers like Subject, From, To, Cc from message payload.

  def _extract_headers(payload: dict, header_names: List[str]) -> Dict[str, str]:
      """
      Extract specified headers from a Gmail message payload.
  
      Args:
          payload: The message payload from Gmail API
          header_names: List of header names to extract
  
      Returns:
          Dict mapping header names to their values
      """
      headers = {}
      for header in payload.get("headers", []):
          if header["name"] in header_names:
              headers[header["name"]] = header["value"]
      return headers

• gmail/gmail_tools.py:243-256 (helper)

  Helper to generate direct Gmail web interface URLs for messages/threads.

  def _generate_gmail_web_url(item_id: str, account_index: int = 0) -> str:
      """
      Generate Gmail web interface URL for a message or thread ID.
      Uses #all to access messages from any Gmail folder/label (not just inbox).
  
      Args:
          item_id: Gmail message ID or thread ID
          account_index: Google account index (default 0 for primary account)
  
      Returns:
          Gmail web interface URL that opens the message/thread in Gmail web interface
      """
      return f"https://mail.google.com/mail/u/{account_index}/#all/{item_id}"