fix(cli-server): parse string args without shlex

packages/uipath/pyproject.toml

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

packages/uipath/src/uipath/_cli/cli_server.py

@@ -2,7 +2,6 @@
 import importlib
 import json
 import os
-import shlex
 import sys
 import tempfile
 import time
@@ -108,15 +107,34 @@ def get_field(message: dict[str, Any], *keys: str) -> Any:
     return None
 
 
-def parse_args(args: str | list[str] | None) -> list[str]:
-    """Parse args into a list of strings."""
+def parse_args(args: Any) -> list[str] | None:
+    """Parse the request `args` field.
+
+    Accepts:
+      - None or [] → []
+      - list[str] → returned as-is
+      - str in the form `'<entrypoint>' '<json>'` → split into two tokens
+
+    Returns None for anything that does not match.
+    """
     if args is None:
         return []
-    if isinstance(args, list):
+    if isinstance(args, list) and all(isinstance(arg, str) for arg in args):
         return args
-    if isinstance(args, str):
-        return shlex.split(args)
-    return []
+    if not isinstance(args, str):
+        return None
+    stripped_args = args.strip()
+    if not stripped_args:
+        return []
+    # Expected shape: 'entrypoint' 'json-payload'
+    if not stripped_args.startswith("'") or not stripped_args.endswith("'"):
+        return None
+    boundary = stripped_args.find("' '", 1)
+    if boundary == -1:
+        return None
+    entrypoint = stripped_args[1:boundary]
+    payload = stripped_args[boundary + 3 : -1]
+    return [entrypoint, payload]
 
 
 async def send_ack(ack_socket_path: str, server_socket_path: str) -> None:
@@ -174,6 +192,17 @@ async def handle_start(request: web.Request) -> web.Response:
 
     args_raw = get_field(message, "args", "Args")
     args = parse_args(args_raw)
+    if args is None:
+        return web.json_response(
+            {
+                "success": False,
+                "error": (
+                    "Invalid field: 'args' must be a list of strings or a "
+                    "string of the form \"'<entrypoint>' '<json>'\""
+                ),
+            },
+            status=400,
+        )
 
     env_vars = get_field(message, "environmentVariables", "EnvironmentVariables") or {}
     working_dir = get_field(message, "workingDirectory", "WorkingDirectory")
@@ -386,7 +415,7 @@ def server(
     {"status": "ready", "socket": "/path/to/server.sock"}
 
     Endpoint: POST /jobs/{job_key}/start
-    Body: {"command": "run", "args": "agent.json '{}'", "environmentVariables": {}, "workingDirectory": "/path"}
+    Body: {"command": "run", "args": "'agent.json' '{}'", "environmentVariables": {}, "workingDirectory": "/path"}
 
     Endpoint: GET /health
     """

packages/uipath/tests/cli/test_server.py

@@ -3,12 +3,13 @@
 import os
 import threading
 import time
+from pathlib import Path
 from typing import Any
 
 import aiohttp
 import pytest
 
-from uipath._cli.cli_server import start_tcp_server
+from uipath._cli.cli_server import parse_args, start_tcp_server
 
 
 def create_uipath_json(script_path: str, entrypoint_name: str = "main"):
@@ -369,3 +370,122 @@ def test_env_restored_after_request(self, server_with_spy):
         assert "SHOULD_NOT_PERSIST" not in os.environ
         for key in baseline:
             assert os.environ.get(key) == baseline[key]
+
+
+class TestParseArgs:
+    def test_none_returns_empty_list(self):
+        assert parse_args(None) == []
+
+    def test_empty_string_returns_empty_list(self):
+        assert parse_args("") == []
+        assert parse_args("   ") == []
+
+    def test_list_of_strings_passthrough(self):
+        assert parse_args(["a", "b"]) == ["a", "b"]
+        assert parse_args([]) == []
+
+    def test_quoted_pair_string(self):
+        assert parse_args("'agent.json' '{}'") == ["agent.json", "{}"]
+
+    def test_quoted_pair_with_json_payload(self):
+        payload = '{"message":"Hello","repeat":3}'
+        result = parse_args(f"'agent.json' '{payload}'")
+        assert result == ["agent.json", payload]
+
+    def test_invalid_inputs_return_none(self):
+        assert parse_args(["a", 1]) is None
+        assert parse_args({"x": 1}) is None
+        assert parse_args(123) is None
+        # unquoted entrypoint not supported
+        assert parse_args("agent.json '{}'") is None
+        # missing boundary quote-space-quote
+        assert parse_args("'agent.json'") is None
+        # unterminated trailing quote
+        assert parse_args("'agent.json' '{}") is None
+
+
+class TestArgsValidationIntegration:
+    @pytest.fixture
+    def server_port(self):
+        import socket
+
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.bind(("127.0.0.1", 0))
+            return s.getsockname()[1]
+
+    @pytest.fixture
+    def server(self, server_port):
+        def run_server():
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            try:
+                loop.run_until_complete(start_tcp_server("127.0.0.1", server_port))
+            except asyncio.CancelledError:
+                pass
+            finally:
+                loop.close()
+
+        thread = threading.Thread(target=run_server, daemon=True)
+        thread.start()
+        time.sleep(0.5)
+        yield server_port
+
+    def test_malformed_string_args_rejected_with_400(self, server):
+        """Args that doesn't match `'entrypoint' 'json'` should 400."""
+        port = server
+
+        async def send():
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"http://127.0.0.1:{port}/jobs/job-1/start",
+                    # entrypoint not single-quoted — invalid
+                    json={"command": "run", "args": "agent.json '{}'"},
+                ) as response:
+                    return response.status, await response.json()
+
+        status, body = asyncio.run(send())
+        assert status == 400
+        assert body["success"] is False
+        assert "args" in body["error"]
+
+    def test_sample_payload_runs_with_string_args(self, server, temp_dir):
+        """POST `args` as a single `'entrypoint' '<json>'` string — the real
+        client wire shape that used to go through shlex.split and now goes
+        through parse_args."""
+        port = server
+
+        entrypoint = "agent.json"
+        args_string = '\'agent.json\' \'{"message":"Hello","repeat":3}\''
+
+        script = """
+from dataclasses import dataclass
+
+@dataclass
+class Input:
+    message: str
+    repeat: int = 1
+
+def main(input: Input) -> str:
+    return (input.message + " ") * input.repeat
+"""
+        with pytest.MonkeyPatch().context() as mp:
+            mp.chdir(temp_dir)
+
+            script_file = "entrypoint.py"
+            (Path(temp_dir) / script_file).write_text(script)
+            (Path(temp_dir) / "uipath.json").write_text(
+                json.dumps({"functions": {entrypoint: f"{script_file}:main"}})
+            )
+
+            async def send():
+                async with aiohttp.ClientSession() as session:
+                    async with session.post(
+                        f"http://127.0.0.1:{port}/jobs/job-sample/start",
+                        json={"command": "run", "args": args_string},
+                    ) as response:
+                        return await response.json()
+
+            response = asyncio.run(send())
+
+            assert response["success"] is True, response
+            assert response["job_key"] == "job-sample"