diff --git a/kobo/apps/audit_log/tests/api/v2/test_api_audit_log.py b/kobo/apps/audit_log/tests/api/v2/test_api_audit_log.py
index aa6d3f3076..3e4a9d7654 100644
--- a/kobo/apps/audit_log/tests/api/v2/test_api_audit_log.py
+++ b/kobo/apps/audit_log/tests/api/v2/test_api_audit_log.py
@@ -6,17 +6,14 @@
 from rest_framework.reverse import reverse
 
 from kobo.apps.audit_log.audit_actions import AuditAction
-from kobo.apps.audit_log.models import (
-    AccessLog,
-    AuditLog,
-    AuditType,
-)
+from kobo.apps.audit_log.models import AccessLog, AuditLog, AuditType
 from kobo.apps.audit_log.tests.test_signals import skip_login_access_log
 from kobo.apps.kobo_auth.shortcuts import User
 from kpi.constants import (
     ACCESS_LOG_SUBMISSION_AUTH_TYPE,
     ACCESS_LOG_SUBMISSION_GROUP_AUTH_TYPE,
 )
+from kpi.models.import_export_task import AccessLogExportTask
 from kpi.tests.base_test_case import BaseTestCase
 from kpi.urls.router_api_v2 import URL_NAMESPACE as ROUTER_URL_NAMESPACE
 
@@ -430,3 +427,180 @@ def test_can_search_access_logs_by_date_including_submission_groups(self):
             group['metadata']['auth_type'],
             ACCESS_LOG_SUBMISSION_GROUP_AUTH_TYPE,
         )
+
+
+class ApiAccessLogsExportTestCase(BaseAuditLogTestCase):
+
+    def get_endpoint_basename(self):
+        return 'access-logs-export-list'
+
+    def test_export_as_anonymous_returns_unauthorized(self):
+        self.client.logout()
+        response = self.client.post(self.url)
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+    def test_export_for_user_returns_success(self):
+        self.force_login_user(User.objects.get(username='anotheruser'))
+        response = self.client.post(self.url)
+        self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+
+    def test_export_for_superuser_commences(self):
+        self.force_login_user(User.objects.get(username='admin'))
+        response = self.client.post(self.url)
+        self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+
+    def test_create_export_task_on_post(self):
+        test_user = User.objects.get(username='anotheruser')
+        self.force_login_user(test_user)
+
+        response = self.client.post(self.url)
+        self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+
+        task = (
+            AccessLogExportTask.objects.filter(user=test_user)
+            .order_by('-date_created')
+            .first()
+        )
+        self.assertIsNotNone(task)
+        self.assertIn(task.status, ['created', 'processing', 'complete'])
+        self.assertFalse(task.get_all_logs)
+
+    def test_get_status_of_tasks(self):
+        test_user = User.objects.get(username='anotheruser')
+        self.force_login_user(test_user)
+
+        AccessLogExportTask.objects.create(
+            user=test_user,
+            get_all_logs=False,
+            data={
+                'type': 'access_logs_export',
+            },
+        )
+
+        response_status = self.client.get(self.url)
+        self.assertEqual(response_status.status_code, status.HTTP_200_OK)
+
+        # Assert the response contains a list of tasks
+        tasks = response_status.json()
+        self.assertIsInstance(tasks, list)
+        self.assertGreater(len(tasks), 0)  # Ensure at least one task is present
+
+        # Assert the structure of the first task in the list
+        first_task = tasks[0]
+        self.assertIn('uid', first_task)
+        self.assertIn('status', first_task)
+        self.assertIn('date_created', first_task)
+
+    def test_multiple_export_tasks_not_allowed(self):
+        test_user = User.objects.get(username='anotheruser')
+        self.force_login_user(test_user)
+
+        response_first = self.client.post(self.url)
+        self.assertEqual(response_first.status_code, status.HTTP_202_ACCEPTED)
+
+        task = (
+            AccessLogExportTask.objects.filter(user=test_user)
+            .order_by('-date_created')
+            .first()
+        )
+        task.status = 'processing'
+        task.save()
+
+        response_second = self.client.post(self.url)
+        self.assertEqual(response_second.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn(
+            'Export task for user access logs already in progress.',
+            response_second.json()['error'],
+        )
+
+
+class AllApiAccessLogsExportTestCase(BaseAuditLogTestCase):
+
+    def get_endpoint_basename(self):
+        return 'all-access-logs-export-list'
+
+    def test_export_as_anonymous_returns_unauthorized(self):
+        self.client.logout()
+        response = self.client.post(self.url)
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+    def test_regular_user_cannot_export_access_logs(self):
+        self.force_login_user(User.objects.get(username='anotheruser'))
+        response = self.client.post(self.url)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_export_access_logs_for_superuser_returns_success(self):
+        self.force_login_user(User.objects.get(username='admin'))
+        response = self.client.post(self.url)
+        self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+
+    def test_superuser_create_export_task_on_post(self):
+        test_superuser = User.objects.get(username='admin')
+        self.force_login_user(test_superuser)
+
+        response = self.client.post(self.url)
+        self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+
+        task = (
+            AccessLogExportTask.objects.filter(user=test_superuser)
+            .order_by('-date_created')
+            .first()
+        )
+        self.assertIsNotNone(task)
+        self.assertIn(task.status, ['created', 'processing', 'complete'])
+        self.assertTrue(task.get_all_logs)
+
+    def test_superuser_get_status_tasks(self):
+        test_superuser = User.objects.get(username='admin')
+        self.force_login_user(test_superuser)
+
+        AccessLogExportTask.objects.create(
+            user=test_superuser,
+            get_all_logs=False,
+            data={
+                'type': 'access_logs_export',
+            },
+        )
+
+        response_status = self.client.get(self.url)
+        self.assertEqual(response_status.status_code, status.HTTP_200_OK)
+
+        # Assert the response contains a list of tasks
+        tasks = response_status.json()
+        self.assertIsInstance(tasks, list)
+        self.assertGreater(len(tasks), 0)  # Ensure at least one task is present
+
+        # Assert the structure of the first task in the list
+        first_task = tasks[0]
+        self.assertIn('uid', first_task)
+        self.assertIn('status', first_task)
+        self.assertIn('date_created', first_task)
+
+    def test_permission_denied_for_non_superusers_on_get_status(self):
+        non_superuser = User.objects.get(username='anotheruser')
+        self.force_login_user(non_superuser)
+
+        response = self.client.get(self.url)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_multiple_export_tasks_not_allowed(self):
+        test_superuser = User.objects.get(username='admin')
+        self.force_login_user(test_superuser)
+
+        response_first = self.client.post(self.url)
+        self.assertEqual(response_first.status_code, status.HTTP_202_ACCEPTED)
+
+        task = (
+            AccessLogExportTask.objects.filter(user=test_superuser)
+            .order_by('-date_created')
+            .first()
+        )
+        task.status = 'processing'
+        task.save()
+
+        response_second = self.client.post(self.url)
+        self.assertEqual(response_second.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn(
+            'Export task for all access logs already in progress.',
+            response_second.json()['error'],
+        )
diff --git a/kobo/apps/audit_log/urls.py b/kobo/apps/audit_log/urls.py
index 0d742a83e2..f2f9625b13 100644
--- a/kobo/apps/audit_log/urls.py
+++ b/kobo/apps/audit_log/urls.py
@@ -1,10 +1,22 @@
 from rest_framework.routers import DefaultRouter
 
-from .views import AccessLogViewSet, AllAccessLogViewSet, AuditLogViewSet
+from .views import (
+    AccessLogsExportViewSet,
+    AccessLogViewSet,
+    AllAccessLogsExportViewSet,
+    AllAccessLogViewSet,
+    AuditLogViewSet,
+)
 
 router = DefaultRouter()
 router.register(r'audit-logs', AuditLogViewSet, basename='audit-log')
 router.register(r'access-logs', AllAccessLogViewSet, basename='all-access-logs')
 router.register(r'access-logs/me', AccessLogViewSet, basename='access-log')
+router.register(
+    r'access-logs/export', AllAccessLogsExportViewSet, basename='all-access-logs-export'
+)
+router.register(
+    r'access-logs/me/export', AccessLogsExportViewSet, basename='access-logs-export'
+)
 
 urlpatterns = []
diff --git a/kobo/apps/audit_log/views.py b/kobo/apps/audit_log/views.py
index 94a15025e7..dce8deaf5a 100644
--- a/kobo/apps/audit_log/views.py
+++ b/kobo/apps/audit_log/views.py
@@ -1,8 +1,11 @@
-from rest_framework import mixins, viewsets
+from rest_framework import mixins, status, viewsets
 from rest_framework.renderers import BrowsableAPIRenderer, JSONRenderer
+from rest_framework.response import Response
 
 from kpi.filters import SearchFilter
+from kpi.models.import_export_task import AccessLogExportTask
 from kpi.permissions import IsAuthenticated
+from kpi.tasks import export_task_in_background
 from .filters import AccessLogPermissionsFilter
 from .models import AccessLog, AuditLog
 from .permissions import SuperUserPermission
@@ -321,3 +324,178 @@ class AccessLogViewSet(AuditLogViewSet):
     permission_classes = (IsAuthenticated,)
     filter_backends = (AccessLogPermissionsFilter,)
     serializer_class = AccessLogSerializer
+
+
+class BaseAccessLogsExportViewSet(viewsets.ViewSet):
+    permission_classes = (IsAuthenticated,)
+    lookup_field = 'uid'
+
+    def create_task(self, request, get_all_logs):
+
+        export_task = AccessLogExportTask.objects.create(
+            user=request.user,
+            get_all_logs=get_all_logs,
+            data={
+                'type': 'access_logs_export',
+            },
+        )
+
+        export_task_in_background.delay(
+            export_task_uid=export_task.uid,
+            username=export_task.user.username,
+            export_task_name='kpi.AccessLogExportTask',
+        )
+        return Response(
+            {f'status: {export_task.status}'},
+            status=status.HTTP_202_ACCEPTED,
+        )
+
+    def list_tasks(self, user=None):
+        tasks = AccessLogExportTask.objects.all()
+        if user is not None:
+            tasks = tasks.filter(user=user)
+        tasks = tasks.order_by('-date_created')
+
+        tasks_data = [
+            {'uid': task.uid, 'status': task.status, 'date_created': task.date_created}
+            for task in tasks
+        ]
+
+        return Response(tasks_data, status=status.HTTP_200_OK)
+
+
+class AccessLogsExportViewSet(BaseAccessLogsExportViewSet):
+    """
+    Access logs export
+
+    Lists all access logs export tasks for the authenticated user
+
+    <pre class="prettyprint">
+    <b>GET</b> /api/v2/access-logs/me/export
+    </pre>
+
+    > Example
+    >
+    >       curl -X GET https://[kpi-url]/access-logs/me/export
+
+    > Response 200
+    >
+    >       [
+    >           {
+    >               "uid": "aleooVUrhe3cRrLY5urRhxLA",
+    >               "status": "complete",
+    >               "date_created": "2024-11-26T21:27:08.403181Z"
+    >           },
+    >           {
+    >               "uid": "aleMzK7RnuaPokb86TZF2N4d",
+    >               "status": "complete",
+    >               "date_created": "2024-11-26T20:18:55.982974Z"
+    >           }
+    >       ]
+
+    ### Creates an export task
+
+    <pre class="prettyprint">
+    <b>POST</b> /api/v2/access-log/me/export
+    </pre>
+
+    > Example
+    >
+    >       curl -X POST https://[kpi-url]/access-logs/me/export
+
+    > Response 202
+    >
+    >       [
+    >           "status: created"
+    >       ]
+    >
+    """
+
+    def create(self, request, *args, **kwargs):
+        if AccessLogExportTask.objects.filter(
+            user=request.user,
+            status=AccessLogExportTask.PROCESSING,
+            get_all_logs=False,
+        ).exists():
+            return Response(
+                {
+                    'error': (
+                        'Export task for user access logs already in progress.'
+                    )
+                },
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        return self.create_task(request, get_all_logs=False)
+
+    def list(self, request, *args, **kwargs):
+        return self.list_tasks(request.user)
+
+
+class AllAccessLogsExportViewSet(BaseAccessLogsExportViewSet):
+    """
+    Access logs export
+
+    Lists all access logs export tasks for all users. Only available to superusers.
+
+    <pre class="prettyprint">
+    <b>GET</b> /api/v2/access-logs/export
+    </pre>
+
+    > Example
+    >
+    >       curl -X GET https://[kpi-url]/access-logs/export
+
+    > Response 200
+    >
+    >       [
+    >           {
+    >               "uid": "aleooVUrhe3cRrLY5urRhxLA",
+    >               "status": "complete",
+    >               "date_created": "2024-11-26T21:27:08.403181Z"
+    >           },
+    >           {
+    >               "uid": "aleMzK7RnuaPokb86TZF2N4d",
+    >               "status": "complete",
+    >               "date_created": "2024-11-26T20:18:55.982974Z"
+    >           }
+    >       ]
+
+    ### Creates an export task
+
+    <pre class="prettyprint">
+    <b>POST</b> /api/v2/access-log/export
+    </pre>
+
+    > Example
+    >
+    >       curl -X POST https://[kpi-url]/access-logs/export
+
+    > Response 202
+    >
+    >       [
+    >           "status: created"
+    >       ]
+    >
+    """
+
+    permission_classes = (SuperUserPermission,)
+
+    def create(self, request, *args, **kwargs):
+        # Check if the superuser has a task running for all
+        if AccessLogExportTask.objects.filter(
+            user=request.user,
+            status=AccessLogExportTask.PROCESSING,
+            get_all_logs=True,
+        ).exists():
+            return Response(
+                {
+                    'error': (
+                        'Export task for all access logs already in progress.'
+                    )
+                },
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        return self.create_task(request, get_all_logs=True)
+
+    def list(self, request, *args, **kwargs):
+        return self.list_tasks()
diff --git a/kpi/tests/test_export_tasks.py b/kpi/tests/test_export_tasks.py
index ce0896ed0e..e4bf673c5e 100644
--- a/kpi/tests/test_export_tasks.py
+++ b/kpi/tests/test_export_tasks.py
@@ -3,7 +3,6 @@
 from django.conf import settings
 from django.core.exceptions import ObjectDoesNotExist
 from django.test import TestCase
-from django.utils import timezone
 
 from kobo.apps.kobo_auth.shortcuts import User
 from kpi.models.import_export_task import ProjectViewExportTask
@@ -34,19 +33,12 @@ def test_export_task_success(self, mock_get_project_view, mock_send_mail):
         self.assertEqual(self.task.status, 'complete')
 
         root_url = settings.KOBOFORM_URL
+        expected_file_path = self.task.result.url
         expected_message = (
-            'Hello {},\n\n'
-            'Your report is complete: {}'
-            '/private-media/{}/exports/'
-            'assets-{}-view_summary-{}.csv\n\n'
-            'Regards,\n'
-            'KoboToolbox'
+            'Hello {},\n\n' 'Your report is complete: {}\n\n' 'Regards,\n' 'KoboToolbox'
         ).format(
             self.user.username,
-            root_url,
-            self.user.username,
-            self.user.username,
-            timezone.now().strftime('%Y-%m-%dT%H%M%SZ'),
+            f'{root_url}{expected_file_path}',
         )
         mock_send_mail.assert_called_once_with(
             subject='Project View Report Complete',