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

Instructions

Searches messages in a user's Gmail account based on a query.
Returns both Message IDs and Thread IDs for each found message, along with Gmail web interface links for manual verification.

Args:
    query (str): The search query. Supports standard Gmail search operators.
    user_google_email (str): The user's Google email address. Required.
    page_size (int): The maximum number of messages to return. Defaults to 10.

Returns:
    str: LLM-friendly structured results with Message IDs, Thread IDs, and clickable Gmail web interface URLs for each found message.

Implementation Reference

• gmail/gmail_tools.py:321-364 (handler)

  The primary handler function for the 'search_gmail_messages' tool. It performs the Gmail API search using the provided query, handles responses safely, and formats the output using a helper function. Includes decorators for tool registration, error handling, and authentication requirements.

  @server.tool()
  @handle_http_errors("search_gmail_messages", is_read_only=True, service_type="gmail")
  @require_google_service("gmail", "gmail_read")
  async def search_gmail_messages(
      service, query: str, user_google_email: str, page_size: int = 10
  ) -> str:
      """
      Searches messages in a user's Gmail account based on a query.
      Returns both Message IDs and Thread IDs for each found message, along with Gmail web interface links for manual verification.
  
      Args:
          query (str): The search query. Supports standard Gmail search operators.
          user_google_email (str): The user's Google email address. Required.
          page_size (int): The maximum number of messages to return. Defaults to 10.
  
      Returns:
          str: LLM-friendly structured results with Message IDs, Thread IDs, and clickable Gmail web interface URLs for each found message.
      """
      logger.info(
          f"[search_gmail_messages] Email: '{user_google_email}', Query: '{query}'"
      )
  
      response = await asyncio.to_thread(
          service.users()
          .messages()
          .list(userId="me", q=query, maxResults=page_size)
          .execute
      )
  
      # Handle potential null response (but empty dict {} is valid)
      if response is None:
          logger.warning("[search_gmail_messages] Null response from Gmail API")
          return f"No response received from Gmail API for query: '{query}'"
  
      messages = response.get("messages", [])
      # Additional safety check for null messages array
      if messages is None:
          messages = []
  
      formatted_output = _format_gmail_results_plain(messages, query)
  
      logger.info(f"[search_gmail_messages] Found {len(messages)} messages")
      return formatted_output

• gmail/gmail_tools.py:258-319 (helper)

  Helper function specifically used by search_gmail_messages to format the list of found messages into a readable string, including IDs, web links, and usage instructions for further tools.

  def _format_gmail_results_plain(messages: list, query: str) -> str:
      """Format Gmail search results in clean, LLM-friendly plain text."""
      if not messages:
          return f"No messages found for query: '{query}'"
  
      lines = [
          f"Found {len(messages)} messages matching '{query}':",
          "",
          "📧 MESSAGES:",
      ]
  
      for i, msg in enumerate(messages, 1):
          # Handle potential null/undefined message objects
          if not msg or not isinstance(msg, dict):
              lines.extend([
                  f"  {i}. Message: Invalid message data",
                  "     Error: Message object is null or malformed",
                  "",
              ])
              continue
  
          # Handle potential null/undefined values from Gmail API
          message_id = msg.get("id")
          thread_id = msg.get("threadId")
  
          # Convert None, empty string, or missing values to "unknown"
          if not message_id:
              message_id = "unknown"
          if not thread_id:
              thread_id = "unknown"
  
          if message_id != "unknown":
              message_url = _generate_gmail_web_url(message_id)
          else:
              message_url = "N/A"
  
          if thread_id != "unknown":
              thread_url = _generate_gmail_web_url(thread_id)
          else:
              thread_url = "N/A"
  
          lines.extend(
              [
                  f"  {i}. Message ID: {message_id}",
                  f"     Web Link: {message_url}",
                  f"     Thread ID: {thread_id}",
                  f"     Thread Link: {thread_url}",
                  "",
              ]
          )
  
      lines.extend(
          [
              "💡 USAGE:",
              "  • Pass the Message IDs **as a list** to get_gmail_messages_content_batch()",
              "    e.g. get_gmail_messages_content_batch(message_ids=[...])",
              "  • Pass the Thread IDs to get_gmail_thread_content() (single) or get_gmail_threads_content_batch() (batch)",
          ]
      )
  
      return "\n".join(lines)

• gmail/gmail_tools.py:321-321 (registration)

  The @server.tool() decorator registers the search_gmail_messages function as an MCP tool.

  @server.tool()