Add a query param to filter out workflow tasks

skyvern/forge/sdk/db/client.py

@@ -438,13 +438,15 @@ async def get_tasks(
         task_status: list[TaskStatus] | None = None,
         workflow_run_id: str | None = None,
         organization_id: str | None = None,
+        only_standalone_tasks: bool = False,
     ) -> list[Task]:
         """
         Get all tasks.
         :param page: Starts at 1
         :param page_size:
         :param task_status:
         :param workflow_run_id:
+        :param only_standalone_tasks:
         :return:
         """
         if page < 1:
@@ -458,6 +460,8 @@ async def get_tasks(
                     query = query.filter(TaskModel.status.in_(task_status))
                 if workflow_run_id:
                     query = query.filter(TaskModel.workflow_run_id == workflow_run_id)
+                if only_standalone_tasks:
+                    query = query.filter(TaskModel.workflow_run_id.is_(None))
                 query = query.order_by(TaskModel.created_at.desc()).limit(page_size).offset(db_page * page_size)
                 tasks = (await session.scalars(query)).all()
                 return [convert_to_task(task, debug_enabled=self.debug_enabled) for task in tasks]

skyvern/forge/sdk/routes/agent_protocol.py

@@ -383,23 +383,31 @@ async def get_agent_tasks(
     task_status: Annotated[list[TaskStatus] | None, Query()] = None,
     workflow_run_id: Annotated[str | None, Query()] = None,
     current_org: Organization = Depends(org_auth_service.get_current_org),
+    only_standalone_tasks: bool = Query(False),
 ) -> Response:
     """
     Get all tasks.
     :param page: Starting page, defaults to 1
     :param page_size: Page size, defaults to 10
     :param task_status: Task status filter
     :param workflow_run_id: Workflow run id filter
+    :param only_standalone_tasks: Only standalone tasks, tasks which are part of a workflow run will be filtered out
     :return: List of tasks with pagination without steps populated. Steps can be populated by calling the
         get_agent_task endpoint.
     """
     analytics.capture("skyvern-oss-agent-tasks-get")
+    if only_standalone_tasks and workflow_run_id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="only_standalone_tasks and workflow_run_id cannot be used together",
+        )
     tasks = await app.DATABASE.get_tasks(
         page,
         page_size,
         task_status=task_status,
         workflow_run_id=workflow_run_id,
         organization_id=current_org.organization_id,
+        only_standalone_tasks=only_standalone_tasks,
     )
     return ORJSONResponse([task.to_task_response().model_dump() for task in tasks])