feat(agent): add tool-output and V2 escalation recipient types [ACTN-10480]

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

@@ -233,6 +233,28 @@ async def _assign_task_spec(
                     }
                 ]
             }
+        elif task_recipient.type == TaskRecipientType.WORKLOAD:
+            request_spec.json = {
+                "taskAssignments": [
+                    {
+                        "taskId": task_key,
+                        "assignmentCriteria": "Workload",
+                        "assigneeNamesOrEmails": task_recipient.values
+                        or [recipient_value],
+                    }
+                ]
+            }
+        elif task_recipient.type == TaskRecipientType.ROUND_ROBIN:
+            request_spec.json = {
+                "taskAssignments": [
+                    {
+                        "taskId": task_key,
+                        "assignmentCriteria": "RoundRobin",
+                        "assigneeNamesOrEmails": task_recipient.values
+                        or [recipient_value],
+                    }
+                ]
+            }
         else:
             request_spec.json = {
                 "taskAssignments": [

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

@@ -22,18 +22,29 @@ class TaskRecipientType(str, enum.Enum):
     GROUP_ID = "GroupId"
     EMAIL = "UserEmail"
     GROUP_NAME = "GroupName"
+    WORKLOAD = "Workload"
+    ROUND_ROBIN = "RoundRobin"
 
 
 class TaskRecipient(BaseModel):
-    """Model representing a task recipient."""
+    """Model representing a task recipient.
+
+    `value` is the single identifier (group name, group id, user id, email, …).
+    `values` is the multi-assignee form used by Workload-with-custom-emails
+    assignments; when set it takes precedence over `value` for the
+    `assigneeNamesOrEmails` payload.
+    """
 
     type: Literal[
         TaskRecipientType.USER_ID,
         TaskRecipientType.GROUP_ID,
         TaskRecipientType.EMAIL,
         TaskRecipientType.GROUP_NAME,
+        TaskRecipientType.WORKLOAD,
+        TaskRecipientType.ROUND_ROBIN,
     ] = Field(..., alias="type")
     value: str = Field(..., alias="value")
+    values: Optional[List[str]] = Field(default=None, alias="values")
     display_name: Optional[str] = Field(default=None, alias="displayName")
 
 

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

@@ -3,9 +3,12 @@
 import pytest
 from pytest_httpx import HTTPXMock
 
+import json
+
 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 import TaskRecipient, TaskRecipientType
 from uipath.platform.common.constants import HEADER_USER_AGENT
 
 
@@ -185,6 +188,167 @@ def test_create_with_assignee(
         assert action.title == "Test Action"
 
 
+def _mock_app_lookup_and_create(
+    httpx_mock: HTTPXMock,
+    base_url: str,
+    org: str,
+    tenant: str,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Common httpx mock setup for app lookup + task creation + assign."""
+    monkeypatch.setenv("UIPATH_TENANT_ID", "test-tenant-id")
+    httpx_mock.add_response(
+        url=f"{base_url}{org}/apps_/default/api/v1/default/deployed-action-apps-schemas?search=test-app&filterByDeploymentTitle=true",
+        status_code=200,
+        json={
+            "deployed": [
+                {
+                    "systemName": "test-app",
+                    "deploymentTitle": "test-app",
+                    "actionSchema": {
+                        "key": "test-key",
+                        "inputs": [],
+                        "outputs": [],
+                        "inOuts": [],
+                        "outcomes": [],
+                    },
+                    "deploymentFolder": {
+                        "fullyQualifiedName": "test-folder-path",
+                        "key": "test-folder-key",
+                    },
+                }
+            ]
+        },
+    )
+    httpx_mock.add_response(
+        url=f"{base_url}{org}{tenant}/orchestrator_/tasks/AppTasks/CreateAppTask",
+        status_code=200,
+        json={"id": 1, "title": "Test Action"},
+    )
+    httpx_mock.add_response(
+        url=f"{base_url}{org}{tenant}/orchestrator_/odata/Tasks/UiPath.Server.Configuration.OData.AssignTasks",
+        status_code=200,
+        json={},
+    )
+
+
+def _assign_request_payload(httpx_mock: HTTPXMock) -> dict[str, Any]:
+    """Return the parsed JSON body of the last AssignTasks request captured by the mock."""
+    assign_request = next(
+        req
+        for req in reversed(httpx_mock.get_requests())
+        if "AssignTasks" in str(req.url)
+    )
+    return json.loads(assign_request.content)
+
+
+class TestAssignTaskSpec:
+    """Tests for the task-assignment payload built by `_assign_task_spec`."""
+
+    def test_assign_workload_recipient_uses_workload_criteria_with_group(
+        self,
+        httpx_mock: HTTPXMock,
+        service: TasksService,
+        base_url: str,
+        org: str,
+        tenant: str,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        _mock_app_lookup_and_create(httpx_mock, base_url, org, tenant, monkeypatch)
+
+        service.create(
+            title="Test Action",
+            app_name="test-app",
+            data={"x": 1},
+            recipient=TaskRecipient(
+                type=TaskRecipientType.WORKLOAD,
+                value="Support Team",
+                displayName="Support Team",
+            ),
+        )
+
+        payload = _assign_request_payload(httpx_mock)
+        assert payload == {
+            "taskAssignments": [
+                {
+                    "taskId": 1,
+                    "assignmentCriteria": "Workload",
+                    "assigneeNamesOrEmails": ["Support Team"],
+                }
+            ]
+        }
+
+    def test_assign_round_robin_recipient_uses_round_robin_criteria(
+        self,
+        httpx_mock: HTTPXMock,
+        service: TasksService,
+        base_url: str,
+        org: str,
+        tenant: str,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        _mock_app_lookup_and_create(httpx_mock, base_url, org, tenant, monkeypatch)
+
+        service.create(
+            title="Test Action",
+            app_name="test-app",
+            data={"x": 1},
+            recipient=TaskRecipient(
+                type=TaskRecipientType.ROUND_ROBIN,
+                value="Support Team",
+                displayName="Support Team",
+            ),
+        )
+
+        payload = _assign_request_payload(httpx_mock)
+        assert payload == {
+            "taskAssignments": [
+                {
+                    "taskId": 1,
+                    "assignmentCriteria": "RoundRobin",
+                    "assigneeNamesOrEmails": ["Support Team"],
+                }
+            ]
+        }
+
+    def test_assign_workload_with_multiple_emails_uses_values_list(
+        self,
+        httpx_mock: HTTPXMock,
+        service: TasksService,
+        base_url: str,
+        org: str,
+        tenant: str,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """Custom-assignees path: Workload criteria with a list of emails."""
+        _mock_app_lookup_and_create(httpx_mock, base_url, org, tenant, monkeypatch)
+
+        service.create(
+            title="Test Action",
+            app_name="test-app",
+            data={"x": 1},
+            recipient=TaskRecipient(
+                type=TaskRecipientType.WORKLOAD,
+                value="alice@example.com",
+                values=["alice@example.com", "bob@example.com"],
+            ),
+        )
+
+        payload = _assign_request_payload(httpx_mock)
+        assert payload == {
+            "taskAssignments": [
+                {
+                    "taskId": 1,
+                    "assignmentCriteria": "Workload",
+                    "assigneeNamesOrEmails": [
+                        "alice@example.com",
+                        "bob@example.com",
+                    ],
+                }
+            ]
+        }
+
+
 def _make_deployed_app(
     name: str,
     folder_path: str,

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

@@ -137,6 +137,9 @@ class AgentEscalationRecipientType(str, CaseInsensitiveEnum):
     ASSET_GROUP_NAME = "AssetGroupName"
     ARGUMENT_EMAIL = "ArgumentEmail"
     ARGUMENT_GROUP_NAME = "ArgumentGroupName"
+    WORKLOAD = "Workload"
+    ROUND_ROBIN = "RoundRobin"
+    CUSTOM_ASSIGNEES = "CustomAssignees"
 
 
 class AgentContextRetrievalMode(str, CaseInsensitiveEnum):
@@ -491,6 +494,9 @@ class AgentA2aResourceConfig(BaseAgentResourceConfig):
     6: AgentEscalationRecipientType.ASSET_GROUP_NAME,
     7: AgentEscalationRecipientType.ARGUMENT_EMAIL,
     8: AgentEscalationRecipientType.ARGUMENT_GROUP_NAME,
+    9: AgentEscalationRecipientType.WORKLOAD,
+    10: AgentEscalationRecipientType.ROUND_ROBIN,
+    11: AgentEscalationRecipientType.CUSTOM_ASSIGNEES,
 }
 
 
@@ -570,14 +576,86 @@ class ArgumentGroupNameRecipient(BaseEscalationRecipient):
     argument_path: str = Field(..., alias="argumentName")
 
 
+class WorkloadRecipient(BaseEscalationRecipient):
+    """Workload-based group assignment.
+
+    The Action Center distributes tasks to the group member with the lightest workload.
+    """
+
+    type: Literal[AgentEscalationRecipientType.WORKLOAD,] = Field(..., alias="type")
+    value: str = Field(..., alias="value")
+    display_name: str = Field(..., alias="displayName")
+
+
+class RoundRobinRecipient(BaseEscalationRecipient):
+    """Round-robin group assignment.
+
+    The Action Center cycles through group members in order on each new task.
+    """
+
+    type: Literal[AgentEscalationRecipientType.ROUND_ROBIN,] = Field(..., alias="type")
+    value: str = Field(..., alias="value")
+    display_name: str = Field(..., alias="displayName")
+
+
+class CustomAssigneesRecipient(BaseEscalationRecipient):
+    """Custom multi-user assignment.
+
+    A channel can carry multiple instances, one per assignee email. All are passed
+    to Action Center together using a Workload assignment criteria.
+    """
+
+    type: Literal[AgentEscalationRecipientType.CUSTOM_ASSIGNEES,] = Field(
+        ..., alias="type"
+    )
+    value: str = Field(..., alias="value")
+    display_name: Optional[str] = Field(default=None, alias="displayName")
+
+
+class ToolOutputRecipient(BaseEscalationRecipient):
+    """Recipient whose value is resolved at runtime from a named tool's output.
+
+    Instead of a literal value entered at design time, this binding points at a
+    field within a named tool's output. The runtime walks the agent's message
+    history, finds the most recent ToolMessage matching `tool_name`, parses its
+    content as JSON, and extracts `output_path` (a top-level field for v1).
+
+    Only the assignment-criteria recipient types that accept a runtime-computed
+    value are supported: USER_ID, GROUP_ID, WORKLOAD, ROUND_ROBIN,
+    CUSTOM_ASSIGNEES. The asset/static/argument types do not participate in
+    tool-output binding (they have their own design-time resolution rules).
+    """
+
+    type: Literal[
+        AgentEscalationRecipientType.USER_ID,
+        AgentEscalationRecipientType.GROUP_ID,
+        AgentEscalationRecipientType.WORKLOAD,
+        AgentEscalationRecipientType.ROUND_ROBIN,
+        AgentEscalationRecipientType.CUSTOM_ASSIGNEES,
+    ] = Field(..., alias="type")
+    source: Literal["toolOutput"] = Field(..., alias="source")
+    tool_name: str = Field(..., alias="toolName")
+    output_path: str = Field(..., alias="outputPath")
+
+
+# Note: order matters in this union — ToolOutputRecipient is listed first so payloads
+# carrying `source: "toolOutput"` match it before the literal variants get a chance.
+# The literal classes don't define a `source` field, so Pydantic's overlap heuristics
+# pick the right class via the presence of required fields (value/displayName vs
+# source/toolName/outputPath). A `Field(discriminator="type")` cannot be used here
+# because multiple classes share the same `type` literals (literal and tool-output
+# variants of the same criteria).
 AgentEscalationRecipient = Annotated[
     Union[
+        ToolOutputRecipient,
         StandardRecipient,
         AssetRecipient,
         ArgumentEmailRecipient,
         ArgumentGroupNameRecipient,
+        WorkloadRecipient,
+        RoundRobinRecipient,
+        CustomAssigneesRecipient,
     ],
-    Field(discriminator="type"),
     BeforeValidator(_normalize_recipient_type),
 ]