feat(tasks): add QuickForm task creation SDK + agent.json model

packages/uipath-platform/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-platform"
-version = "0.1.59"
+version = "0.1.60"
 description = "HTTP client library for programmatic access to UiPath Platform"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

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

@@ -196,6 +196,94 @@
     return normalized
 
 
+# TaskType.QuickFormTask value, matching the Orchestrator enum (UiPath.Orchestrator.DataContracts.TaskType).
+_TASK_TYPE_QUICKFORM = 6
+
+
+def _create_quickform_spec(
+    data: Optional[Dict[str, Any]],
+    title: str,
+    task_schema_key: str,
+    schema: Dict[str, Any],
+    creator_job_key: Optional[str] = None,
+    folder_key: Optional[str] = None,
+    folder_path: 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:
+    """Build the RequestSpec for Orchestrator's GenericTasks/CreateTask endpoint.
+
+    Sets TaskType=QuickFormTask. Mirrors _create_spec but skips the AppTask-specific
+    shape (no appId, no action-schema-derived fieldSet/actionSet) and instead sends
+    taskSchemaKey + inline schema together.
+
+    Both taskSchemaKey AND schema are sent on every call: the Agents runtime has no
+    Action Center package.uploaded subscriber populating the TaskSchemas table, so
+    Orchestrator upserts the schema (keyed by taskSchemaKey) and then creates the task
+    in the same call.
+
+    Wire contract: UiPath/Orchestrator/src/Core/Application/Dto/Tasks/TaskCreateRequest.cs.
+    """
+    json_payload: Dict[str, Any] = {
+        "type": _TASK_TYPE_QUICKFORM,
+        "taskSchemaKey": task_schema_key,
+        "schema": schema,
+        "title": title,
+        "data": data if data is not None else {},
+    }
+
+    if creator_job_key is not None:
+        json_payload["creatorJobKey"] = creator_job_key
+
+    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:
+        cfg_folder_key = UiPathConfig.folder_key
+        job_key = UiPathConfig.job_key
+        process_key = UiPathConfig.process_uuid
+
+        task_source_metadata: Dict[str, Any] = {
+            "InstanceId": trace_id,
+            "FolderKey": cfg_folder_key,
+            "JobKey": job_key,
+            "ProcessKey": process_key,
+        }
+
+        json_payload["taskSource"] = {
+            "sourceName": source_name,
+            "sourceId": project_id,
+            "taskSourceMetadata": task_source_metadata,
+        }
+
+    return RequestSpec(
+        method="POST",
+        endpoint=Endpoint("/orchestrator_/tasks/GenericTasks/CreateTask"),
+        json=json_payload,
+        headers=header_folder(folder_key, folder_path),
+    )
+
+
 def _retrieve_action_spec(
     action_key: str,
     app_folder_key: Optional[str],
@@ -506,6 +594,154 @@
             )
         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: Dict[str, Any],
+        data: Optional[Dict[str, Any]] = None,
+        *,
+        folder_path: Optional[str] = None,
+        folder_key: Optional[str] = None,
+        assignee: 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,
+        creator_job_key: Optional[str] = None,
+        source_name: str = "Agent",
+    ) -> Task:
+        """Creates a new QuickForm task asynchronously.
+
+        QuickForm tasks are schema-first HITL tasks rendered by FormLib in Action
+        Center. Both task_schema_key AND schema are required: the Agents runtime
+        does not pre-populate TaskSchemas via a package.uploaded subscriber, so
+        Orchestrator upserts the schema (keyed by task_schema_key) and creates
+        the task in the same call.
+
+        Args:
+            title: The title of the task.
+            task_schema_key: UUID key of the schema. Used as the key under which
+                Orchestrator stores/looks up the schema in TaskSchemas.
+            schema: The HITL schema body to register/upsert. Sent inline on every
+                call.
+            data: Optional dictionary containing input data for the task.
+            folder_path: Optional folder path for the task. Required by the
+                Orchestrator controller (RequireOrganizationUnit) unless
+                folder_key is provided.
+            folder_key: Optional folder key, alternative to folder_path.
+            assignee: Optional username or email to assign the task to.
+            recipient: Optional structured recipient (user id / group id /
+                email). Resolved via identity service before assignment.
+            priority: Optional priority. Low / Medium / High / Critical.
+            labels: Optional list of labels for the task.
+            is_actionable_message_enabled: Whether actionable notifications are
+                enabled for this task.
+            actionable_message_metadata: Optional metadata override. For
+                QuickForm, when null, Orchestrator derives it from the
+                referenced TaskSchema.
+            creator_job_key: Optional. Identifies the job that triggered the
+                inline schema creation/upsert.
+            source_name: Source name on TaskSource. Defaults to 'Agent'.
+
+        Returns:
+            Task: The created task object.
+        """
+        spec = _create_quickform_spec(
+            title=title,
+            data=data,
+            task_schema_key=task_schema_key,
+            schema=schema,
+            creator_job_key=creator_job_key,
+            folder_key=folder_key,
+            folder_path=folder_path,
+            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,
+            content=spec.content,
+            headers=spec.headers,
+        )
+        json_response = response.json()
+        if assignee or recipient:
+            assign_spec = await _assign_task_spec(
+                self, json_response["id"], assignee, 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)
+
+    @traced(name="tasks_create_quickform", run_type="uipath")
+    def create_quickform(
+        self,
+        title: str,
+        task_schema_key: str,
+        schema: Dict[str, Any],
+        data: Optional[Dict[str, Any]] = None,
+        *,
+        folder_path: Optional[str] = None,
+        folder_key: Optional[str] = None,
+        assignee: 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,
+        creator_job_key: Optional[str] = None,
+        source_name: str = "Agent",
+    ) -> Task:
+        """Create a new QuickForm task synchronously.
+
+        See :meth:`create_quickform_async` for parameter docs.
+        """
+        spec = _create_quickform_spec(
+            title=title,
+            data=data,
+            task_schema_key=task_schema_key,
+            schema=schema,
+            creator_job_key=creator_job_key,
+            folder_key=folder_key,
+            folder_path=folder_path,
+            priority=priority,
+            labels=labels,
+            is_actionable_message_enabled=is_actionable_message_enabled,
+            actionable_message_metadata=actionable_message_metadata,
+            source_name=source_name,
+        )
+
+        response = self.request(
+            spec.method,
+            spec.endpoint,
+            json=spec.json,
+            content=spec.content,
+            headers=spec.headers,
+        )
+        json_response = response.json()
+        if assignee or recipient:
+            assign_spec = asyncio.run(
+                _assign_task_spec(self, json_response["id"], assignee, recipient)
+            )
+            self.request(
+                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/pyproject.toml

@@ -1,13 +1,13 @@
 [project]
 name = "uipath"
-version = "2.10.74"
+version = "2.10.75"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"
 dependencies = [
   "uipath-core>=0.5.17, <0.6.0",
   "uipath-runtime>=0.11.0, <0.12.0",
-  "uipath-platform>=0.1.59, <0.2.0",
+  "uipath-platform>=0.1.60, <0.2.0",
   "click>=8.3.1",
   "httpx>=0.28.1",
   "pyjwt>=2.10.1",

packages/uipath/src/uipath/agent/models/agent.py

@@ -726,6 +726,14 @@ class AgentEscalationChannel(BaseCfg):
     )
     priority: Optional[str] = None
     labels: List[str] = Field(default_factory=list)
+    # QuickForm fields — set only on channels backed by a .hitl.json schema.
+    # schema_id is the UUID key under which the schema is registered in Orchestrator's
+    # TaskSchemas table. schema_body is the inline schema body, sent on every task creation
+    # so Orchestrator can upsert it (the Agents runtime has no separate registration step).
+    # The Python attribute is schema_body (not schema) to avoid shadowing
+    # pydantic.BaseModel.schema(); the JSON alias is "schema" to match the wire format.
+    schema_id: Optional[str] = Field(None, alias="schemaId")
+    schema_body: Optional[Dict[str, Any]] = Field(None, alias="schema")
 
     @model_validator(mode="before")
     @classmethod
@@ -769,6 +777,23 @@ class AgentIxpVsEscalationResourceConfig(BaseAgentResourceConfig):
     )
 
 
+class AgentQuickFormEscalationResourceConfig(BaseAgentResourceConfig):
+    """Quick Form Agent escalation resource configuration model (escalationType=2).
+
+    Quick Form escalations render a schema-first HITL task in Action Center via FormLib.
+    The schema (and its key) live on the channel (see AgentEscalationChannel.schema_id /
+    schema) and are sent inline to Orchestrator's GenericTasks/CreateTask endpoint.
+    """
+
+    id: Optional[str] = Field(None, alias="id")
+    resource_type: Literal[AgentResourceType.ESCALATION] = Field(
+        alias="$resourceType", default=AgentResourceType.ESCALATION, frozen=True
+    )
+    channels: List[AgentEscalationChannel] = Field(alias="channels")
+    is_agent_memory_enabled: bool = Field(default=False, alias="isAgentMemoryEnabled")
+    escalation_type: Literal[2] = Field(default=2, alias="escalationType")
+
+
 class BaseAgentToolResourceConfig(BaseAgentResourceConfig):
     """Base agent tool resource configuration model."""
 
@@ -978,6 +1003,7 @@ class AgentUnknownToolResourceConfig(BaseAgentToolResourceConfig):
     Union[
         Annotated[AgentEscalationResourceConfig, Tag(0)],
         Annotated[AgentIxpVsEscalationResourceConfig, Tag(1)],
+        Annotated[AgentQuickFormEscalationResourceConfig, Tag(2)],
     ],
     Discriminator(lambda v: v.get("escalation_type") or v.get("escalationType") or 0),
 ]