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

Instructions

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}"