diff --git a/reviewboard/__init__.py b/reviewboard/__init__.py
index 3536685d301f38057a98f20a723f20fed073fafa..67fea28f592f6aeb5339bc1070eff53de1d67152 100644
--- a/reviewboard/__init__.py
+++ b/reviewboard/__init__.py
@@ -101,6 +101,7 @@ def initialize(
     load_extensions: bool = True,
     setup_logging: bool = True,
     setup_templates: bool = True,
+    setup_task_runner: bool = True,
 ) -> None:
     """Begin initialization of Review Board.
 
@@ -112,6 +113,10 @@ def initialize(
     will be called automatically in a standard install. If you are writing
     an extension or management command, you do not need to call this yourself.
 
+    Version Changed:
+        TBD:
+        Added the ``setup_task_runner`` argument.
+
     Args:
         load_extensions (bool, optional):
             Whether extensions should be automatically loaded upon
@@ -131,6 +136,12 @@ def initialize(
 
             Keep in mind that many pieces of functionality, such as avatars
             and some management commands, may be impacted by this setting.
+
+        setup_task_runner (bool, optional):
+            Whether to set up Celery for task running.
+
+            Version Added:
+                TBD
     """
     import logging
     import os
@@ -166,12 +177,14 @@ def initialize(
         # Set up logging.
         log.init_logging()
 
+    logger = logging.getLogger(__name__)
+
     load_site_config()
 
     if (setup_templates or load_extensions) and not is_running_test:
         if settings.DEBUG:
-            logging.debug("Log file for Review Board v%s (PID %s)" %
-                          (get_version_string(), os.getpid()))
+            logger.debug('Log file for Review Board v%s (PID %s)',
+                         get_version_string(), os.getpid())
 
         # Generate the AJAX serial, used for AJAX request caching.
         generate_ajax_serial()
@@ -202,9 +215,14 @@ def initialize(
                 # attempt to load any extensions yet.
                 pass
         else:
-            logging.warning('Extensions will not be loaded. The site must '
-                            'be upgraded from Review Board %s to %s.',
-                            siteconfig.version, installed_version)
+            logger.warning('Extensions will not be loaded. The site must '
+                           'be upgraded from Review Board %s to %s.',
+                           siteconfig.version, installed_version)
+
+    if setup_task_runner and not is_running_test:
+        from reviewboard.celery import setup_for_webserver
+
+        setup_for_webserver(siteconfig)
 
     signals.initializing.send(sender=None)
 
diff --git a/reviewboard/admin/forms/tasks_settings.py b/reviewboard/admin/forms/tasks_settings.py
new file mode 100644
index 0000000000000000000000000000000000000000..3d2c3d11c0f1d95d3b683e5eda53058221045e64
--- /dev/null
+++ b/reviewboard/admin/forms/tasks_settings.py
@@ -0,0 +1,135 @@
+"""Administration form for asynchronous tasks.
+
+Version Added:
+    TBD
+"""
+
+from __future__ import annotations
+
+from django import forms
+from django.utils.translation import gettext_lazy as _
+from djblets.siteconfig.forms import SiteSettingsForm
+
+from reviewboard.admin.siteconfig import load_site_config
+from reviewboard.celery import BrokerType
+
+
+class AMQPTasksSettingsForm(SiteSettingsForm):
+    """Settings subform for AMQP-based task brokers.
+
+    Version Added:
+        TBD
+    """
+
+    task_broker_amqp_url = forms.CharField(
+        required=True,
+        label=_('Broker URL'),
+        help_text=_(
+            'The URL to the AMQP broker to use.'
+            # TODO: link to docs?
+        ))
+
+    class Meta:
+        """Metadata for the form."""
+
+        title = _('AMQP Broker Settings')
+        fieldsets = (
+            (None, {
+                'classes': ('wide', 'hidden'),
+                'fields': ['task_broker_amqp_url'],
+            }),
+        )
+
+
+class FilesystemTasksSettingsForm(SiteSettingsForm):
+    """Settings subform for filesystem-basked task brokers.
+
+    Version Added:
+        TBD
+    """
+
+    task_broker_fs_location = forms.CharField(
+        label=_('Storage directory'),
+        help_text=_(
+            'The directory to use for communication between the web server '
+            'and the worker. This must be writable by the web server.'
+        ),
+        required=True,
+        widget=forms.TextInput(attrs={'size': '60'}),
+    )
+
+    class Meta:
+        """Metadata for the form."""
+
+        title = _('Filesystem Broker Settings')
+        fieldsets = (
+            (None, {
+                'classes': ('wide', 'hidden'),
+                'fields': ['task_broker_fs_location'],
+            }),
+        )
+
+
+class TasksSettingsForm(SiteSettingsForm):
+    """Settings form for asynchronous tasks.
+
+    Version Added:
+        TBD
+    """
+
+    task_worker_enabled = forms.BooleanField(
+        required=False,
+        label=_('Use asynchronous task runner'),
+        help_text=_(
+            'Handle certain tasks in a worker process. This requires '
+            'running the Review Board worker.'
+        ))
+
+    task_broker_type = forms.ChoiceField(
+        label=_('Broker type'),
+        help_text=_('Type of broker to use for asynchronous tasks.'),
+        choices=BrokerType.choices)
+
+    def __init__(self, *args, **kwargs) -> None:
+        """Initialize the form.
+
+        Args:
+            *args (tuple):
+                Positional arguments for the parent class.
+
+            **kwargs (dict):
+                Keyword arguments for the parent class.
+        """
+        super().__init__(*args, **kwargs)
+
+        self.broker_backend_forms = {
+            'amqp': AMQPTasksSettingsForm(*args, **kwargs),
+            'filesystem': FilesystemTasksSettingsForm(*args, **kwargs)
+        }
+
+    def save(self) -> None:
+        """Save the form.
+
+        This will write the new configuration to the database. It will then
+        force a site configuration reload.
+        """
+        super().save()
+
+        broker_type = self.cleaned_data['task_broker_type']
+
+        if broker_type in self.broker_backend_forms:
+            backend_form = self.broker_backend_forms[broker_type]
+            backend_form.save()
+
+        load_site_config()
+
+    class Meta:
+        """Metadata for the form."""
+
+        title = _('Task Runner Settings')
+        subforms = (
+            {
+                'subforms_attr': 'broker_backend_forms',
+                'controller_field': 'task_broker_type',
+            },
+        )
diff --git a/reviewboard/admin/siteconfig.py b/reviewboard/admin/siteconfig.py
index 3bf28895184640acd93a97f130cfe2da8cadac33..c651b99ebfb16c55451be43b2b654a66890a7a62 100644
--- a/reviewboard/admin/siteconfig.py
+++ b/reviewboard/admin/siteconfig.py
@@ -5,7 +5,7 @@ from __future__ import annotations
 import logging
 import os
 import re
-from typing import Optional, cast
+from typing import Optional, TYPE_CHECKING, cast
 
 from django.conf import settings, global_settings
 from django.core.exceptions import ImproperlyConfigured
@@ -30,22 +30,27 @@ from reviewboard.notifications.email.message import EmailMessage
 from reviewboard.search.search_backends.whoosh import WhooshBackend
 from reviewboard.signals import site_settings_loaded
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+    from djblets.siteconfig.django_settings import SiteConfigurationSettingsMap
+
 
 # A mapping of our supported storage backend names to backend class paths.
-storage_backend_map = {
+storage_backend_map: Mapping[str, str] = {
     'builtin': 'django.core.files.storage.FileSystemStorage',
     's3':      'storages.backends.s3.S3Storage',
     'swift':   'swift.storage.SwiftStorage',
 }
 
 
-log_settings_map = log_siteconfig.settings_map
+log_settings_map: SiteConfigurationSettingsMap = log_siteconfig.settings_map
 log_settings_defaults = log_siteconfig.defaults
 
 
 # A mapping of siteconfig setting names to Django settings.py names.
 # This also contains all the djblets-provided mappings as well.
-settings_map = {
+settings_map: SiteConfigurationSettingsMap = {
     'auth_digest_file_location':       'DIGEST_FILE_LOCATION',
     'auth_digest_realm':               'DIGEST_REALM',
     'auth_ldap_anon_bind_uid':         'LDAP_ANON_BIND_UID',
@@ -175,6 +180,12 @@ defaults.update({
     'search_backend_settings': {},
     'search_on_the_fly_indexing': False,
 
+    # Task runner settings.
+    'task_worker_enabled': False,
+    'task_broker_type': 'filesystem',
+    'task_broker_fs_location': os.path.join(settings.SITE_DATA_DIR,
+                                            'task-broker'),
+
     # Overwrite this.
     'site_media_url': f'{settings.SITE_ROOT}media/',
 })
diff --git a/reviewboard/admin/urls.py b/reviewboard/admin/urls.py
index d4714563e1c14bfd39fd139aaba95e5dddb3f875..11033834cfe8c7331f16cd52d2f8887ddd8c56eb 100644
--- a/reviewboard/admin/urls.py
+++ b/reviewboard/admin/urls.py
@@ -1,28 +1,6 @@
-#
-# reviewboard/admin/urls.py -- URLs for the admin app
-#
-# Copyright (c) 2008-2009  Christian Hammond
-# Copyright (c) 2009  David Trowbridge
-#
-# Permission is hereby granted, free of charge, to any person obtaining
-# a copy of this software and associated documentation files (the
-# "Software"), to deal in the Software without restriction, including
-# without limitation the rights to use, copy, modify, merge, publish,
-# distribute, sublicense, and/or sell copies of the Software, and to
-# permit persons to whom the Software is furnished to do so, subject to
-# the following conditions:
-#
-# The above copyright notice and this permission notice shall be included
-# in all copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
-# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
-# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
-# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
-# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
-# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
-# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
-#
+"""Admin URLs."""
+
+from __future__ import annotations
 
 from django.urls import include, path
 from django.views.generic import RedirectView
@@ -39,6 +17,7 @@ from reviewboard.admin.forms.review_settings import ReviewSettingsForm
 from reviewboard.admin.forms.search_settings import SearchSettingsForm
 from reviewboard.admin.forms.storage_settings import StorageSettingsForm
 from reviewboard.admin.forms.support_settings import SupportSettingsForm
+from reviewboard.admin.forms.tasks_settings import TasksSettingsForm
 
 
 urlpatterns = [
@@ -139,6 +118,13 @@ urlpatterns = [
                  'form_class': SearchSettingsForm,
              },
              name='settings-search'),
+
+        path('tasks/',
+             views.site_settings,
+             kwargs={
+             'form_class': TasksSettingsForm,
+             },
+             name='settings-tasks'),
     ])),
 
     path('widget-activity/', views.widget_activity),
diff --git a/reviewboard/celery.py b/reviewboard/celery.py
new file mode 100644
index 0000000000000000000000000000000000000000..626b71f803eae204ed316b8b9611e52c1abd391b
--- /dev/null
+++ b/reviewboard/celery.py
@@ -0,0 +1,184 @@
+"""Celery definitions for asynchronous tasks.
+
+Version Added:
+    TBD
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import TYPE_CHECKING, cast
+
+from celery import Celery
+from celery.signals import worker_init
+from django.conf import settings
+from django.db.models import TextChoices
+from django.utils.translation import gettext_lazy as _
+
+from reviewboard import get_version_string
+
+if TYPE_CHECKING:
+    from typing import Any
+
+    from celery.apps.worker import Worker
+    from djblets.siteconfig.models import SiteConfiguration
+
+
+logger = logging.getLogger(__name__)
+
+
+#: The main Celery instance.
+#:
+#: Version Added:
+#:     TBD
+celery = Celery('reviewboard')
+
+
+class BrokerType(TextChoices):
+    """Choices for the broker type field."""
+
+    FILESYSTEM = 'filesystem', _('Filesystem')
+    AMQP = 'amqp', _('AMQP')
+
+
+def setup_for_webserver(
+    siteconfig: SiteConfiguration,
+) -> None:
+    """Set up Celery when running inside the webserver context.
+
+    Args:
+        siteconfig (djblets.siteconfig.models.SiteConfiguration):
+            The site config object.
+    """
+    logger.debug('Setting up Review Board Celery.')
+
+    _load_config(siteconfig)
+    celery.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
+
+    from djblets.siteconfig.signals import siteconfig_reloaded
+
+    siteconfig_reloaded.connect(_on_siteconfig_reloaded)
+
+
+@worker_init.connect
+def setup_for_worker(
+    sender: Worker,
+    **kwargs,
+) -> None:
+    """Set up Celery when running as a worker.
+
+    Args:
+        sender (celery.apps.worker.Worker):
+            The worker instance.
+
+        **kwargs (dict, unused):
+            Unused keyword arguments.
+    """
+    logger.debug('Setting up environment for Celery worker.')
+    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'reviewboard.settings')
+
+    from django import setup
+    from django.apps import apps
+
+    if not apps.ready:
+        setup()
+
+    from djblets.siteconfig.models import SiteConfiguration
+    from reviewboard.admin.siteconfig import load_site_config
+    from reviewboard.extensions.base import get_extension_manager
+
+    logger.debug('Loading Review Board siteconfig')
+    load_site_config()
+
+    siteconfig = SiteConfiguration.objects.get_current()
+    installed_version = get_version_string()
+
+    if siteconfig.version != installed_version:
+        raise Exception(
+            f'Review Board must be upgraded from {installed_version} to '
+            f'{siteconfig.version} before the Celery worker can run.')
+
+    _load_config(siteconfig)
+    logger.debug('Autodiscovering tasks')
+    celery.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
+
+    logger.debug('Loading Review Board extensions')
+    get_extension_manager().load()
+
+
+def _on_siteconfig_reloaded(
+    sender: None,
+    siteconfig: SiteConfiguration,
+    old_siteconfig: SiteConfiguration,
+    **kwargs,
+) -> None:
+    """Handler for when the siteconfig is reloaded.
+
+    Args:
+        sender (None):
+            The signal sender.
+
+        siteconfig (djblets.siteconfig.models.SiteConfiguration):
+            The new siteconfig instance.
+
+        old_siteconfig (djblets.siteconfig.models.SiteConfiguration):
+            The old siteconfig instance.
+
+        **kwargs (dict, unused):
+            Unused keyword arguments.
+    """
+    logger.info('Updating Celery config after siteconfig reload')
+    _load_config(siteconfig)
+
+
+def _load_config(
+    siteconfig: SiteConfiguration,
+) -> None:
+    """Load the celery configuration from siteconfig.
+
+    Args:
+        siteconfig (djblets.siteconfig.models.SiteConfiguration):
+            The site config object.
+    """
+    broker_type = siteconfig.get('task_broker_type')
+
+    config: dict[str, Any] = {}
+
+    if broker_type == BrokerType.FILESYSTEM:
+        fs_location = siteconfig.get('task_broker_fs_location')
+
+        if not isinstance(fs_location, str):
+            fs_location = os.path.join(settings.SITE_DATA_DIR, 'task-broker')
+
+        if not os.path.exists(fs_location):
+            try:
+                os.mkdir(fs_location)
+            except OSError as e:
+                logger.exception(
+                    'Filesystem broker location %s does not exist. Attempted '
+                    'creation failed: %s',
+                    fs_location, e)
+
+                return
+        elif not os.path.isdir(fs_location):
+            logger.error(
+                'Filesystem broker location %s exists but is not a directory.',
+                fs_location)
+
+            return
+
+        config['broker_url'] = 'filesystem://'
+        config['broker_transport_options'] = {
+            'data_folder_in': fs_location,
+            'data_folder_out': fs_location,
+        }
+    elif broker_type == BrokerType.AMQP:
+        config['broker_url'] = siteconfig.get('task_broker_amqp_url')
+    else:
+        logger.error('Invalid broker type "%s" in siteconfig',
+                     broker_type)
+
+        return
+
+    celery.conf.update(**config)
diff --git a/reviewboard/dependencies.py b/reviewboard/dependencies.py
index f25245e2a719791f6e7532e3d5f1399bb5916fd2..ef71784e38d41f20bd27be9ca83af3488e5da611 100644
--- a/reviewboard/dependencies.py
+++ b/reviewboard/dependencies.py
@@ -51,6 +51,7 @@ djblets_version = '~=6.0a0.dev0'
 #: All dependencies required to install Review Board.
 package_dependencies = {
     'bleach': '~=6.0.0',
+    'celery': '~=5.5.0',
     'cryptography': '~=41.0.7',
     'Django': django_version,
     'django-cors-headers': '~=3.11.0',
diff --git a/reviewboard/extensions/base.py b/reviewboard/extensions/base.py
index 15b21c251f13bf0dc6479bb81efaf90a871c5420..cb215fa706c334dc06637542e1483f43b22679bf 100644
--- a/reviewboard/extensions/base.py
+++ b/reviewboard/extensions/base.py
@@ -1,10 +1,17 @@
 """Base support for writing custom extensions."""
 
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from djblets.extensions.extension import (Extension as DjbletsExtension,
                                           JSExtension as DjbletsJSExtension)
 from djblets.extensions.manager import (ExtensionManager as
                                         DjbletsExtensionManager)
 
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
 
 class Extension(DjbletsExtension):
     """Base class for custom extensions.
@@ -34,13 +41,48 @@ class ExtensionManager(DjbletsExtensionManager):
     See the Djblets :py:class:`~djblets.extensions.manager.ExtensionManager`
     documentation for a full class reference.
     """
-    pass
+
+    def _add_to_installed_apps(
+        self,
+        extension: Extension,
+    ) -> Sequence[str]:
+        """Add an extension's apps to the installed apps.
+
+        This extends the base class implementation to also autodiscover any
+        Celery tasks within the extension app.
+
+        Returns:
+            list of str:
+            A list of the newly-added apps.
+        """
+        new_apps = super()._add_to_installed_apps(extension)
+
+        from reviewboard.celery import celery
+        celery.autodiscover_tasks(new_apps, force=True)
+
+        return new_apps
+
+    def _remove_from_installed_apps(
+        self,
+        extension: Extension,
+    ) -> None:
+        """Remove an extension's apps from the installed apps.
+
+        This extends the base class implementation to remove any registered
+        Celery tasks.
+        """
+        apps = extension.apps or [extension.info.app_name]
+
+        from reviewboard.celery import celery
+        celery.loader.task_modules.difference_update(apps)
+
+        return super()._remove_from_installed_apps(extension)
 
 
 _extension_manager = None
 
 
-def get_extension_manager():
+def get_extension_manager() -> ExtensionManager:
     """Return the extension manager used by Review Board.
 
     The same instance will be returned every time.
diff --git a/reviewboard/settings.py b/reviewboard/settings.py
index 5171668f507cc4c4ef6d297b7c188479d1a61d71..6271aadecf8cbcf32368b3a68b5acd6e958ce9e7 100644
--- a/reviewboard/settings.py
+++ b/reviewboard/settings.py
@@ -694,4 +694,8 @@ CUSTOM_PYGMENTS_LEXERS = {
 }
 
 
+# Celery settings.
+CELERY_TASK_EAGER_PROPAGATES = True
+
+
 fail_if_missing_dependencies()
diff --git a/reviewboard/tasks.py b/reviewboard/tasks.py
new file mode 100644
index 0000000000000000000000000000000000000000..4667749603409aacc667507c5c15f2574b61ce18
--- /dev/null
+++ b/reviewboard/tasks.py
@@ -0,0 +1,320 @@
+"""Base definitions for tasks.
+
+Version Added:
+    TBD
+"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from dataclasses import dataclass, replace
+from typing import Generic, TYPE_CHECKING, cast
+
+from celery import Celery, Task as CeleryTask, shared_task
+from django.conf import settings
+from pydantic import BaseModel
+from typing_extensions import ParamSpec, Self, TypeVar
+
+from reviewboard.celery import celery
+
+if TYPE_CHECKING:
+    from typing import Any, Callable
+
+    from celery.result import AsyncResult
+
+
+_P = ParamSpec('_P')
+
+
+class Task(CeleryTask, Generic[_P]):
+    """Base class for tasks.
+
+    This provides a layer on top of celery.Task to improve a couple things:
+
+    * Tasks which take Pydantic models as arguments will automatically
+      serialize the data when calling the task (whereas celery.Task requires
+      the caller to call :py:meth:`~pydantic.BaseModel.model_dump()`.
+    * This provides a ``django-tasks``-like ``.using().enqueue()`` API. This
+      allows us to override task options at call time without losing all the
+      type data for args/kwargs, and additionally allows overriding the celery
+      app instance, which is not possible with vanilla celery.
+
+    Version Added:
+        TBD
+    """
+
+    #: The queue to use for the task.
+    queue: str | None
+
+    #: The celery instance to use.
+    _app: (Celery | None) = None
+
+    def _get_app(self) -> Celery:
+        """Return the celery app.
+
+        Returns:
+            celery.Celery:
+            The celery app to use for the task.
+        """
+        if self._app:
+            return self._app
+        else:
+            return cast(Celery, super()._get_app())
+
+    def using(
+        self,
+        app: (Celery | None) = None,
+        queue: (str | None) = None,
+    ) -> Self:
+        """Return a new version of the task with different defaults.
+
+        Args:
+            app (celery.Celery, optional):
+                The celery instance to invoke the task on.
+
+            queue (str, optional):
+                The queue to use.
+
+        Returns:
+            Task:
+            A copy of the task with different defaults.
+        """
+        changes: dict[str, Any] = {}
+
+        if app is not None:
+            changes['app'] = app
+
+        if queue is not None:
+            changes['queue'] = queue
+
+        copy = deepcopy(self)
+
+        if app is not None:
+            copy.app = app
+
+        if queue is not None:
+            copy.queue = queue
+
+        return copy
+
+    def enqueue(
+        self,
+        *args: _P.args,
+        **kwargs: _P.kwargs,
+    ) -> None:
+        """Queue the task.
+
+        This will queue up the task, as well as dump any Pydantic models into a
+        serializable form.
+
+        Args:
+            *args (tuple):
+                Positional arguments to pass through to the parent class.
+
+            **kwargs (dict):
+                Keyword arguments to send to the task.
+        """
+        self.apply_async(
+            tuple(
+                _maybe_serialize_model(arg)
+                for arg in args
+            ),
+            {
+                key: _maybe_serialize_model(value)
+                for key, value in kwargs.items()
+            },
+            queue=self.queue)
+
+
+class MaybeEagerTask(Task[_P]):
+    """Base class for tasks that should (maybe) run eager in the devserver.
+
+    This should be used with @task(base=MaybeEagerTask).
+
+    Version Added:
+        TBD
+    """
+
+    def apply_async(self, *args, **kwargs) -> AsyncResult:
+        """Run a task asynchronously.
+
+        Args:
+            *args (tuple):
+                Positional arguments to pass through to the parent class.
+
+            **kwargs (dict):
+                Keyword arguments to pass through to the parent class.
+        """
+        if not settings.PRODUCTION:
+            # If we're not running in production, we may want to run tasks
+            # eagerly. Unfortunately, Celery has no way of doing this on a
+            # task-by-task basis, only changing the app config.
+            celery = self._get_app()
+            celery.conf.task_always_eager = True
+
+            try:
+                return super().apply_async(*args, **kwargs)
+            finally:
+                celery.conf.task_always_eager = False
+        else:
+            return super().apply_async(*args, **kwargs)
+
+
+@dataclass(frozen=True)
+class ExternalTask(Generic[_P]):
+    """An object for external tasks that works similar to Task.
+
+    Version Added:
+        TBD
+    """
+
+    ######################
+    # Instance variables #
+    ######################
+
+    #: The task's name.
+    name: str
+
+    #: The queue to use for the task.
+    queue: str | None
+
+    #: The celery instance to use.
+    app: Celery | None = None
+
+    def using(
+        self,
+        app: (Celery | None) = None,
+        queue: str | None = None,
+    ) -> Self:
+        """Return a new version of the task with different defaults.
+
+        Args:
+            app (celery.Celery, optional):
+                The celery instance to invoke the task on.
+
+            queue (str, optional):
+                The queue to use.
+
+        Returns:
+            ExternalTask:
+            A copy of the task with different defaults.
+        """
+        changes: dict[str, Any] = {}
+
+        if app is not None:
+            changes['app'] = app
+
+        if queue is not None:
+            changes['queue'] = queue
+
+        return replace(self, **changes)
+
+    def enqueue(
+        self,
+        *args: _P.args,
+        **kwargs: _P.kwargs,
+    ) -> AsyncResult:
+        """Enqueue the task.
+
+        Args:
+            *args (tuple):
+                Positional arguments for the task.
+
+            **kwargs (dict):
+                Keyword arguments for the task.
+        """
+        app = self.app or celery
+
+        return app.send_task(
+            args=tuple(
+                _maybe_serialize_model(arg)
+                for arg in args
+            ),
+            kwargs={
+                key: _maybe_serialize_model(value)
+                for key, value in kwargs.items()
+            },
+            name=self.name,
+            queue=self.queue,
+        )
+
+
+def external_task(
+    name: str,
+    queue: (str | None) = None,
+) -> Callable[[Callable[_P, None]], ExternalTask[_P]]:
+    """Decorate a stub to create an external task.
+
+    This will create an object that works similar to celery.Task, but for tasks
+    which are implemented completely external to the current codebase.
+
+    Version Added:
+        TBD
+    """
+    def wrapper(
+        f: Callable[_P, None],
+    ) -> ExternalTask[_P]:
+        return ExternalTask[_P](name, queue)
+
+    return wrapper
+
+
+_TaskBaseT = TypeVar('_TaskBaseT', default=Task[_P])
+
+
+def task(
+    *args,
+    base: type[Task[_P]] = Task[_P],
+    **kwargs,
+) -> Callable[[Callable[_P, None]], Task[_P]]:
+    """Decorate a method to create a local task.
+
+    This will create a task for a method that is available in the context of
+    the webserver. If the kwargs include base=MaybeEagerTask, this task can be
+    used within the devserver without requiring a worker.
+
+    Args:
+        *args (tuple):
+            Positional arguments to pass to the :py:meth:`~celery.shared_task`
+            function.
+
+        base (type, optional):
+            The type to use for the task class instance.
+
+        **kwargs (dict):
+            Keyword arguments to pass to the :py:meth:`~celery.shared_task`
+            function.
+
+    Version Added:
+        TBD
+    """
+    def wrapper(
+        f: Callable[_P, None],
+    ) -> Task[_P]:
+        return cast(Task[_P], shared_task(*args, base=base, **kwargs))
+
+    return wrapper
+
+
+def _maybe_serialize_model(
+    data: Any,
+) -> Any:
+    """Possibly convert a Pydantic model to data.
+
+    Celery tasks currently support Pydantic deserialization on the task side,
+    but require the task invocation to pass in dict-like data. This method can
+    be applied to args/kwargs for a task to handle the conversion.
+    """
+    if isinstance(data, BaseModel):
+        return data.model_dump()
+    else:
+        return data
+
+
+__all__ = [
+    'ExternalTask',
+    'MaybeEagerTask',
+    'Task',
+    'external_task',
+    'task',
+]
diff --git a/reviewboard/templates/admin/sidebar.html b/reviewboard/templates/admin/sidebar.html
index 6de551c3d527e9f53c6a909b55a6028a51f1d08e..55f25c65deec980db7c9ddf5a61ff613f271ba0e 100644
--- a/reviewboard/templates/admin/sidebar.html
+++ b/reviewboard/templates/admin/sidebar.html
@@ -16,6 +16,7 @@
  <ul class="rb-c-sidebar__items">
 {% admin_subnav "settings-general" _("General") %}
 {% admin_subnav "settings-authentication" _("Authentication") %}
+{% admin_subnav "settings-tasks" _("Task Runner") %}
 {% admin_subnav "settings-avatars" _("Avatars") %}
 {% admin_subnav "settings-email" _("E-Mail") %}
 {% admin_subnav "settings-reviews" _("Review Workflow") %}
