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

Instructions

Searches for messages in Google Chat spaces by text content.

Returns:
    str: A formatted list of messages matching the search query.

Implementation Reference

• gchat/chat_tools.py:155-155 (registration)

  Registers the search_messages function as an MCP tool using the @server.tool() decorator.

  @server.tool()

• gchat/chat_tools.py:158-226 (handler)

  The core implementation of the search_messages tool. Searches Google Chat messages by query string, optionally within a specific space or across multiple spaces. Handles API calls, formats results, and logs activity.

  async def search_messages(
      service,
      user_google_email: str,
      query: str,
      space_id: Optional[str] = None,
      page_size: int = 25
  ) -> str:
      """
      Searches for messages in Google Chat spaces by text content.
  
      Returns:
          str: A formatted list of messages matching the search query.
      """
      logger.info(f"[search_messages] Email={user_google_email}, Query='{query}'")
  
      # If specific space provided, search within that space
      if space_id:
          response = await asyncio.to_thread(
              service.spaces().messages().list(
                  parent=space_id,
                  pageSize=page_size,
                  filter=f'text:"{query}"'
              ).execute
          )
          messages = response.get('messages', [])
          context = f"space '{space_id}'"
      else:
          # Search across all accessible spaces (this may require iterating through spaces)
          # For simplicity, we'll search the user's spaces first
          spaces_response = await asyncio.to_thread(
              service.spaces().list(pageSize=100).execute
          )
          spaces = spaces_response.get('spaces', [])
  
          messages = []
          for space in spaces[:10]:  # Limit to first 10 spaces to avoid timeout
              try:
                  space_messages = await asyncio.to_thread(
                      service.spaces().messages().list(
                          parent=space.get('name'),
                          pageSize=5,
                          filter=f'text:"{query}"'
                      ).execute
                  )
                  space_msgs = space_messages.get('messages', [])
                  for msg in space_msgs:
                      msg['_space_name'] = space.get('displayName', 'Unknown')
                  messages.extend(space_msgs)
              except HttpError:
                  continue  # Skip spaces we can't access
          context = "all accessible spaces"
  
      if not messages:
          return f"No messages found matching '{query}' in {context}."
  
      output = [f"Found {len(messages)} messages matching '{query}' in {context}:"]
      for msg in messages:
          sender = msg.get('sender', {}).get('displayName', 'Unknown Sender')
          create_time = msg.get('createTime', 'Unknown Time')
          text_content = msg.get('text', 'No text content')
          space_name = msg.get('_space_name', 'Unknown Space')
  
          # Truncate long messages
          if len(text_content) > 100:
              text_content = text_content[:100] + "..."
  
          output.append(f"- [{create_time}] {sender} in '{space_name}': {text_content}")
  
      return "\n".join(output)

• gchat/chat_tools.py:158-170 (schema)

  Function signature and docstring defining the input parameters and output for the search_messages tool.

  async def search_messages(
      service,
      user_google_email: str,
      query: str,
      space_id: Optional[str] = None,
      page_size: int = 25
  ) -> str:
      """
      Searches for messages in Google Chat spaces by text content.
  
      Returns:
          str: A formatted list of messages matching the search query.
      """