introduce permission checker to task create api (#368)

skyvern/forge/agent_functions.py

@@ -1,5 +1,6 @@
 from playwright.async_api import Page
 
+from skyvern.exceptions import StepTerminationError
 from skyvern.forge import app
 from skyvern.forge.async_operations import AsyncOperation
 from skyvern.forge.sdk.models import Organization, Step, StepStatus
@@ -33,7 +34,7 @@ class AgentFunction:
 
         can_execute = has_valid_task_status and has_valid_step_status and has_no_running_steps
         if not can_execute:
-            raise Exception(f"Cannot execute step. Reasons: {reasons}, Step: {step}")
+            raise StepTerminationError(step_id=step.step_id, reason="Cannot execute step. Reasons: {reasons}")
 
     def generate_async_operations(
         self,

skyvern/forge/sdk/core/permissions/permission_checker_factory.py

@@ -0,0 +1,13 @@
+from skyvern.forge.sdk.core.permissions.permission_checkers import NoopPermissionChecker, PermissionChecker
+
+
+class PermissionCheckerFactory:
+    __instance: PermissionChecker = NoopPermissionChecker()
+
+    @staticmethod
+    def get_instance() -> PermissionChecker:
+        return PermissionCheckerFactory.__instance
+
+    @staticmethod
+    def set_instance(permission_checker: PermissionChecker) -> None:
+        PermissionCheckerFactory.__instance = permission_checker

skyvern/forge/sdk/core/permissions/permission_checkers.py

@@ -0,0 +1,14 @@
+import abc
+
+from skyvern.forge.sdk.models import Organization
+
+
+class PermissionChecker(abc.ABC):
+    @abc.abstractmethod
+    async def check(self, organization: Organization) -> None:
+        pass
+
+
+class NoopPermissionChecker(PermissionChecker):
+    async def check(self, organization: Organization) -> None:
+        return

skyvern/forge/sdk/routes/agent_protocol.py

@@ -11,6 +11,7 @@ from skyvern.exceptions import StepNotFound
 from skyvern.forge import app
 from skyvern.forge.sdk.artifact.models import Artifact, ArtifactType
 from skyvern.forge.sdk.core import skyvern_context
+from skyvern.forge.sdk.core.permissions.permission_checker_factory import PermissionCheckerFactory
 from skyvern.forge.sdk.core.security import generate_skyvern_signature
 from skyvern.forge.sdk.executor.factory import AsyncExecutorFactory
 from skyvern.forge.sdk.models import Organization, Step
@@ -99,6 +100,7 @@ async def create_agent_task(
     x_max_steps_override: Annotated[int | None, Header()] = None,
 ) -> CreateTaskResponse:
     analytics.capture("skyvern-oss-agent-task-create", data={"url": task.url})
+    await PermissionCheckerFactory.get_instance().check(current_org)
 
     if current_org and current_org.organization_name == "CoverageCat":
         task.proxy_location = ProxyLocation.RESIDENTIAL