diff --git a/docs/config.rst b/docs/config.rst
index f613bb94..fa10d25d 100644
--- a/docs/config.rst
+++ b/docs/config.rst
@@ -540,3 +540,19 @@ Default: None
 Sets the URI to which an OAuth 2.0 server redirects the user after successful authentication and authorization.
 
 `oauth2_redirect_uri` option should be used with :ref:`auth`, :ref:`auth_provider`, :ref:`oauth2_key` and :ref:`oauth2_secret` options.
+
+.. _read_only:
+
+read_only
+~~~~~~~~~
+
+Default: False
+
+Enables read only mode, disabling all control operations in the UI and API.
+
+When read only mode is enabled, Flower will not allow any control operations to be performed on the system.
+
+Example::
+
+    $ celery flower --read_only
+
diff --git a/flower/api/control.py b/flower/api/control.py
index 41f39c36..c8b7326b 100644
--- a/flower/api/control.py
+++ b/flower/api/control.py
@@ -51,8 +51,12 @@ def post(self, workername):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
+:statuscode 403: read only mode is enabled
 :statuscode 404: unknown worker
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         if not self.is_worker(workername):
             raise web.HTTPError(404, f"Unknown worker '{workername}'")
 
@@ -90,9 +94,12 @@ def post(self, workername):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
-:statuscode 403: pool restart is not enabled (see CELERYD_POOL_RESTARTS)
+:statuscode 403: pool restart is not enabled (see CELERYD_POOL_RESTARTS) or read only mode is enabled
 :statuscode 404: unknown worker
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         if not self.is_worker(workername):
             raise web.HTTPError(404, f"Unknown worker '{workername}'")
 
@@ -139,10 +146,13 @@ def post(self, workername):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
-:statuscode 403: failed to grow
+:statuscode 403: failed to grow or read only mode is enabled
 :statuscode 404: unknown worker
         """
 
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         if not self.is_worker(workername):
             raise web.HTTPError(404, f"Unknown worker '{workername}'")
 
@@ -190,10 +200,13 @@ def post(self, workername):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
-:statuscode 403: failed to shrink
+:statuscode 403: failed to shrink or read only mode is enabled
 :statuscode 404: unknown worker
         """
 
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         if not self.is_worker(workername):
             raise web.HTTPError(404, f"Unknown worker '{workername}'")
 
@@ -243,9 +256,11 @@ def post(self, workername):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
-:statuscode 403: autoscaling is not enabled (see CELERYD_AUTOSCALER)
+:statuscode 403: autoscaling is not enabled (see CELERYD_AUTOSCALER) or read only mode is enabled
 :statuscode 404: unknown worker
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
 
         if not self.is_worker(workername):
             raise web.HTTPError(404, f"Unknown worker '{workername}'")
@@ -299,9 +314,12 @@ def post(self, workername):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
-:statuscode 403: failed to add consumer
+:statuscode 403: failed to add consumer or read only mode is enabled
 :statuscode 404: unknown worker
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         if not self.is_worker(workername):
             raise web.HTTPError(404, f"Unknown worker '{workername}'")
 
@@ -352,9 +370,12 @@ def post(self, workername):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
-:statuscode 403: failed to cancel consumer
+:statuscode 403: failed to cancel consumer or read only mode is enabled
 :statuscode 404: unknown worker
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         if not self.is_worker(workername):
             raise web.HTTPError(404, f"Unknown worker '{workername}'")
 
@@ -406,7 +427,11 @@ def post(self, taskid):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
+:statuscode 403: read only mode is enabled
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         logger.info("Revoking task '%s'", taskid)
         terminate = self.get_argument('terminate', default=False, type=bool)
         signal = self.get_argument('signal', default='SIGTERM', type=str)
@@ -447,8 +472,12 @@ def post(self, taskname):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
+:statuscode 403: read only mode is enabled
 :statuscode 404: unknown task/worker
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         workername = self.get_argument('workername')
         hard = self.get_argument('hard', default=None, type=float)
         soft = self.get_argument('soft', default=None, type=float)
@@ -507,8 +536,12 @@ def post(self, taskname):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
+:statuscode 403: read only mode is enabled
 :statuscode 404: unknown task/worker
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         workername = self.get_argument('workername')
         ratelimit = self.get_argument('ratelimit')
 
diff --git a/flower/api/tasks.py b/flower/api/tasks.py
index 730c290e..c08abf1d 100644
--- a/flower/api/tasks.py
+++ b/flower/api/tasks.py
@@ -117,8 +117,12 @@ async def post(self, taskname):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
+:statuscode 403: read only mode is enabled
 :statuscode 404: unknown task
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         args, kwargs, options = self.get_task_args()
         logger.debug("Invoking a task '%s' with '%s' and '%s'",
                      taskname, args, kwargs)
@@ -192,8 +196,12 @@ def post(self, taskname):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
+:statuscode 403: read only mode is enabled
 :statuscode 404: unknown task
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         args, kwargs, options = self.get_task_args()
         logger.debug("Invoking a task '%s' with '%s' and '%s'",
                      taskname, args, kwargs)
@@ -254,8 +262,12 @@ def post(self, taskname):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
+:statuscode 403: read only mode is enabled
 :statuscode 404: unknown task
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         args, kwargs, options = self.get_task_args()
         logger.debug("Invoking task '%s' with '%s' and '%s'",
                      taskname, args, kwargs)
@@ -344,8 +356,12 @@ def post(self, taskid):
 :reqheader Authorization: optional OAuth token to authenticate
 :statuscode 200: no error
 :statuscode 401: unauthorized request
+:statuscode 403: read only mode is enabled
 :statuscode 503: result backend is not configured
         """
+        if self.application.options.read_only:
+            raise web.HTTPError(403, "Read only mode is enabled")
+
         logger.info("Aborting task '%s'", taskid)
 
         result = AbortableAsyncResult(taskid)
diff --git a/flower/options.py b/flower/options.py
index 083d4b5b..294f4ee0 100644
--- a/flower/options.py
+++ b/flower/options.py
@@ -68,6 +68,7 @@
 define("url_prefix", type=str, help="base url prefix")
 define("task_runtime_metric_buckets", type=float, default=Histogram.DEFAULT_BUCKETS,
        multiple=True, help="histogram latency bucket value")
-
+define("read_only", type=bool, default=False,
+       help="enable read only mode, disabling all operations")
 
 default_options = options
diff --git a/flower/templates/worker.html b/flower/templates/worker.html
index 9dd64daf..fb646999 100644
--- a/flower/templates/worker.html
+++ b/flower/templates/worker.html
@@ -16,6 +16,7 @@
         <h3 id="workername">{{ worker['name'] }}</h3>
       </div>
 
+      {% if not read_only %}
       <div class="btn-group float-end" role="group" aria-label="Button Group">
         <button id="worker-group" type="button" class="btn btn-primary dropdown-toggle" data-bs-toggle="dropdown"
           aria-expanded="false">
@@ -28,6 +29,7 @@ <h3 id="workername">{{ worker['name'] }}</h3>
           <li><a id="worker-refresh-all" class="dropdown-item" data-bs-dismiss="dropdown">Refresh All</a></li>
         </ul>
       </div>
+      {% end %}
 
       <div class="tabbable">
         <ul class="nav nav-tabs" role="tablist">
@@ -84,6 +86,7 @@ <h3 id="workername">{{ worker['name'] }}</h3>
                   </table>
                 </div>
 
+                {% if not read_only %}
                 <div class="col-lg-4 container">
                   <form class="mx-auto">
                     <legend class="form-label mt-md-5">Pool size control</legend>
@@ -103,6 +106,7 @@ <h3 id="workername">{{ worker['name'] }}</h3>
                     </div>
                   </form>
                 </div>
+                {% end %}
               </div>
 
               {% if worker['stats'].get('autoscaler', None) %}
@@ -154,7 +158,9 @@ <h3 id="workername">{{ worker['name'] }}</h3>
                   <th>Queue arguments</th>
                   <th>Binding arguments</th>
                   <th>Auto delete</th>
+                  {% if not read_only %}
                   <th style="width: 125px;"></th>
+                  {% end %}
                 </tr>
               </thead>
               <tbody id="worker-queues">
@@ -170,12 +176,15 @@ <h3 id="workername">{{ worker['name'] }}</h3>
                   <td>{{ queue['queue_arguments'] }}</td>
                   <td>{{ queue['binding_arguments'] }}</td>
                   <td>{{ queue['auto_delete'] }}</td>
-                  <td><button id="worker-cancel-consumer-{{ queue['name'] }}" class="btn btn-danger text-nowrap">Cancel
-                      Consumer</button></td>
+                  {% if not read_only %}
+                    <td><button id="worker-cancel-consumer-{{ queue['name'] }}" class="btn btn-danger text-nowrap">Cancel
+                        Consumer</button></td>
+                  {% end %}
                 </tr>
                 {% end %}
               </tbody>
             </table>
+            {% if not read_only %}
             <div class="control-group col-lg-3">
               <div class="input-group mb-3">
                 <input id="add-consumer-name" type="text" class="form-control" placeholder="New consumer"
@@ -183,6 +192,7 @@ <h3 id="workername">{{ worker['name'] }}</h3>
                 <button class="btn btn-primary mx-1" type="button" id="worker-add-consumer">Add</button>
               </div>
             </div>
+            {% end %}
           </div> <!-- end queues tab -->
 
           <div class="tab-pane fade" id="tab-tasks" role="tabpanel" aria-labelledby="tab-tasks">
@@ -304,8 +314,10 @@ <h3 id="workername">{{ worker['name'] }}</h3>
                       <div class="form-group">
                         <div class="input-group">
                           <input class="form-control form-control-sm" type="number">
+                          {% if not read_only %}
                           <button class="btn btn-primary btn-sm mx-1" type="button"
                             id="task-rate-limit-{{taskname}}">Apply</button>
+                          {% end %}
                         </div>
                       </div>
                     </td>
@@ -313,10 +325,12 @@ <h3 id="workername">{{ worker['name'] }}</h3>
                       <div class="form-group">
                         <div class="input-group">
                           <input class="form-control form-control-sm" type="number">
+                          {% if not read_only %}
                           <button class="btn btn-primary btn-sm mx-1" type="button"
                             id="task-timeout-soft-{{taskname}}">Soft</button>
                           <button class="btn btn-primary btn-sm mx-1" type="button"
                             id="task-timeout-hard-{{taskname}}">Hard</button>
+                          {% end %}
                         </div>
                       </div>
                     </td>
@@ -387,4 +401,5 @@ <h3 id="workername">{{ worker['name'] }}</h3>
       </div>
     </div>
   </div>
-  {% end %}
\ No newline at end of file
+</div>
+{% end %}
diff --git a/flower/views/workers.py b/flower/views/workers.py
index defd0469..ca0a40b3 100644
--- a/flower/views/workers.py
+++ b/flower/views/workers.py
@@ -24,7 +24,11 @@ async def get(self, name):
         if 'stats' not in worker:
             raise web.HTTPError(404, f"Unable to get stats for '{name}' worker")
 
-        self.render("worker.html", worker=dict(worker, name=name))
+        self.render(
+            "worker.html",
+            worker=dict(worker, name=name),
+            read_only=self.application.options.read_only,
+        )
 
 
 class WorkersView(BaseHandler):
diff --git a/tests/unit/api/test_control.py b/tests/unit/api/test_control.py
index 4eb279be..b66d3092 100644
--- a/tests/unit/api/test_control.py
+++ b/tests/unit/api/test_control.py
@@ -31,6 +31,13 @@ def test_shutdown(self):
         self.assertEqual(200, r.code)
         celery.control.broadcast.assert_called_once_with('shutdown',
                                                          destination=['test'])
+    def test_shutdown_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.broadcast = MagicMock()
+            r = self.post('/api/worker/shutdown/test', body={})
+            self.assertEqual(403, r.code)
+            celery.control.broadcast.assert_not_called()
 
     def test_pool_restart(self):
         celery = self._app.capp
@@ -44,6 +51,14 @@ def test_pool_restart(self):
             reply=True,
         )
 
+    def test_pool_restart_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.broadcast = MagicMock()
+            r = self.post('/api/worker/pool/restart/test', body={})
+            self.assertEqual(403, r.code)
+            celery.control.broadcast.assert_not_called()
+
     def test_pool_grow(self):
         celery = self._app.capp
         celery.control.pool_grow = MagicMock(return_value=[{'test': 'ok'}])
@@ -52,6 +67,14 @@ def test_pool_grow(self):
         celery.control.pool_grow.assert_called_once_with(
             n=3, reply=True, destination=['test'])
 
+    def test_pool_grow_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.pool_grow = MagicMock()
+            r = self.post('/api/worker/pool/grow/test', body={'n': 3})
+            self.assertEqual(403, r.code)
+            celery.control.pool_grow.assert_not_called()
+
     def test_pool_shrink(self):
         celery = self._app.capp
         celery.control.pool_shrink = MagicMock(return_value=[{'test': 'ok'}])
@@ -60,6 +83,14 @@ def test_pool_shrink(self):
         celery.control.pool_shrink.assert_called_once_with(
             n=1, reply=True, destination=['test'])
 
+    def test_pool_shrink_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.pool_shrink = MagicMock()
+            r = self.post('/api/worker/pool/shrink/test', body={})
+            self.assertEqual(403, r.code)
+            celery.control.pool_shrink.assert_not_called()
+
     def test_pool_autoscale(self):
         celery = self._app.capp
         celery.control.broadcast = MagicMock(return_value=[{'test': 'ok'}])
@@ -70,6 +101,14 @@ def test_pool_autoscale(self):
             'autoscale',
             reply=True, destination=['test'],
             arguments={'min': 2, 'max': 5})
+    def test_pool_autoscale_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.broadcast = MagicMock()
+            r = self.post('/api/worker/pool/autoscale/test',
+                          body={'min': 2, 'max': 5})
+            self.assertEqual(403, r.code)
+            celery.control.broadcast.assert_not_called()
 
     def test_add_consumer(self):
         celery = self._app.capp
@@ -83,6 +122,15 @@ def test_add_consumer(self):
             reply=True, destination=['test'],
             arguments={'queue': 'foo'})
 
+    def test_add_consumer_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.broadcast = MagicMock()
+            r = self.post('/api/worker/queue/add-consumer/test',
+                          body={'queue': 'foo'})
+            self.assertEqual(403, r.code)
+            celery.control.broadcast.assert_not_called()
+
     def test_cancel_consumer(self):
         celery = self._app.capp
         celery.control.broadcast = MagicMock(
@@ -95,6 +143,15 @@ def test_cancel_consumer(self):
             reply=True, destination=['test'],
             arguments={'queue': 'foo'})
 
+    def test_cancel_consumer_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.broadcast = MagicMock()
+            r = self.post('/api/worker/queue/cancel-consumer/test',
+                          body={'queue': 'foo'})
+            self.assertEqual(403, r.code)
+            celery.control.broadcast.assert_not_called()
+
     def test_task_timeout(self):
         celery = self._app.capp
         celery.control.time_limit = MagicMock(
@@ -109,6 +166,14 @@ def test_task_timeout(self):
             'celery.map', hard=3.1, soft=1.2, destination=['foo'],
             reply=True)
 
+    def test_task_timeout_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.time_limit = MagicMock()
+            r = self.post('/api/task/timeout/celery.map',
+                          body={'workername': 'foo', 'hard': 3.1, 'soft': 1.2})
+            self.assertEqual(403, r.code)
+            celery.control.time_limit.assert_not_called()
     def test_task_timeout_failure_returns_worker_error_message(self):
         celery = self._app.capp
         celery.control.time_limit = MagicMock(
@@ -132,6 +197,15 @@ def test_task_ratelimit(self):
         celery.control.rate_limit.assert_called_once_with(
             'celery.map', '20', destination=['foo'], reply=True)
 
+    def test_task_ratelimit_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.rate_limit = MagicMock()
+            r = self.post('/api/task/rate-limit/celery.map',
+                          body={'workername': 'foo', 'ratelimit': 20})
+            self.assertEqual(403, r.code)
+            celery.control.rate_limit.assert_not_called()
+
     def test_task_ratelimit_non_integer(self):
         celery = self._app.capp
         celery.control.rate_limit = MagicMock(
@@ -176,6 +250,14 @@ def test_revoke(self):
                                                       terminate=False,
                                                       signal='SIGTERM')
 
+    def test_revoke_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.revoke = MagicMock()
+            r = self.post('/api/task/revoke/test', body={})
+            self.assertEqual(403, r.code)
+            celery.control.revoke.assert_not_called()
+
     def test_terminate(self):
         celery = self._app.capp
         celery.control.revoke = MagicMock()
@@ -185,6 +267,14 @@ def test_terminate(self):
                                                       terminate=True,
                                                       signal='SIGTERM')
 
+    def test_terminate_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.control.revoke = MagicMock()
+            r = self.post('/api/task/revoke/test', body={'terminate': True})
+            self.assertEqual(403, r.code)
+            celery.control.revoke.assert_not_called()
+
     def test_terminate_signal(self):
         celery = self._app.capp
         celery.control.revoke = MagicMock()
diff --git a/tests/unit/api/test_tasks.py b/tests/unit/api/test_tasks.py
index 551957d7..66da4c2a 100644
--- a/tests/unit/api/test_tasks.py
+++ b/tests/unit/api/test_tasks.py
@@ -2,11 +2,12 @@
 import time
 from collections import OrderedDict
 from datetime import datetime, timedelta
-from unittest.mock import Mock, PropertyMock, patch
+from unittest.mock import MagicMock, Mock, PropertyMock, patch
 
 import celery.states as states
 from celery.events import Event
 from celery.result import AsyncResult
+from tornado.options import options
 
 from flower.events import EventsState
 from tests.unit.utils import task_succeeded_events
@@ -35,6 +36,15 @@ def test_apply(self):
         self.assertEqual(result, json.loads(body)['result'])
         task.apply_async.assert_called_once_with(args=[], kwargs={})
 
+    def test_apply_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.tasks['foo'] = Mock()
+            celery.tasks['foo'].apply_async = MagicMock()
+            r = self.post('/api/task/apply/foo', body='')
+            self.assertEqual(403, r.code)
+            celery.tasks['foo'].apply_async.assert_not_called()
+
 
 class AsyncApplyTests(BaseApiTestCase):
     def test_async_apply(self):
@@ -87,6 +97,15 @@ def test_async_apply_expires_datetime(self):
         task.apply_async.assert_called_once_with(
             args=[], kwargs={}, expires=tomorrow)
 
+    def test_async_apply_read_only(self):
+        with patch.object(options.mockable(), 'read_only', True):
+            celery = self._app.capp
+            celery.tasks['foo'] = Mock()
+            celery.tasks['foo'].apply_async = MagicMock()
+            r = self.post('/api/task/async-apply/foo', body={})
+            self.assertEqual(403, r.code)
+            celery.tasks['foo'].apply_async.assert_not_called()
+
 
 class MockTasks: