diff --git a/skyvern/forge/sdk/routes/agent_protocol.py b/skyvern/forge/sdk/routes/agent_protocol.py
index 2b401ff3..c4d761b3 100644
--- a/skyvern/forge/sdk/routes/agent_protocol.py
+++ b/skyvern/forge/sdk/routes/agent_protocol.py
@@ -481,8 +481,9 @@ async def get_task_actions(
 )
 async def execute_workflow(
     background_tasks: BackgroundTasks,
-    workflow_id: str,
+    workflow_id: str,  # this is the workflow_permanent_id
     workflow_request: WorkflowRequestBody,
+    version: int | None = None,
     current_org: Organization = Depends(org_auth_service.get_current_org),
     x_api_key: Annotated[str | None, Header()] = None,
     x_max_steps_override: Annotated[int | None, Header()] = None,
@@ -493,8 +494,9 @@ async def execute_workflow(
     workflow_run = await app.WORKFLOW_SERVICE.setup_workflow_run(
         request_id=request_id,
         workflow_request=workflow_request,
-        workflow_id=workflow_id,
+        workflow_permanent_id=workflow_id,
         organization_id=current_org.organization_id,
+        version=version,
         max_steps_override=x_max_steps_override,
     )
     if x_max_steps_override:
@@ -502,7 +504,7 @@ async def execute_workflow(
     await AsyncExecutorFactory.get_executor().execute_workflow(
         background_tasks=background_tasks,
         organization_id=current_org.organization_id,
-        workflow_id=workflow_id,
+        workflow_id=workflow_run.workflow_id,
         workflow_run_id=workflow_run.workflow_run_id,
         max_steps_override=x_max_steps_override,
         api_key=x_api_key,
diff --git a/skyvern/forge/sdk/workflow/service.py b/skyvern/forge/sdk/workflow/service.py
index 183c62ec..35b55dec 100644
--- a/skyvern/forge/sdk/workflow/service.py
+++ b/skyvern/forge/sdk/workflow/service.py
@@ -68,8 +68,9 @@ class WorkflowService:
         self,
         request_id: str | None,
         workflow_request: WorkflowRequestBody,
-        workflow_id: str,
+        workflow_permanent_id: str,
         organization_id: str,
+        version: int | None = None,
         max_steps_override: int | None = None,
     ) -> WorkflowRun:
         """
@@ -83,13 +84,15 @@ class WorkflowService:
         :return: The created workflow run.
         """
         # Validate the workflow and the organization
-        workflow = await self.get_workflow(workflow_id=workflow_id, organization_id=organization_id)
+        workflow = await self.get_workflow_by_permanent_id(
+            workflow_permanent_id=workflow_permanent_id,
+            organization_id=organization_id,
+            version=version,
+        )
         if workflow is None:
-            LOG.error(f"Workflow {workflow_id} not found")
-            raise WorkflowNotFound(workflow_id=workflow_id)
-        if workflow.organization_id != organization_id:
-            LOG.error(f"Workflow {workflow_id} does not belong to organization {organization_id}")
-            raise WorkflowOrganizationMismatch(workflow_id=workflow_id, organization_id=organization_id)
+            LOG.error(f"Workflow {workflow_permanent_id} not found", workflow_version=version)
+            raise WorkflowNotFound(workflow_permanent_id=workflow_permanent_id, version=version)
+        workflow_id = workflow.workflow_id
         if workflow_request.proxy_location is None and workflow.proxy_location is not None:
             workflow_request.proxy_location = workflow.proxy_location
         if workflow_request.webhook_callback_url is None and workflow.webhook_callback_url is not None:
@@ -583,7 +586,7 @@ class WorkflowService:
             )
             outputs = None
         return WorkflowRunStatusResponse(
-            workflow_id=workflow_id,
+            workflow_id=workflow.workflow_permanent_id,
             workflow_run_id=workflow_run_id,
             status=workflow_run.status,
             proxy_location=workflow_run.proxy_location,