mgmt flow functions

README.md

@@ -1096,6 +1096,62 @@ exported_theme = descope_client.mgmt.flow.export_theme()
 imported_theme = descope_client.mgmt.flow.import_flow(
     theme={}
 )
+
+# Run a flow with the given flow id and options.
+# You can use the FlowRunOptions class or pass a dict directly.
+from descope import FlowRunOptions
+
+# Using FlowRunOptions class
+result = descope_client.mgmt.flow.run_flow(
+    flow_id="my-flow-id",
+    options=FlowRunOptions(
+        input={"key": "value"},
+        preview=True,
+        tenant="tenant-id",
+    ),
+)
+
+# Or using a dict
+result = descope_client.mgmt.flow.run_flow(
+    flow_id="my-flow-id",
+    options={
+        "input": {"key": "value"},
+        "preview": True,
+        "tenant": "tenant-id",
+    },
+)
+
+# Run a flow asynchronously for long-running flows.
+# This returns immediately with an execution ID that can be used to check the result later.
+from descope import FlowRunOptions
+
+# Start an async flow run
+async_result = descope_client.mgmt.flow.run_flow_async(
+    flow_id="my-flow-id",
+    options=FlowRunOptions(
+        flow_input={"key": "value"},
+        preview=True,
+        tenant="tenant-id",
+    ),
+)
+execution_id = async_result["executionId"]
+
+# Or using a dict
+async_result = descope_client.mgmt.flow.run_flow_async(
+    flow_id="my-flow-id",
+    options={
+        "input": {"key": "value"},
+        "preview": True,
+        "tenant": "tenant-id",
+    },
+)
+execution_id = async_result["executionId"]
+
+# Check the result of an async flow execution
+# Poll this endpoint until the flow completes
+result = descope_client.mgmt.flow.get_flow_async_result(
+    execution_id=execution_id,
+)
 ```
 
 ### Query SSO Groups

descope/__init__.py

@@ -26,6 +26,7 @@
     DescoperRBAC,
     DescoperRole,
     DescoperTagRole,
+    FlowRunOptions,
     MgmtKeyProjectRole,
     MgmtKeyReBac,
     MgmtKeyStatus,

descope/management/common.py

@@ -209,6 +209,9 @@ class MgmtV1:
     flow_delete_path = "/v1/mgmt/flow/delete"
     flow_import_path = "/v1/mgmt/flow/import"
     flow_export_path = "/v1/mgmt/flow/export"
+    flow_run_path = "/v1/mgmt/flow/run"
+    flow_async_run_path = "/v1/mgmt/flow/async/run"
+    flow_async_result_path = "/v1/mgmt/flow/async/result"
 
     # theme
     theme_import_path = "/v1/mgmt/theme/import"
@@ -283,6 +286,42 @@ def __init__(
         self.refresh_duration = refresh_duration
 
 
+class FlowRunOptions:
+    """
+    Options for running a flow.
+    """
+
+    def __init__(
+        self,
+        flow_input: Optional[Dict[str, Any]] = None,
+        preview: Optional[bool] = None,
+        tenant: Optional[str] = None,
+    ):
+        self.flow_input = flow_input
+        self.preview = preview
+        self.tenant = tenant
+
+    def to_dict(self) -> Dict[str, Any]:
+        result: Dict[str, Any] = {}
+        if self.flow_input is not None:
+            result["input"] = self.flow_input
+        if self.preview is not None:
+            result["preview"] = self.preview
+        if self.tenant is not None:
+            result["tenant"] = self.tenant
+        return result
+
+    @staticmethod
+    def from_dict(options: Optional[dict]) -> Optional["FlowRunOptions"]:
+        if options is None:
+            return None
+        return FlowRunOptions(
+            flow_input=options.get("input"),
+            preview=options.get("preview"),
+            tenant=options.get("tenant"),
+        )
+
+
 class MgmtLoginOptions:
     def __init__(
         self,

descope/management/flow.py

@@ -1,7 +1,7 @@
-from typing import List
+from typing import List, Optional, Union
 
 from descope._http_base import HTTPBase
-from descope.management.common import MgmtV1
+from descope.management.common import FlowRunOptions, MgmtV1
 
 
 class Flow(HTTPBase):
@@ -146,3 +146,95 @@ def import_theme(
             },
         )
         return response.json()
+
+    def run_flow(
+        self,
+        flow_id: str,
+        options: Optional[Union[FlowRunOptions, dict]] = None,
+    ) -> dict:
+        """
+        Run a flow with the given flow id and options.
+
+        Args:
+        flow_id (str): the flow id to run.
+        options (Optional[Union[FlowRunOptions, dict]]): optional flow run options containing:
+            - input: optional input data to pass to the flow.
+            - preview: optional flag to run the flow in preview mode.
+            - tenant: optional tenant ID to run the flow for.
+
+        Return value (dict):
+        Return dict with the flow execution result.
+
+        Raise:
+        AuthException: raised if run operation fails
+        """
+        body: dict = {"flowId": flow_id}
+
+        if options is not None:
+            if isinstance(options, dict):
+                options = FlowRunOptions.from_dict(options)
+            if options is not None:
+                body.update(options.to_dict())
+
+        response = self._http.post(
+            MgmtV1.flow_run_path,
+            body=body,
+        )
+        return response.json()
+
+    def run_flow_async(
+        self,
+        flow_id: str,
+        options: Optional[Union[FlowRunOptions, dict]] = None,
+    ) -> dict:
+        """
+        Run a flow asynchronously with the given flow id and options.
+
+        Args:
+        flow_id (str): the flow id to run.
+        options (Optional[Union[FlowRunOptions, dict]]): optional flow run options containing:
+            - input: optional input data to pass to the flow.
+            - preview: optional flag to run the flow in preview mode.
+            - tenant: optional tenant ID to run the flow for.
+
+        Return value (dict):
+        Return dict with the async flow execution result.
+
+        Raise:
+        AuthException: raised if run operation fails
+        """
+        body: dict = {"flowId": flow_id}
+
+        if options is not None:
+            if isinstance(options, dict):
+                options = FlowRunOptions.from_dict(options)
+            if options is not None:
+                body.update(options.to_dict())
+
+        response = self._http.post(
+            MgmtV1.flow_async_run_path,
+            body=body,
+        )
+        return response.json()
+
+    def get_flow_async_result(
+        self,
+        execution_id: str,
+    ) -> dict:
+        """
+        Get the result of an async flow execution.
+
+        Args:
+        execution_id (str): the execution id returned from run_flow_async.
+
+        Return value (dict):
+        Return dict with the async flow execution result.
+
+        Raise:
+        AuthException: raised if the operation fails
+        """
+        response = self._http.post(
+            MgmtV1.flow_async_result_path,
+            body={"executionId": execution_id},
+        )
+        return response.json()