diff --git a/authentik/admin/templates/administration/flow/import.html b/authentik/admin/templates/administration/flow/import.html
deleted file mode 100644
index 14eca092a..000000000
--- a/authentik/admin/templates/administration/flow/import.html
+++ /dev/null
@@ -1,13 +0,0 @@
-{% extends base_template|default:"generic/form.html" %}
-
-{% load i18n %}
-
-{% block above_form %}
-<h1>
-{% trans 'Import Flow' %}
-</h1>
-{% endblock %}
-
-{% block action %}
-{% trans 'Import Flow' %}
-{% endblock %}
diff --git a/authentik/admin/urls.py b/authentik/admin/urls.py
index 03ae5b98f..6402e6efa 100644
--- a/authentik/admin/urls.py
+++ b/authentik/admin/urls.py
@@ -2,7 +2,6 @@
 from django.urls import path
 
 from authentik.admin.views import (
-    flows,
     outposts,
     outposts_service_connections,
     policies,
@@ -99,27 +98,6 @@ urlpatterns = [
         stages_invitations.InvitationCreateView.as_view(),
         name="stage-invitation-create",
     ),
-    # Flows
-    path(
-        "flows/create/",
-        flows.FlowCreateView.as_view(),
-        name="flow-create",
-    ),
-    path(
-        "flows/import/",
-        flows.FlowImportView.as_view(),
-        name="flow-import",
-    ),
-    path(
-        "flows/<uuid:pk>/update/",
-        flows.FlowUpdateView.as_view(),
-        name="flow-update",
-    ),
-    path(
-        "flows/<uuid:pk>/execute/",
-        flows.FlowDebugExecuteView.as_view(),
-        name="flow-execute",
-    ),
     # Property Mappings
     path(
         "property-mappings/create/",
diff --git a/authentik/admin/views/flows.py b/authentik/admin/views/flows.py
deleted file mode 100644
index 01d1b54fa..000000000
--- a/authentik/admin/views/flows.py
+++ /dev/null
@@ -1,108 +0,0 @@
-"""authentik Flow administration"""
-from django.contrib import messages
-from django.contrib.auth.mixins import LoginRequiredMixin
-from django.contrib.auth.mixins import (
-    PermissionRequiredMixin as DjangoPermissionRequiredMixin,
-)
-from django.contrib.messages.views import SuccessMessageMixin
-from django.http import HttpRequest, HttpResponse
-from django.urls import reverse_lazy
-from django.utils.translation import gettext as _
-from django.views.generic import DetailView, FormView, UpdateView
-from guardian.mixins import PermissionRequiredMixin
-
-from authentik.flows.exceptions import FlowNonApplicableException
-from authentik.flows.forms import FlowForm, FlowImportForm
-from authentik.flows.models import Flow
-from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER
-from authentik.flows.transfer.importer import FlowImporter
-from authentik.flows.views import SESSION_KEY_PLAN, FlowPlanner
-from authentik.lib.utils.urls import redirect_with_qs
-from authentik.lib.views import CreateAssignPermView, bad_request_message
-
-
-class FlowCreateView(
-    SuccessMessageMixin,
-    LoginRequiredMixin,
-    DjangoPermissionRequiredMixin,
-    CreateAssignPermView,
-):
-    """Create new Flow"""
-
-    model = Flow
-    form_class = FlowForm
-    permission_required = "authentik_flows.add_flow"
-
-    template_name = "generic/create.html"
-    success_url = reverse_lazy("authentik_core:if-admin")
-    success_message = _("Successfully created Flow")
-
-
-class FlowUpdateView(
-    SuccessMessageMixin,
-    LoginRequiredMixin,
-    PermissionRequiredMixin,
-    UpdateView,
-):
-    """Update flow"""
-
-    model = Flow
-    form_class = FlowForm
-    permission_required = "authentik_flows.change_flow"
-
-    template_name = "generic/update.html"
-    success_url = reverse_lazy("authentik_core:if-admin")
-    success_message = _("Successfully updated Flow")
-
-
-class FlowDebugExecuteView(LoginRequiredMixin, PermissionRequiredMixin, DetailView):
-    """Debug exectue flow, setting the current user as pending user"""
-
-    model = Flow
-    permission_required = "authentik_flows.view_flow"
-
-    # pylint: disable=unused-argument
-    def get(self, request: HttpRequest, pk: str) -> HttpResponse:
-        """Debug exectue flow, setting the current user as pending user"""
-        flow: Flow = self.get_object()
-        planner = FlowPlanner(flow)
-        planner.use_cache = False
-        try:
-            plan = planner.plan(self.request, {PLAN_CONTEXT_PENDING_USER: request.user})
-            self.request.session[SESSION_KEY_PLAN] = plan
-        except FlowNonApplicableException as exc:
-            return bad_request_message(
-                request,
-                _(
-                    "Flow not applicable to current user/request: %(messages)s"
-                    % {"messages": str(exc)}
-                ),
-            )
-        return redirect_with_qs(
-            "authentik_core:if-flow",
-            self.request.GET,
-            flow_slug=flow.slug,
-        )
-
-
-class FlowImportView(LoginRequiredMixin, FormView):
-    """Import flow from JSON Export; only allowed for superusers
-    as these flows can contain python code"""
-
-    form_class = FlowImportForm
-    template_name = "administration/flow/import.html"
-    success_url = reverse_lazy("authentik_core:if-admin")
-
-    def dispatch(self, request, *args, **kwargs):
-        if not request.user.is_superuser:
-            return self.handle_no_permission()
-        return super().dispatch(request, *args, **kwargs)
-
-    def form_valid(self, form: FlowImportForm) -> HttpResponse:
-        importer = FlowImporter(form.cleaned_data["flow"].read().decode())
-        successful = importer.apply()
-        if not successful:
-            messages.error(self.request, _("Failed to import flow."))
-        else:
-            messages.success(self.request, _("Successfully imported flow."))
-        return super().form_valid(form)
diff --git a/authentik/api/v2/config.py b/authentik/api/v2/config.py
index e878f024e..858f7f5e2 100644
--- a/authentik/api/v2/config.py
+++ b/authentik/api/v2/config.py
@@ -11,7 +11,7 @@ from rest_framework.viewsets import ViewSet
 from authentik.lib.config import CONFIG
 
 
-class LinkSerializer(Serializer):
+class FooterLinkSerializer(Serializer):
     """Links returned in Config API"""
 
     href = CharField(read_only=True)
@@ -29,7 +29,7 @@ class ConfigSerializer(Serializer):
 
     branding_logo = CharField(read_only=True)
     branding_title = CharField(read_only=True)
-    ui_footer_links = ListField(child=LinkSerializer(), read_only=True)
+    ui_footer_links = ListField(child=FooterLinkSerializer(), read_only=True)
 
     error_reporting_enabled = BooleanField(read_only=True)
     error_reporting_environment = CharField(read_only=True)
diff --git a/authentik/core/api/users.py b/authentik/core/api/users.py
index 1584a6c50..396b8f390 100644
--- a/authentik/core/api/users.py
+++ b/authentik/core/api/users.py
@@ -13,6 +13,7 @@ from rest_framework.viewsets import ModelViewSet
 
 from authentik.admin.api.metrics import CoordinateSerializer, get_events_per_1h
 from authentik.api.decorators import permission_required
+from authentik.core.api.utils import LinkSerializer
 from authentik.core.middleware import (
     SESSION_IMPERSONATE_ORIGINAL_USER,
     SESSION_IMPERSONATE_USER,
@@ -57,18 +58,6 @@ class SessionUserSerializer(Serializer):
         raise NotImplementedError
 
 
-class UserRecoverySerializer(Serializer):
-    """Recovery link for a user to reset their password"""
-
-    link = CharField()
-
-    def create(self, validated_data: dict) -> Model:
-        raise NotImplementedError
-
-    def update(self, instance: Model, validated_data: dict) -> Model:
-        raise NotImplementedError
-
-
 class UserMetricsSerializer(Serializer):
     """User Metrics"""
 
@@ -142,7 +131,7 @@ class UserViewSet(ModelViewSet):
 
     @permission_required("authentik_core.reset_user_password")
     @swagger_auto_schema(
-        responses={"200": UserRecoverySerializer(many=False)},
+        responses={"200": LinkSerializer(many=False)},
     )
     @action(detail=True)
     # pylint: disable=invalid-name, unused-argument
diff --git a/authentik/core/api/utils.py b/authentik/core/api/utils.py
index 438efa2df..5b39c0cc4 100644
--- a/authentik/core/api/utils.py
+++ b/authentik/core/api/utils.py
@@ -49,3 +49,15 @@ class CacheSerializer(Serializer):
 
     def update(self, instance: Model, validated_data: dict) -> Model:
         raise NotImplementedError
+
+
+class LinkSerializer(Serializer):
+    """Returns a single link"""
+
+    link = CharField()
+
+    def create(self, validated_data: dict) -> Model:
+        raise NotImplementedError
+
+    def update(self, instance: Model, validated_data: dict) -> Model:
+        raise NotImplementedError
diff --git a/authentik/flows/api/flows.py b/authentik/flows/api/flows.py
index bf541b13c..c27878e9a 100644
--- a/authentik/flows/api/flows.py
+++ b/authentik/flows/api/flows.py
@@ -4,6 +4,8 @@ from dataclasses import dataclass
 from django.core.cache import cache
 from django.db.models import Model
 from django.http.response import HttpResponseBadRequest, JsonResponse
+from django.urls import reverse
+from django.utils.translation import gettext as _
 from drf_yasg import openapi
 from drf_yasg.utils import no_body, swagger_auto_schema
 from guardian.shortcuts import get_objects_for_user
@@ -21,11 +23,16 @@ from rest_framework.viewsets import ModelViewSet
 from structlog.stdlib import get_logger
 
 from authentik.api.decorators import permission_required
-from authentik.core.api.utils import CacheSerializer
+from authentik.core.api.utils import CacheSerializer, LinkSerializer
+from authentik.flows.exceptions import FlowNonApplicableException
 from authentik.flows.models import Flow
-from authentik.flows.planner import cache_key
+from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlanner, cache_key
 from authentik.flows.transfer.common import DataclassEncoder
 from authentik.flows.transfer.exporter import FlowExporter
+from authentik.flows.transfer.importer import FlowImporter
+from authentik.flows.views import SESSION_KEY_PLAN
+from authentik.lib.utils.urls import redirect_with_qs
+from authentik.lib.views import bad_request_message
 
 LOGGER = get_logger()
 
@@ -57,7 +64,7 @@ class FlowSerializer(ModelSerializer):
 
 
 class FlowDiagramSerializer(Serializer):
-    """response of the flow's /diagram/ action"""
+    """response of the flow's diagram action"""
 
     diagram = CharField(read_only=True)
 
@@ -89,14 +96,14 @@ class FlowViewSet(ModelViewSet):
     search_fields = ["name", "slug", "designation", "title"]
     filterset_fields = ["flow_uuid", "name", "slug", "designation"]
 
-    @permission_required("authentik_flows.view_flow_cache")
+    @permission_required(None, ["authentik_flows.view_flow_cache"])
     @swagger_auto_schema(responses={200: CacheSerializer(many=False)})
     @action(detail=False)
     def cache_info(self, request: Request) -> Response:
         """Info about cached flows"""
         return Response(data={"count": len(cache.keys("flow_*"))})
 
-    @permission_required("authentik_flows.clear_flow_cache")
+    @permission_required(None, ["authentik_flows.clear_flow_cache"])
     @swagger_auto_schema(
         request_body=no_body,
         responses={204: "Successfully cleared cache", 400: "Bad request"},
@@ -109,7 +116,61 @@ class FlowViewSet(ModelViewSet):
         LOGGER.debug("Cleared flow cache", keys=len(keys))
         return Response(status=204)
 
-    @permission_required("authentik_flows.export_flow")
+    @permission_required(
+        None,
+        [
+            "authentik_flows.add_flow",
+            "authentik_flows.change_flow",
+            "authentik_flows.add_flowstagebinding",
+            "authentik_flows.change_flowstagebinding",
+            "authentik_flows.add_stage",
+            "authentik_flows.change_stage",
+            "authentik_policies.add_policy",
+            "authentik_policies.change_policy",
+            "authentik_policies.add_policybinding",
+            "authentik_policies.change_policybinding",
+            "authentik_stages_prompt.add_prompt",
+            "authentik_stages_prompt.change_prompt",
+        ],
+    )
+    @swagger_auto_schema(
+        request_body=no_body,
+        manual_parameters=[
+            openapi.Parameter(
+                name="file",
+                in_=openapi.IN_FORM,
+                type=openapi.TYPE_FILE,
+                required=True,
+            )
+        ],
+        responses={204: "Successfully imported flow", 400: "Bad request"},
+    )
+    @action(detail=False, methods=["POST"], parser_classes=(MultiPartParser,))
+    def import_flow(self, request: Request) -> Response:
+        """Import flow from .akflow file"""
+        file = request.FILES.get("file", None)
+        if not file:
+            return HttpResponseBadRequest()
+        importer = FlowImporter(file.read().decode())
+        valid = importer.validate()
+        if not valid:
+            return HttpResponseBadRequest()
+        successful = importer.apply()
+        if not successful:
+            return Response(status=204)
+        return HttpResponseBadRequest()
+
+    @permission_required(
+        "authentik_flows.export_flow",
+        [
+            "authentik_flows.view_flow",
+            "authentik_flows.view_flowstagebinding",
+            "authentik_flows.view_stage",
+            "authentik_policies.view_policy",
+            "authentik_policies.view_policybinding",
+            "authentik_stages_prompt.view_prompt",
+        ],
+    )
     @swagger_auto_schema(
         responses={
             "200": openapi.Response(
@@ -220,3 +281,32 @@ class FlowViewSet(ModelViewSet):
         app.background = icon
         app.save()
         return Response({})
+
+    @swagger_auto_schema(
+        responses={200: LinkSerializer(many=False)},
+    )
+    @action(detail=True)
+    # pylint: disable=unused-argument
+    def execute(self, request: Request, slug: str):
+        """Execute flow for current user"""
+        flow: Flow = self.get_object()
+        planner = FlowPlanner(flow)
+        planner.use_cache = False
+        try:
+            plan = planner.plan(self.request, {PLAN_CONTEXT_PENDING_USER: request.user})
+            self.request.session[SESSION_KEY_PLAN] = plan
+        except FlowNonApplicableException as exc:
+            return bad_request_message(
+                request,
+                _(
+                    "Flow not applicable to current user/request: %(messages)s"
+                    % {"messages": str(exc)}
+                ),
+            )
+        return Response(
+            {
+                "link": request._request.build_absolute_uri(
+                    reverse("authentik_core:if-flow", kwargs={"flow_slug": flow.slug})
+                )
+            }
+        )
diff --git a/authentik/flows/forms.py b/authentik/flows/forms.py
index 2e8ca9d14..f8082aa67 100644
--- a/authentik/flows/forms.py
+++ b/authentik/flows/forms.py
@@ -1,35 +1,10 @@
 """Flow and Stage forms"""
-
 from django import forms
-from django.core.validators import FileExtensionValidator
-from django.forms import ValidationError
-from django.utils.translation import gettext_lazy as _
 
-from authentik.flows.models import Flow, FlowStageBinding, Stage
-from authentik.flows.transfer.importer import FlowImporter
+from authentik.flows.models import FlowStageBinding, Stage
 from authentik.lib.widgets import GroupedModelChoiceField
 
 
-class FlowForm(forms.ModelForm):
-    """Flow Form"""
-
-    class Meta:
-
-        model = Flow
-        fields = [
-            "name",
-            "title",
-            "slug",
-            "designation",
-            "background",
-        ]
-        widgets = {
-            "name": forms.TextInput(),
-            "title": forms.TextInput(),
-            "background": forms.FileInput(),
-        }
-
-
 class FlowStageBindingForm(forms.ModelForm):
     """FlowStageBinding Form"""
 
@@ -56,20 +31,3 @@ class FlowStageBindingForm(forms.ModelForm):
         widgets = {
             "name": forms.TextInput(),
         }
-
-
-class FlowImportForm(forms.Form):
-    """Form used for flow importing"""
-
-    flow = forms.FileField(
-        validators=[FileExtensionValidator(allowed_extensions=["akflow"])]
-    )
-
-    def clean_flow(self):
-        """Check if the flow is valid and rewind the file to the start"""
-        flow = self.cleaned_data["flow"].read()
-        valid = FlowImporter(flow.decode()).validate()
-        if not valid:
-            raise ValidationError(_("Flow invalid."))
-        self.cleaned_data["flow"].seek(0)
-        return self.cleaned_data["flow"]
diff --git a/swagger.yaml b/swagger.yaml
index ea5f770bc..6acf37ae3 100755
--- a/swagger.yaml
+++ b/swagger.yaml
@@ -2267,7 +2267,7 @@ paths:
         '200':
           description: ''
           schema:
-            $ref: '#/definitions/UserRecovery'
+            $ref: '#/definitions/Link'
         '403':
           description: Authentication credentials were invalid, absent or insufficient.
           schema:
@@ -3898,6 +3898,29 @@ paths:
       tags:
         - flows
     parameters: []
+  /flows/instances/import_flow/:
+    post:
+      operationId: flows_instances_import_flow
+      description: Import flow from .akflow file
+      parameters:
+        - name: file
+          in: formData
+          required: true
+          type: file
+      responses:
+        '204':
+          description: Successfully imported flow
+        '400':
+          description: Bad request
+        '403':
+          description: Authentication credentials were invalid, absent or insufficient.
+          schema:
+            $ref: '#/definitions/GenericError'
+      consumes:
+        - multipart/form-data
+      tags:
+        - flows
+    parameters: []
   /flows/instances/{slug}/:
     get:
       operationId: flows_instances_read
@@ -4033,6 +4056,35 @@ paths:
         type: string
         format: slug
         pattern: ^[-a-zA-Z0-9_]+$
+  /flows/instances/{slug}/execute/:
+    get:
+      operationId: flows_instances_execute
+      description: Execute flow for current user
+      parameters: []
+      responses:
+        '200':
+          description: ''
+          schema:
+            $ref: '#/definitions/Link'
+        '403':
+          description: Authentication credentials were invalid, absent or insufficient.
+          schema:
+            $ref: '#/definitions/GenericError'
+        '404':
+          description: Object does not exist or caller has insufficient permissions
+            to access it.
+          schema:
+            $ref: '#/definitions/APIException'
+      tags:
+        - flows
+    parameters:
+      - name: slug
+        in: path
+        description: Visible in the URL.
+        required: true
+        type: string
+        format: slug
+        pattern: ^[-a-zA-Z0-9_]+$
   /flows/instances/{slug}/export/:
     get:
       operationId: flows_instances_export
@@ -14908,7 +14960,7 @@ definitions:
         items:
           $ref: '#/definitions/Coordinate'
         readOnly: true
-  UserRecovery:
+  Link:
     required:
       - link
     type: object
@@ -17035,7 +17087,7 @@ definitions:
         title: Metadata
         type: string
         readOnly: true
-  Link:
+  FooterLink:
     type: object
     properties:
       href:
@@ -17064,7 +17116,7 @@ definitions:
       ui_footer_links:
         type: array
         items:
-          $ref: '#/definitions/Link'
+          $ref: '#/definitions/FooterLink'
         readOnly: true
       error_reporting_enabled:
         title: Error reporting enabled
diff --git a/web/src/pages/applications/ApplicationForm.ts b/web/src/pages/applications/ApplicationForm.ts
index 7806daa61..25a1bd943 100644
--- a/web/src/pages/applications/ApplicationForm.ts
+++ b/web/src/pages/applications/ApplicationForm.ts
@@ -26,7 +26,7 @@ export class ApplicationForm extends Form<Application> {
         }
     }
 
-    send = (data: Application): Promise<Application> => {
+    send = (data: Application): Promise<Application | void> => {
         let writeOp: Promise<Application>;
         if (this.application) {
             writeOp = new CoreApi(DEFAULT_CONFIG).coreApplicationsUpdate({