MCP Google Workspace Server

Instructions

Implementation Reference

• src/mcp_google_suite/server.py:280-293 (handler)

  MCP tool handler that validates input arguments and delegates to DriveService.search_files method.

  async def _handle_drive_search_files(
      self, context: GoogleWorkspaceContext, arguments: dict
  ) -> Dict[str, Any]:
      """Handle drive search files requests."""
      query = arguments.get("query")
      page_size = arguments.get("page_size", 10)
  
      if not query:
          raise ValueError("Search query is required")
  
      logger.debug(f"Drive search request - Query: {query}, Page Size: {page_size}")
      result = await context.drive.search_files(query=query, page_size=page_size)
      logger.debug(f"Drive search completed - Found {len(result.get('files', []))} files")
      return result

• src/mcp_google_suite/server.py:57-72 (schema)

  Input schema defining the parameters for the drive_search_files tool: query (required string) and optional page_size (integer, default 10).

  types.Tool(
      name="drive_search_files",
      description="Search for files in Google Drive",
      inputSchema={
          "type": "object",
          "properties": {
              "query": {"type": "string", "description": "Search query"},
              "page_size": {
                  "type": "integer",
                  "description": "Number of results to return",
                  "default": 10,
              },
          },
          "required": ["query"],
      },
  ),

• src/mcp_google_suite/server.py:176-182 (registration)

  Dynamic registration of tool handlers by matching _handle_{tool.name} methods to tool names in the registry.

  # Register tool handlers
  for tool in self._get_tools_list():
      handler_name = f"_handle_{tool.name}"
      if hasattr(self, handler_name):
          handler = getattr(self, handler_name)
          self._tool_registry[tool.name] = handler
          logger.debug(f"Registered handler for {tool.name}")

• src/mcp_google_suite/drive/service.py:14-26 (helper)

  DriveService method implementing the file search logic using Google Drive API v3, querying files and handling errors.

  def search_files(self, query: str, page_size: int = 10) -> Dict[str, Any]:
      """Search for files in Google Drive."""
      try:
          results = (
              self.service.files()
              .list(q=query, pageSize=page_size, fields="files(id, name, mimeType, webViewLink)")
              .execute()
          )
  
          return {"success": True, "files": results.get("files", [])}
      except HttpError as error:
          return {"success": False, **self.handle_error(error)}