feat(agent): QuickForm escalation + tasks-service create_quickform_async

packages/uipath-platform/src/uipath/platform/action_center/_tasks_service.py

@@ -164,6 +164,68 @@ def _create_spec(
     )
 
 
+def _create_quickform_spec(
+    title: str,
+    task_schema_key: str,
+    schema: Optional[Dict[str, Any]],
+    data: Optional[Dict[str, Any]],
+    *,
+    folder_path: Optional[str] = None,
+    folder_key: Optional[str] = None,
+    priority: Optional[str] = None,
+    labels: Optional[List[str]] = None,
+    is_actionable_message_enabled: Optional[bool] = None,
+    actionable_message_metadata: Optional[Dict[str, Any]] = None,
+    source_name: str = "Agent",
+) -> RequestSpec:
+    json_payload: Dict[str, Any] = {
+        "type": "QuickFormTask",
+        "title": title,
+        "taskSchemaKey": task_schema_key,
+        "data": data if data is not None else {},
+    }
+    if schema is not None:
+        json_payload["schema"] = schema
+    if priority and (normalized_priority := _normalize_priority(priority)):
+        json_payload["priority"] = normalized_priority
+    if labels is not None:
+        json_payload["tags"] = [
+            {
+                "Name": label,
+                "DisplayName": label,
+                "Value": label,
+                "DisplayValue": label,
+            }
+            for label in labels
+        ]
+    if is_actionable_message_enabled is not None:
+        json_payload["isActionableMessageEnabled"] = is_actionable_message_enabled
+    if actionable_message_metadata is not None:
+        json_payload["actionableMessageMetaData"] = actionable_message_metadata
+
+    project_id = UiPathConfig.project_id
+    trace_id = UiPathConfig.trace_id
+    if project_id and trace_id:
+        json_payload["taskSource"] = {
+            "sourceName": source_name,
+            "sourceId": project_id,
+            "jobId": UiPathConfig.job_key,
+            "taskSourceMetadata": {
+                "InstanceId": trace_id,
+                "FolderKey": UiPathConfig.folder_key,
+                "JobKey": UiPathConfig.job_key,
+                "ProcessKey": UiPathConfig.process_uuid,
+            },
+        }
+
+    return RequestSpec(
+        method="POST",
+        endpoint=Endpoint("/orchestrator_/tasks/GenericTasks/CreateTask"),
+        json=json_payload,
+        headers=header_folder(folder_key, folder_path),
+    )
+
+
 def _normalize_priority(priority: str | None) -> str | None:
     """Normalize priority string to match API expectations.
 
@@ -422,6 +484,81 @@ async def create_async(
             )
         return Task.model_validate(json_response)
 
+    @traced(name="tasks_create_quickform", run_type="uipath")
+    async def create_quickform_async(
+        self,
+        title: str,
+        task_schema_key: str,
+        schema: Optional[Dict[str, Any]] = None,
+        data: Optional[Dict[str, Any]] = None,
+        *,
+        folder_path: Optional[str] = None,
+        folder_key: Optional[str] = None,
+        recipient: Optional[TaskRecipient] = None,
+        priority: Optional[str] = None,
+        labels: Optional[List[str]] = None,
+        is_actionable_message_enabled: Optional[bool] = None,
+        actionable_message_metadata: Optional[Dict[str, Any]] = None,
+        source_name: str = "Agent",
+    ) -> Task:
+        """Create a QuickForm HITL task via Orchestrator's GenericTasks endpoint.
+
+        The schema (and its ``task_schema_key``) is sent inline so Orchestrator
+        upserts it on every call — the Agents runtime owns no separate schema
+        registration step. Mirrors ``create_async`` for the assignment hop:
+        the create call returns the task, then a follow-up assign request
+        binds it to the resolved recipient.
+
+        Args:
+            title: Title of the task as shown in Action Center.
+            task_schema_key: UUID under which the FormLib schema is stored.
+            schema: Optional inline FormLib schema (fields/outcomes) to upsert
+                against ``task_schema_key``.
+            data: Optional pre-filled form data.
+            folder_path: Optional folder path the task is created in.
+            folder_key: Optional folder key the task is created in.
+            recipient: Optional ``TaskRecipient`` to assign the task to.
+            priority: Optional priority (``Low``/``Medium``/``High``/``Critical``).
+            labels: Optional list of labels.
+            is_actionable_message_enabled: Toggle actionable notifications.
+            actionable_message_metadata: Custom actionable notification payload.
+            source_name: ``TaskSource`` ``sourceName`` value, default ``Agent``.
+
+        Returns:
+            The created ``Task``.
+        """
+        spec = _create_quickform_spec(
+            title=title,
+            task_schema_key=task_schema_key,
+            schema=schema,
+            data=data,
+            folder_path=folder_path,
+            folder_key=folder_key,
+            priority=priority,
+            labels=labels,
+            is_actionable_message_enabled=is_actionable_message_enabled,
+            actionable_message_metadata=actionable_message_metadata,
+            source_name=source_name,
+        )
+        response = await self.request_async(
+            spec.method,
+            spec.endpoint,
+            json=spec.json,
+            headers=spec.headers,
+        )
+        json_response = response.json()
+        if recipient:
+            assign_spec = await _assign_task_spec(
+                self, json_response["id"], None, recipient
+            )
+            await self.request_async(
+                assign_spec.method,
+                assign_spec.endpoint,
+                json=assign_spec.json,
+                content=assign_spec.content,
+            )
+        return Task.model_validate(json_response)
+
     @resource_override(
         resource_type="app",
         resource_identifier="app_name",

packages/uipath-platform/tests/services/test_actions_service.py

@@ -1,11 +1,16 @@
+import json
 from typing import Any
 
 import pytest
 from pytest_httpx import HTTPXMock
 
 from uipath.platform import UiPathApiConfig, UiPathExecutionContext
 from uipath.platform.action_center import Task
-from uipath.platform.action_center._tasks_service import TasksService
+from uipath.platform.action_center._tasks_service import (
+    TasksService,
+    _create_quickform_spec,
+)
+from uipath.platform.action_center.tasks import TaskRecipient, TaskRecipientType
 from uipath.platform.common.constants import HEADER_USER_AGENT
 
 
@@ -555,3 +560,213 @@ def test_create_raises_when_no_folder_key_or_path_provided(
                 app_name="my-app",
                 app_folder_path=None,
             )
+
+
+class TestCreateQuickFormSpec:
+    """Unit tests for the pure `_create_quickform_spec` payload builder."""
+
+    def test_minimal_payload_shape(self) -> None:
+        spec = _create_quickform_spec(
+            title="Approve invoice",
+            task_schema_key="schema-uuid",
+            schema=None,
+            data=None,
+        )
+
+        assert spec.method == "POST"
+        assert str(spec.endpoint) == "/orchestrator_/tasks/GenericTasks/CreateTask"
+        assert spec.json == {
+            "type": "QuickFormTask",
+            "title": "Approve invoice",
+            "taskSchemaKey": "schema-uuid",
+            "data": {},
+        }
+
+    def test_inlines_schema_when_provided(self) -> None:
+        form_schema = {"fields": [{"name": "amount", "type": "number"}]}
+
+        spec = _create_quickform_spec(
+            title="t",
+            task_schema_key="k",
+            schema=form_schema,
+            data={"amount": 42},
+        )
+
+        assert spec.json is not None
+        assert spec.json["schema"] == form_schema
+        assert spec.json["data"] == {"amount": 42}
+
+    def test_normalizes_priority_and_shapes_labels(self) -> None:
+        spec = _create_quickform_spec(
+            title="t",
+            task_schema_key="k",
+            schema=None,
+            data=None,
+            priority="high",
+            labels=["urgent", "finance"],
+        )
+
+        assert spec.json is not None
+        assert spec.json["priority"] == "High"
+        assert spec.json["tags"] == [
+            {
+                "Name": "urgent",
+                "DisplayName": "urgent",
+                "Value": "urgent",
+                "DisplayValue": "urgent",
+            },
+            {
+                "Name": "finance",
+                "DisplayName": "finance",
+                "Value": "finance",
+                "DisplayValue": "finance",
+            },
+        ]
+
+    def test_omits_optional_fields_when_unset(self) -> None:
+        spec = _create_quickform_spec(
+            title="t",
+            task_schema_key="k",
+            schema=None,
+            data=None,
+        )
+
+        assert spec.json is not None
+        assert "schema" not in spec.json
+        assert "priority" not in spec.json
+        assert "tags" not in spec.json
+        assert "isActionableMessageEnabled" not in spec.json
+        assert "actionableMessageMetaData" not in spec.json
+
+    def test_includes_actionable_message_fields_when_set(self) -> None:
+        spec = _create_quickform_spec(
+            title="t",
+            task_schema_key="k",
+            schema=None,
+            data=None,
+            is_actionable_message_enabled=True,
+            actionable_message_metadata={"channel": "teams"},
+        )
+
+        assert spec.json is not None
+        assert spec.json["isActionableMessageEnabled"] is True
+        assert spec.json["actionableMessageMetaData"] == {"channel": "teams"}
+
+    def test_propagates_folder_headers(self) -> None:
+        spec = _create_quickform_spec(
+            title="t",
+            task_schema_key="k",
+            schema=None,
+            data=None,
+            folder_key="folder-key-value",
+        )
+
+        assert spec.headers.get("x-uipath-folderkey") == "folder-key-value"
+
+
+class TestCreateQuickFormAsync:
+    """Integration tests for `TasksService.create_quickform_async`."""
+
+    @pytest.mark.anyio
+    async def test_posts_to_generic_tasks_endpoint(
+        self,
+        httpx_mock: HTTPXMock,
+        service: TasksService,
+        base_url: str,
+        org: str,
+        tenant: str,
+    ) -> None:
+        httpx_mock.add_response(
+            url=f"{base_url}{org}{tenant}/orchestrator_/tasks/GenericTasks/CreateTask",
+            status_code=200,
+            json={"id": 42, "title": "Approve invoice"},
+        )
+
+        task = await service.create_quickform_async(
+            title="Approve invoice",
+            task_schema_key="schema-uuid",
+            schema={"fields": []},
+        )
+
+        assert isinstance(task, Task)
+        assert task.id == 42
+
+        sent = httpx_mock.get_request()
+        assert sent is not None
+        assert sent.method == "POST"
+        assert (
+            str(sent.url)
+            == f"{base_url}{org}{tenant}/orchestrator_/tasks/GenericTasks/CreateTask"
+        )
+        body = json.loads(sent.content)
+        assert body["type"] == "QuickFormTask"
+        assert body["taskSchemaKey"] == "schema-uuid"
+        assert body["schema"] == {"fields": []}
+
+    @pytest.mark.anyio
+    async def test_assigns_after_create_when_recipient_provided(
+        self,
+        httpx_mock: HTTPXMock,
+        service: TasksService,
+        base_url: str,
+        org: str,
+        tenant: str,
+    ) -> None:
+        httpx_mock.add_response(
+            url=f"{base_url}{org}{tenant}/orchestrator_/tasks/GenericTasks/CreateTask",
+            status_code=200,
+            json={"id": 99, "title": "t"},
+        )
+        httpx_mock.add_response(
+            url=f"{base_url}{org}{tenant}/orchestrator_/odata/Tasks/UiPath.Server.Configuration.OData.AssignTasks",
+            status_code=200,
+            json={},
+        )
+
+        task = await service.create_quickform_async(
+            title="t",
+            task_schema_key="k",
+            recipient=TaskRecipient(
+                type=TaskRecipientType.EMAIL,
+                value="approver@example.com",
+            ),
+        )
+
+        assert isinstance(task, Task)
+        assert task.id == 99
+
+        requests = httpx_mock.get_requests()
+        assert len(requests) == 2
+
+        assign_request = requests[1]
+        assign_body = json.loads(assign_request.content)
+        assert assign_body == {
+            "taskAssignments": [
+                {
+                    "taskId": 99,
+                    "assignmentCriteria": "SingleUser",
+                    "userNameOrEmail": "approver@example.com",
+                }
+            ]
+        }
+
+    @pytest.mark.anyio
+    async def test_skips_assign_when_no_recipient(
+        self,
+        httpx_mock: HTTPXMock,
+        service: TasksService,
+        base_url: str,
+        org: str,
+        tenant: str,
+    ) -> None:
+        httpx_mock.add_response(
+            url=f"{base_url}{org}{tenant}/orchestrator_/tasks/GenericTasks/CreateTask",
+            status_code=200,
+            json={"id": 1, "title": "t"},
+        )
+
+        await service.create_quickform_async(title="t", task_schema_key="k")
+
+        requests = httpx_mock.get_requests()
+        assert len(requests) == 1
+        assert "CreateTask" in str(requests[0].url)