mgmt flow functions

README.md

@@ -1096,6 +1096,30 @@ 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",
+    },
+)
 ```
 
 ### Query SSO Groups

descope/__init__.py

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

descope/management/common.py

@@ -209,6 +209,7 @@ 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"
 
     # theme
     theme_import_path = "/v1/mgmt/theme/import"
@@ -283,6 +284,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,38 @@ 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()

tests/management/test_flow.py

@@ -2,7 +2,7 @@
 
 from descope import AuthException, DescopeClient
 from descope.common import DEFAULT_TIMEOUT_SECONDS
-from descope.management.common import MgmtV1
+from descope.management.common import MgmtV1, FlowRunOptions
 
 from .. import common
 
@@ -234,3 +234,107 @@ def test_import_theme(self):
                 verify=True,
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
+
+    def test_run_flow(self):
+        client = DescopeClient(
+            self.dummy_project_id,
+            self.public_key_dict,
+            False,
+            self.dummy_management_key,
+        )
+
+        # Test failed run flow
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = False
+            self.assertRaises(
+                AuthException,
+                client.mgmt.flow.run_flow,
+                "test-flow",
+            )
+
+        # Test success run flow with no options
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            self.assertIsNotNone(client.mgmt.flow.run_flow("test-flow"))
+            mock_post.assert_called_with(
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.flow_run_path}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={"flowId": "test-flow"},
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+        # Test success run flow with dict options
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            self.assertIsNotNone(
+                client.mgmt.flow.run_flow(
+                    "test-flow",
+                    {"input": {"key": "value"}, "preview": True, "tenant": "tenant-id"},
+                )
+            )
+            mock_post.assert_called_with(
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.flow_run_path}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={
+                    "flowId": "test-flow",
+                    "input": {"key": "value"},
+                    "preview": True,
+                    "tenant": "tenant-id",
+                },
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+        # Test success run flow with FlowRunOptions object
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            options = FlowRunOptions(
+                flow_input={"key": "value"},
+                preview=True,
+                tenant="tenant-id",
+            )
+            self.assertIsNotNone(client.mgmt.flow.run_flow("test-flow", options))
+            mock_post.assert_called_with(
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.flow_run_path}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={
+                    "flowId": "test-flow",
+                    "input": {"key": "value"},
+                    "preview": True,
+                    "tenant": "tenant-id",
+                },
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+    def test_flow_run_options_from_dict(self):
+        # Test from_dict with None returns None
+        self.assertIsNone(FlowRunOptions.from_dict(None))
+
+        # Test from_dict with valid dict
+        options = FlowRunOptions.from_dict(
+            {"input": {"key": "value"}, "preview": True, "tenant": "tenant-id"}
+        )
+        self.assertIsNotNone(options)
+        self.assertEqual(options.flow_input, {"key": "value"})
+        self.assertEqual(options.preview, True)
+        self.assertEqual(options.tenant, "tenant-id")