feat: extend invoke process with attachments support (#1221)

Author: cotovanu-cristian

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.6.17"
+version = "2.6.18"
 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"

src/uipath/platform/attachments/attachments.py

@@ -3,7 +3,7 @@
 import uuid
 from dataclasses import dataclass
 from enum import Enum
-from typing import Optional
+from typing import Any, Optional
 
 from pydantic import BaseModel, Field
 
@@ -18,10 +18,10 @@ class AttachmentMode(str, Enum):
 class Attachment(BaseModel):
     """Model representing an attachment. Id 'None' is used for uploads."""
 
-    id: Optional[uuid.UUID] = Field(None, alias="ID")
+    id: uuid.UUID = Field(..., alias="ID")
     full_name: str = Field(..., alias="FullName")
     mime_type: str = Field(..., alias="MimeType")
-    metadata: Optional[dict[str, str]] = Field(None, alias="Metadata")
+    metadata: Optional[dict[str, Any]] = Field(None, alias="Metadata")
     model_config = {
         "title": "UiPathAttachment",
         "validate_by_name": True,

src/uipath/platform/common/interrupt_models.py

@@ -9,6 +9,7 @@
 )
 
 from ..action_center.tasks import Task, TaskRecipient
+from ..attachments import Attachment
 from ..context_grounding import (
     BatchTransformCreationResponse,
     BatchTransformOutputColumn,
@@ -33,6 +34,7 @@ class InvokeProcess(BaseModel):
     process_folder_path: str | None = None
     process_folder_key: str | None = None
     input_arguments: dict[str, Any] | None
+    attachments: list[Attachment] | None = None
 
 
 class WaitJob(BaseModel):

src/uipath/platform/orchestrator/_processes_service.py

@@ -6,6 +6,7 @@
 from ..._utils import Endpoint, RequestSpec, header_folder, resource_override
 from ..._utils.constants import ENV_JOB_KEY, HEADER_JOB_KEY
 from ...tracing import traced
+from ..attachments import Attachment
 from ..common import BaseService, FolderContext, UiPathApiConfig, UiPathExecutionContext
 from ._attachments_service import AttachmentsService
 from .job import Job
@@ -19,6 +20,8 @@ class ProcessesService(FolderContext, BaseService):
     specific business tasks.
     """
 
+    _INPUT_ARGUMENTS_SIZE_LIMIT = 10000
+
     def __init__(
         self,
         config: UiPathApiConfig,
@@ -37,6 +40,7 @@ def invoke(
         *,
         folder_key: Optional[str] = None,
         folder_path: Optional[str] = None,
+        attachments: Optional[list[Attachment]] = None,
     ) -> Job:
         """Start execution of a process by its name.
 
@@ -45,6 +49,7 @@ def invoke(
         Args:
             name (str): The name of the process to execute.
             input_arguments (Optional[Dict[str, Any]]): The input arguments to pass to the process.
+            attachments (Optional[list]): List of Attachment objects to pass to the process.
             folder_key (Optional[str]): The key of the folder to execute the process in. Override the default one set in the SDK config.
             folder_path (Optional[str]): The path of the folder to execute the process in. Override the default one set in the SDK config.
 
@@ -72,9 +77,11 @@ def invoke(
         """
         input_data = self._handle_input_arguments(
             input_arguments=input_arguments,
+            attachments=attachments,
             folder_key=folder_key,
             folder_path=folder_path,
         )
+
         spec = self._invoke_spec(
             name,
             input_data=input_data,
@@ -101,6 +108,7 @@ async def invoke_async(
         *,
         folder_key: Optional[str] = None,
         folder_path: Optional[str] = None,
+        attachments: Optional[list[Attachment]] = None,
     ) -> Job:
         """Asynchronously start execution of a process by its name.
 
@@ -109,6 +117,7 @@ async def invoke_async(
         Args:
             name (str): The name of the process to execute.
             input_arguments (Optional[Dict[str, Any]]): The input arguments to pass to the process.
+            attachments (Optional[list]): List of Attachment objects to pass to the process.
             folder_key (Optional[str]): The key of the folder to execute the process in. Override the default one set in the SDK config.
             folder_path (Optional[str]): The path of the folder to execute the process in. Override the default one set in the SDK config.
 
@@ -132,6 +141,7 @@ async def main():
         """
         input_data = await self._handle_input_arguments_async(
             input_arguments=input_arguments,
+            attachments=attachments,
             folder_key=folder_key,
             folder_path=folder_path,
         )
@@ -157,71 +167,104 @@ async def main():
     def custom_headers(self) -> Dict[str, str]:
         return self.folder_headers
 
+    @staticmethod
+    def _prepare_link_attachments(
+        attachments: Optional[list[Attachment]],
+    ) -> Optional[list[Dict[str, str]]]:
+        """Format attachments for process invocation payload."""
+        if not attachments:
+            return None
+
+        link_attachments = [
+            {"attachmentId": str(att.id)} for att in attachments if att.id is not None
+        ]
+        return link_attachments if link_attachments else None
+
     def _handle_input_arguments(
         self,
         input_arguments: Optional[Dict[str, Any]] = None,
+        attachments: Optional[list[Attachment]] = None,
         *,
         folder_key: Optional[str] = None,
         folder_path: Optional[str] = None,
-    ) -> Dict[str, str]:
-        """Handle input arguments, storing as attachment if they exceed size limit.
+    ) -> Dict[str, Any]:
+        """Handle input arguments and attachments, storing as attachment if they exceed size limit.
 
         Args:
             input_arguments: The input arguments to process
+            attachments: List of Attachment objects to pass to the process
             folder_key: The folder key for attachment storage
             folder_path: The folder path for attachment storage
 
         Returns:
-            Dict containing either "InputArguments" or "InputFile" key
+            Dict containing either "InputArguments" or "InputFile" key, and optionally "Attachments"
         """
+        result: Dict[str, Any] = {}
+
+        # handle input arguments
         if not input_arguments:
-            return {"InputArguments": json.dumps({})}
-
-        # If payload exceeds limit, store as attachment
-        payload_json = json.dumps(input_arguments)
-        if len(payload_json) > 10000:  # 10k char limit
-            attachment_id = self._attachments_service.upload(
-                name=f"{uuid.uuid4()}.json",
-                content=payload_json,
-                folder_key=folder_key,
-                folder_path=folder_path,
-            )
-            return {"InputFile": str(attachment_id)}
+            result["InputArguments"] = json.dumps({})
         else:
-            return {"InputArguments": payload_json}
+            # If payload exceeds limit, store as attachment
+            payload_json = json.dumps(input_arguments)
+            if len(payload_json) > self._INPUT_ARGUMENTS_SIZE_LIMIT:
+                attachment_id = self._attachments_service.upload(
+                    name=f"{uuid.uuid4()}.json",
+                    content=payload_json,
+                    folder_key=folder_key,
+                    folder_path=folder_path,
+                )
+                result["InputFile"] = str(attachment_id)
+            else:
+                result["InputArguments"] = payload_json
+
+        link_attachments = self._prepare_link_attachments(attachments)
+        if link_attachments:
+            result["Attachments"] = link_attachments
+
+        return result
 
     async def _handle_input_arguments_async(
         self,
         input_arguments: Optional[Dict[str, Any]] = None,
+        attachments: Optional[list[Attachment]] = None,
         *,
         folder_key: Optional[str] = None,
         folder_path: Optional[str] = None,
-    ) -> Dict[str, str]:
-        """Handle input arguments, storing as attachment if they exceed size limit.
+    ) -> Dict[str, Any]:
+        """Handle input arguments and attachments, storing as attachment if they exceed size limit.
 
         Args:
             input_arguments: The input arguments to process
+            attachments: List of Attachment objects to pass to the process
             folder_key: The folder key for attachment storage
             folder_path: The folder path for attachment storage
 
         Returns:
-            Dict containing either "InputArguments" or "InputFile" key
+            Dict containing either "InputArguments" or "InputFile" key, and optionally "Attachments"
         """
+        result: Dict[str, Any] = {}
+
         if not input_arguments:
-            return {"InputArguments": json.dumps({})}
-
-        # If payload exceeds limit, store as attachment
-        payload_json = json.dumps(input_arguments)
-        if len(payload_json) > 10000:  # 10k char limit
-            attachment_id = await self._attachments_service.upload_async(
-                name=f"{uuid.uuid4()}.json",
-                content=payload_json,
-                folder_key=folder_key,
-                folder_path=folder_path,
-            )
-            return {"InputFile": str(attachment_id)}
+            result["InputArguments"] = json.dumps({})
         else:
-            return {"InputArguments": payload_json}
+            payload_json = json.dumps(input_arguments)
+            if len(payload_json) > self._INPUT_ARGUMENTS_SIZE_LIMIT:
+                attachment_id = await self._attachments_service.upload_async(
+                    name=f"{uuid.uuid4()}.json",
+                    content=payload_json,
+                    folder_key=folder_key,
+                    folder_path=folder_path,
+                )
+                result["InputFile"] = str(attachment_id)
+            else:
+                result["InputArguments"] = payload_json
+
+        formatted_attachments = self._prepare_link_attachments(attachments)
+        if formatted_attachments:
+            result["Attachments"] = formatted_attachments
+
+        return result
 
     def _invoke_spec(
         self,

src/uipath/platform/resume_triggers/_protocol.py

@@ -774,6 +774,7 @@ async def _handle_job_trigger(
             job = await uipath.processes.invoke_async(
                 name=value.name,
                 input_arguments=value.input_arguments,
+                attachments=value.attachments,
                 folder_path=value.process_folder_path,
                 folder_key=value.process_folder_key,
             )

tests/cli/test_hitl.py

@@ -891,6 +891,7 @@ async def test_create_resume_trigger_invoke_process(
             mock_process_invoke_async.assert_called_once_with(
                 name=invoke_process.name,
                 input_arguments=invoke_process.input_arguments,
+                attachments=None,
                 folder_path=invoke_process.process_folder_path,
                 folder_key=None,
             )

tests/sdk/services/test_attachments_service.py

@@ -993,6 +993,7 @@ def test_open_write_mode(
         file_name = "test_write_file.txt"
         file_content = b"Content to write"
         attachment = Attachment(  # type: ignore[call-arg]
+            ID=uuid.uuid4(),
             FullName=file_name,
             MimeType="text/plain",
         )
@@ -1144,6 +1145,7 @@ async def test_open_async_write_mode(
         file_name = "test_write_file_async.txt"
         file_content = b"Content to write async"
         attachment = Attachment(  # type: ignore[call-arg]
+            ID=uuid.uuid4(),
             FullName=file_name,
             MimeType="text/plain",
         )