diff --git a/packages/uipath-platform/src/uipath/platform/common/__init__.py b/packages/uipath-platform/src/uipath/platform/common/__init__.py index cefd92075..f97efd513 100644 --- a/packages/uipath-platform/src/uipath/platform/common/__init__.py +++ b/packages/uipath-platform/src/uipath/platform/common/__init__.py @@ -27,6 +27,14 @@ from ._user_agent import user_agent_value from .auth import TokenData from .dynamic_schema import jsonschema_to_pydantic +from ..hitl import ( + HitlFieldDirection, + HitlFieldType, + HitlSchema, + HitlSchemaField, + HitlSchemaOutcome, + pydantic_to_hitl_schema, +) from .interrupt_models import ( CreateBatchTransform, CreateDeepRag, @@ -102,6 +110,12 @@ "validate_pagination_params", "EndpointManager", "jsonschema_to_pydantic", + "HitlFieldDirection", + "HitlFieldType", + "HitlSchema", + "HitlSchemaField", + "HitlSchemaOutcome", + "pydantic_to_hitl_schema", "ConnectionResourceOverwrite", "EntityResourceOverwrite", "GenericResourceOverwrite", diff --git a/packages/uipath-platform/src/uipath/platform/common/interrupt_models.py b/packages/uipath-platform/src/uipath/platform/common/interrupt_models.py index 3b2468551..dfe2ba290 100644 --- a/packages/uipath-platform/src/uipath/platform/common/interrupt_models.py +++ b/packages/uipath-platform/src/uipath/platform/common/interrupt_models.py @@ -10,6 +10,7 @@ from ..action_center.tasks import Task, TaskRecipient from ..attachments import Attachment +from ..hitl.models import HitlSchema from ..context_grounding import ( BatchTransformCreationResponse, BatchTransformOutputColumn, @@ -58,7 +59,12 @@ class WaitJobRaw(WaitJob): class CreateTask(BaseModel): - """Model representing an action creation.""" + """Model representing an action creation. + + When *schema* is set the runtime creates a schema-driven **QuickForm** task + (``GenericTasks/CreateTask``, ``type=6``) instead of an action-app task. + No pre-deployed Action App is required in that case. + """ title: str data: dict[str, Any] | None = None @@ -73,6 +79,7 @@ class CreateTask(BaseModel): is_actionable_message_enabled: bool | None = None actionable_message_metadata: dict[str, Any] | None = None source_name: str = "Agent" + hitl_schema: HitlSchema | None = None class CreateEscalation(CreateTask): diff --git a/packages/uipath-platform/src/uipath/platform/hitl/__init__.py b/packages/uipath-platform/src/uipath/platform/hitl/__init__.py new file mode 100644 index 000000000..dda65c70f --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/hitl/__init__.py @@ -0,0 +1,47 @@ +"""HITL form schema types and generator. + +Provides Python-native representations of the HITL schema format and a utility +to generate schemas from Pydantic models. The generated schema can be attached +to :class:`~uipath.platform.common.CreateTask` or +:class:`~uipath.platform.common.CreateEscalation` so the runtime creates a +QuickForm task with an inline schema instead of requiring a pre-deployed +Action App. + +Example:: + + from uipath.platform.hitl import HitlSchema, pydantic_to_hitl_schema + from pydantic import BaseModel, Field + + class ReviewInputs(BaseModel): + flagged_content: str = Field(title="Flagged Content") + reason: str = Field(title="Reason") + + class ReviewOutputs(BaseModel): + decision: str = Field(default="", title="Decision") + notes: str = Field(default="", title="Notes") + + schema = pydantic_to_hitl_schema( + input_model=ReviewInputs, + output_model=ReviewOutputs, + outcomes=["Approve", "Reject"], + title="Content Review", + ) +""" + +from .models import ( + HitlFieldDirection, + HitlFieldType, + HitlSchema, + HitlSchemaField, + HitlSchemaOutcome, +) +from .schema_gen import pydantic_to_hitl_schema + +__all__ = [ + "HitlFieldDirection", + "HitlFieldType", + "HitlSchema", + "HitlSchemaField", + "HitlSchemaOutcome", + "pydantic_to_hitl_schema", +] diff --git a/packages/uipath-platform/src/uipath/platform/hitl/models.py b/packages/uipath-platform/src/uipath/platform/hitl/models.py new file mode 100644 index 000000000..0f06107cb --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/hitl/models.py @@ -0,0 +1,76 @@ +"""Python representation of the HITL form schema format. + +Mirrors the TypeScript ``HitlSchema`` interface from ``@uipath/hitl-schema-types`` +and is compatible with the QuickForm task endpoint +(``GenericTasks/CreateTask``, ``type=6``). +""" + +from enum import Enum +from typing import Any + +from pydantic import BaseModel, Field + + +class HitlFieldType(str, Enum): + """Types supported by HITL form fields.""" + + STRING = "string" + NUMBER = "number" + INTEGER = "integer" + FLOAT = "float" + DOUBLE = "double" + BOOLEAN = "boolean" + DATE = "date" + DATETIME = "datetime" + FILE = "file" + OBJECT = "object" + ARRAY = "array" + + +class HitlFieldDirection(str, Enum): + """Direction of a HITL form field. + + - ``INPUT`` — read-only display, pre-populated from task data. + - ``OUTPUT`` — editable by the human reviewer, written to a variable on submit. + - ``IN_OUT`` — both pre-populated and editable. + """ + + INPUT = "input" + OUTPUT = "output" + IN_OUT = "inOut" + + +class HitlSchemaField(BaseModel): + """A single field in a HITL form schema.""" + + id: str + label: str | None = None + type: HitlFieldType = HitlFieldType.STRING + direction: HitlFieldDirection = HitlFieldDirection.INPUT + required: bool | None = None + + +class HitlSchemaOutcome(BaseModel): + """An outcome button in a HITL form schema (e.g. Approve / Reject).""" + + id: str + label: str | None = None + + +class HitlSchema(BaseModel): + """A dynamic HITL form schema generated from Python type annotations. + + Attach to :class:`~uipath.platform.common.CreateTask` or + :class:`~uipath.platform.common.CreateEscalation` so the runtime creates a + schema-driven QuickForm task instead of an action-app task. No pre-deployed + Action App is required when a schema is provided. + """ + + id: str | None = None + title: str | None = None + fields: list[HitlSchemaField] = Field(default_factory=list) + outcomes: list[HitlSchemaOutcome] = Field(default_factory=list) + + def to_wire_format(self) -> dict[str, Any]: + """Serialise to the dict expected by the QuickForm task API.""" + return self.model_dump(mode="json", exclude_none=True) diff --git a/packages/uipath-platform/src/uipath/platform/hitl/schema_gen.py b/packages/uipath-platform/src/uipath/platform/hitl/schema_gen.py new file mode 100644 index 000000000..8b1ae54dd --- /dev/null +++ b/packages/uipath-platform/src/uipath/platform/hitl/schema_gen.py @@ -0,0 +1,143 @@ +"""Generate a :class:`HitlSchema` from Python type annotations. + +The primary entry point is :func:`pydantic_to_hitl_schema`, which converts +Pydantic ``BaseModel`` classes into the :class:`HitlSchema` format consumed by +Action Center's QuickForm task endpoint. +""" + +from __future__ import annotations + +import inspect +import re +from types import UnionType +from typing import Any, Union, get_args, get_origin + +from pydantic import BaseModel + +from .models import ( + HitlFieldDirection, + HitlFieldType, + HitlSchema, + HitlSchemaField, + HitlSchemaOutcome, +) + + +def _annotation_to_hitl_type(annotation: Any) -> HitlFieldType: + """Map a Python type annotation to the closest :class:`HitlFieldType`.""" + if annotation is None or annotation is inspect.Parameter.empty: + return HitlFieldType.STRING + + origin = get_origin(annotation) + args = get_args(annotation) + + # Handle Optional[T] / T | None (both typing.Union and PEP 604 UnionType) + if origin is Union or isinstance(annotation, UnionType): + non_none = [a for a in args if a is not type(None)] + if non_none: + return _annotation_to_hitl_type(non_none[0]) + return HitlFieldType.STRING + + if origin in (list, tuple): + return HitlFieldType.ARRAY + if origin is dict: + return HitlFieldType.OBJECT + + if not inspect.isclass(annotation): + return HitlFieldType.STRING + + # Check bool before int — bool is a subclass of int + if issubclass(annotation, bool): + return HitlFieldType.BOOLEAN + if issubclass(annotation, int): + return HitlFieldType.INTEGER + if issubclass(annotation, float): + return HitlFieldType.NUMBER + if issubclass(annotation, str): + return HitlFieldType.STRING + if issubclass(annotation, (list, tuple)): + return HitlFieldType.ARRAY + if issubclass(annotation, dict): + return HitlFieldType.OBJECT + if issubclass(annotation, BaseModel): + return HitlFieldType.OBJECT + + return HitlFieldType.STRING + + +def _label_from_name(name: str) -> str: + """Produce a human-readable label from a snake_case or PascalCase name.""" + # PascalCase → insert space before each capital that follows a lowercase letter + spaced = re.sub(r"(?<=[a-z])(?=[A-Z])", " ", name) + # Underscores → spaces + spaced = spaced.replace("_", " ") + return spaced.title() + + +def pydantic_to_hitl_schema( + *, + input_model: type[BaseModel] | None = None, + output_model: type[BaseModel] | None = None, + outcomes: list[str] | None = None, + title: str | None = None, +) -> HitlSchema: + """Generate a :class:`HitlSchema` from Pydantic input and output models. + + Fields from *input_model* become ``direction="input"`` fields (read-only, + pre-populated from task data). Fields from *output_model* become + ``direction="output"`` fields (editable by the human reviewer). + + The field ``id`` is taken from ``field_info.alias`` when one is set, + otherwise from the Python field name. Set an alias on a Pydantic field + to control the id that Action Center uses (useful when the form contract + requires PascalCase names but Python convention is snake_case). + + Args: + input_model: Pydantic model for data shown to the reviewer. + output_model: Pydantic model for data the reviewer fills in. + outcomes: Outcome button labels. Defaults to ``["Approve", "Reject"]``. + title: Human-readable title for the form. + + Returns: + A :class:`HitlSchema` ready to attach to + :class:`~uipath.platform.common.CreateTask`. + """ + if outcomes is None: + outcomes = ["Approve", "Reject"] + + fields: list[HitlSchemaField] = [] + + if input_model is not None: + for name, field_info in input_model.model_fields.items(): + field_id = field_info.alias or name + fields.append( + HitlSchemaField( + id=str(field_id), + label=field_info.title or _label_from_name(str(field_id)), + type=_annotation_to_hitl_type(field_info.annotation), + direction=HitlFieldDirection.INPUT, + required=True if field_info.is_required() else None, + ) + ) + + if output_model is not None: + for name, field_info in output_model.model_fields.items(): + field_id = field_info.alias or name + fields.append( + HitlSchemaField( + id=str(field_id), + label=field_info.title or _label_from_name(str(field_id)), + type=_annotation_to_hitl_type(field_info.annotation), + direction=HitlFieldDirection.OUTPUT, + required=None, + ) + ) + + return HitlSchema( + title=title, + fields=fields, + outcomes=[ + HitlSchemaOutcome(id=outcome.lower(), label=outcome) + for outcome in outcomes + ], + ) diff --git a/packages/uipath-platform/src/uipath/platform/resume_triggers/_protocol.py b/packages/uipath-platform/src/uipath/platform/resume_triggers/_protocol.py index b2dbae787..052afca25 100644 --- a/packages/uipath-platform/src/uipath/platform/resume_triggers/_protocol.py +++ b/packages/uipath-platform/src/uipath/platform/resume_triggers/_protocol.py @@ -1,5 +1,6 @@ """Implementation of UiPath resume trigger protocols.""" +import hashlib import json import os import uuid @@ -69,6 +70,17 @@ ) +def _schema_key(schema: Any) -> str: + """Return a deterministic UUID derived from the schema's wire format. + + The same schema content always maps to the same key so Orchestrator can + upsert rather than duplicate schema records. + """ + wire = json.dumps(schema.to_wire_format(), sort_keys=True) + digest = hashlib.sha256(wire.encode()).digest()[:16] + return str(uuid.UUID(bytes=digest)) + + def _try_convert_to_json_format(value: str | None) -> Any: """Attempts to parse a string as JSON and returns the parsed object or original string. @@ -625,21 +637,41 @@ async def _handle_task_trigger( resume_trigger.item_key = value.action.key elif isinstance(value, (CreateTask, CreateEscalation)): uipath = UiPath() - action = await uipath.tasks.create_async( - title=value.title, - app_name=value.app_name if value.app_name else "", - app_folder_path=value.app_folder_path or None, - app_folder_key=value.app_folder_key or None, - app_key=value.app_key if value.app_key else "", - assignee=value.assignee if value.assignee else "", - recipient=value.recipient if value.recipient else "", - data=value.data, - priority=value.priority, - labels=value.labels, - is_actionable_message_enabled=value.is_actionable_message_enabled, - actionable_message_metadata=value.actionable_message_metadata, - source_name=value.source_name, - ) + if getattr(value, "hitl_schema", None) is not None: + # Dynamic-schema path: create a QuickForm task with the inline schema. + # No pre-deployed Action App is required. + task_schema_key = _schema_key(value.hitl_schema) + action = await uipath.tasks.create_quickform_async( + title=value.title, + task_schema_key=task_schema_key, + schema=value.hitl_schema.to_wire_format(), + data=value.data, + folder_path=value.app_folder_path or None, + folder_key=value.app_folder_key or None, + assignee=value.assignee if value.assignee else None, + recipient=value.recipient if value.recipient else None, + priority=value.priority, + labels=value.labels, + is_actionable_message_enabled=value.is_actionable_message_enabled, + actionable_message_metadata=value.actionable_message_metadata, + source_name=value.source_name, + ) + else: + action = await uipath.tasks.create_async( + title=value.title, + app_name=value.app_name if value.app_name else "", + app_folder_path=value.app_folder_path or None, + app_folder_key=value.app_folder_key or None, + app_key=value.app_key if value.app_key else "", + assignee=value.assignee if value.assignee else "", + recipient=value.recipient if value.recipient else "", + data=value.data, + priority=value.priority, + labels=value.labels, + is_actionable_message_enabled=value.is_actionable_message_enabled, + actionable_message_metadata=value.actionable_message_metadata, + source_name=value.source_name, + ) if not action: raise Exception("Failed to create action") resume_trigger.item_key = action.key diff --git a/packages/uipath-platform/tests/services/test_hitl_schema.py b/packages/uipath-platform/tests/services/test_hitl_schema.py new file mode 100644 index 000000000..cf32eb3e2 --- /dev/null +++ b/packages/uipath-platform/tests/services/test_hitl_schema.py @@ -0,0 +1,204 @@ +"""Unit tests for the HITL dynamic schema types and generator.""" + +import json + +import pytest +from pydantic import BaseModel, Field + +from uipath.platform.common import CreateEscalation, CreateTask +from uipath.platform.hitl import ( + HitlFieldDirection, + HitlFieldType, + HitlSchema, + HitlSchemaField, + HitlSchemaOutcome, + pydantic_to_hitl_schema, +) +from uipath.platform.resume_triggers._protocol import _schema_key + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +class _Inputs(BaseModel): + text: str = Field(title="Text") + count: int = Field(default=0, title="Count") + active: bool = Field(default=True, title="Active") + + +class _Outputs(BaseModel): + decision: str = Field(default="", title="Decision") + notes: str | None = Field(default=None, title="Notes") + + +# --------------------------------------------------------------------------- +# HitlSchema wire format +# --------------------------------------------------------------------------- + + +def test_to_wire_format_serialises_enums_as_strings(): + schema = HitlSchema( + title="Test", + fields=[ + HitlSchemaField( + id="f1", + label="F1", + type=HitlFieldType.INTEGER, + direction=HitlFieldDirection.OUTPUT, + ) + ], + outcomes=[HitlSchemaOutcome(id="approve", label="Approve")], + ) + wire = schema.to_wire_format() + assert wire["fields"][0]["type"] == "integer" + assert wire["fields"][0]["direction"] == "output" + assert wire["outcomes"][0]["id"] == "approve" + + +def test_to_wire_format_excludes_none_fields(): + field = HitlSchemaField(id="x", type=HitlFieldType.STRING, direction=HitlFieldDirection.INPUT) + schema = HitlSchema(fields=[field]) + wire = schema.to_wire_format() + # 'label' and 'required' are None → excluded + assert "label" not in wire["fields"][0] + assert "required" not in wire["fields"][0] + # 'title' and 'id' at top level are None/missing → excluded + assert "title" not in wire + assert "id" not in wire + + +# --------------------------------------------------------------------------- +# pydantic_to_hitl_schema +# --------------------------------------------------------------------------- + + +def test_input_fields_have_input_direction(): + schema = pydantic_to_hitl_schema(input_model=_Inputs) + directions = {f.id: f.direction for f in schema.fields} + assert all(d == HitlFieldDirection.INPUT for d in directions.values()) + + +def test_output_fields_have_output_direction(): + schema = pydantic_to_hitl_schema(output_model=_Outputs) + directions = {f.id: f.direction for f in schema.fields} + assert all(d == HitlFieldDirection.OUTPUT for d in directions.values()) + + +def test_field_ids_match_python_names(): + schema = pydantic_to_hitl_schema(input_model=_Inputs) + ids = {f.id for f in schema.fields} + assert ids == {"text", "count", "active"} + + +def test_field_ids_use_alias_when_set(): + class AliasModel(BaseModel): + guardrail_name: str = Field(alias="GuardrailName", default="") + + schema = pydantic_to_hitl_schema(input_model=AliasModel) + assert schema.fields[0].id == "GuardrailName" + + +def test_type_mapping(): + schema = pydantic_to_hitl_schema(input_model=_Inputs) + types = {f.id: f.type for f in schema.fields} + assert types["text"] == HitlFieldType.STRING + assert types["count"] == HitlFieldType.INTEGER + assert types["active"] == HitlFieldType.BOOLEAN + + +def test_optional_field_maps_to_inner_type(): + schema = pydantic_to_hitl_schema(output_model=_Outputs) + types = {f.id: f.type for f in schema.fields} + # notes: str | None → STRING + assert types["notes"] == HitlFieldType.STRING + + +def test_default_outcomes(): + schema = pydantic_to_hitl_schema(input_model=_Inputs) + assert [o.id for o in schema.outcomes] == ["approve", "reject"] + assert [o.label for o in schema.outcomes] == ["Approve", "Reject"] + + +def test_custom_outcomes(): + schema = pydantic_to_hitl_schema(outcomes=["Yes", "No", "Maybe"]) + assert [o.id for o in schema.outcomes] == ["yes", "no", "maybe"] + + +def test_title_propagates(): + schema = pydantic_to_hitl_schema(title="My Form") + assert schema.title == "My Form" + + +def test_combined_input_and_output(): + schema = pydantic_to_hitl_schema(input_model=_Inputs, output_model=_Outputs) + input_ids = {f.id for f in schema.fields if f.direction == HitlFieldDirection.INPUT} + output_ids = {f.id for f in schema.fields if f.direction == HitlFieldDirection.OUTPUT} + assert input_ids == {"text", "count", "active"} + assert output_ids == {"decision", "notes"} + + +def test_required_field_is_true_for_mandatory(): + # 'text' has no default → required + schema = pydantic_to_hitl_schema(input_model=_Inputs) + field_map = {f.id: f for f in schema.fields} + assert field_map["text"].required is True + # 'count' has default → not required (None in schema) + assert field_map["count"].required is None + + +def test_output_fields_never_required(): + schema = pydantic_to_hitl_schema(output_model=_Outputs) + for f in schema.fields: + assert f.required is None + + +# --------------------------------------------------------------------------- +# CreateTask.hitl_schema field +# --------------------------------------------------------------------------- + + +def test_create_task_accepts_hitl_schema(): + schema = pydantic_to_hitl_schema(input_model=_Inputs) + task = CreateTask(title="T", hitl_schema=schema) + assert task.hitl_schema is schema + + +def test_create_task_hitl_schema_defaults_to_none(): + task = CreateTask(title="T") + assert task.hitl_schema is None + + +def test_create_escalation_inherits_hitl_schema(): + schema = pydantic_to_hitl_schema(input_model=_Inputs) + escalation = CreateEscalation(title="E", hitl_schema=schema) + assert escalation.hitl_schema is schema + + +# --------------------------------------------------------------------------- +# _schema_key determinism +# --------------------------------------------------------------------------- + + +def test_schema_key_is_deterministic(): + schema = pydantic_to_hitl_schema(input_model=_Inputs, title="Test") + key1 = _schema_key(schema) + key2 = _schema_key(schema) + assert key1 == key2 + + +def test_different_schemas_produce_different_keys(): + schema_a = pydantic_to_hitl_schema(input_model=_Inputs, title="A") + schema_b = pydantic_to_hitl_schema(input_model=_Inputs, title="B") + assert _schema_key(schema_a) != _schema_key(schema_b) + + +def test_schema_key_is_valid_uuid_format(): + import uuid + + schema = pydantic_to_hitl_schema(input_model=_Inputs) + key = _schema_key(schema) + # Should not raise + parsed = uuid.UUID(key) + assert str(parsed) == key