feat(skills): add skills support to advanced agents with sandboxed cli tool

src/uipath_langchain/agent/advanced/agent.py

@@ -87,12 +87,16 @@ def create_advanced_agent(
     response_format: ResponseFormat[Any] | None = None,
     memory: Sequence[str] = (),
     middleware: Sequence[AgentMiddleware[Any, Any]] = (),
+    skills: Sequence[str] | None = None,
 ) -> CompiledStateGraph[Any, Any, Any, Any]:
     """Create a deepagents agent with planning, filesystem, and sub-agent tools.
 
     ``memory`` is a list of file paths loaded via deepagents' ``MemoryMiddleware``:
     each is read from ``backend`` and injected into the system prompt every turn,
     and the model maintains them with ``edit_file``. Empty disables the middleware.
+
+    ``skills`` is a list of skill source paths for deepagents' ``SkillsMiddleware``;
+    ``None`` or empty disables it (mirroring ``_create_deep_agent``'s contract).
     """
     return _create_deep_agent(
         model=model,
@@ -103,6 +107,7 @@ def create_advanced_agent(
         response_format=response_format,
         memory=list(memory) or None,
         middleware=list(middleware),
+        skills=list(skills) if skills else None,
     )
 
 
@@ -115,6 +120,7 @@ def create_advanced_agent_graph(
     input_schema: type[BaseModel] | None,
     output_schema: type[BaseModel],
     build_user_message: Callable[[dict[str, Any]], str],
+    skills: Sequence[str] | None = None,
 ) -> StateGraph[Any, Any, Any, Any]:
     """Wrap the advanced agent in a parent graph that maps typed I/O to/from messages.
 
@@ -153,6 +159,7 @@ def create_advanced_agent_graph(
             if runtime_system_prompt_key is not None
             else []
         ),
+        skills=skills,
     )
 
     wrapper_state = create_state_with_input(input_schema)
@@ -215,6 +222,7 @@ def create_conversational_advanced_agent_graph(
     tools: Sequence[BaseTool],
     system_prompt: str,
     backend: BackendProtocol | BackendFactory | None,
+    skills: Sequence[str] | None = None,
 ) -> StateGraph[Any, Any, Any, Any]:
     """Wrap the advanced agent in a parent graph that speaks the conversational contract.
 
@@ -237,6 +245,7 @@ def create_conversational_advanced_agent_graph(
         system_prompt=system_prompt,
         backend=backend,
         memory=memory_sources,
+        skills=skills,
     )
 
     class ConversationalAdvancedAgentOutput(BaseModel):

src/uipath_langchain/agent/tools/internal_tools/__init__.py

@@ -1,5 +1,6 @@
 """Internal Tool creation and management for LowCode agents."""
 
 from .internal_tool_factory import create_internal_tool
+from .uipath_cli_tool import create_uipath_cli_tool
 
-__all__ = ["create_internal_tool"]
+__all__ = ["create_internal_tool", "create_uipath_cli_tool"]

src/uipath_langchain/agent/tools/internal_tools/uipath_cli_tool.py

@@ -0,0 +1,251 @@
+"""Sandboxed ``uip`` CLI runner injected into advanced agents.
+
+Unlike the resource-selected internal tools built by ``create_internal_tool``, this
+tool is injected programmatically by the agent graph builder when UiPath skills are
+active — the agent invokes it to run the ``uip`` commands a skill prescribes. It runs
+exactly one ``uip`` invocation per call.
+
+"""
+
+import asyncio
+import logging
+import os
+import shlex
+import shutil
+from pathlib import Path
+from typing import Any
+
+from langchain_core.tools import StructuredTool
+from pydantic import BaseModel, Field
+from uipath.eval.mocks import mockable
+from uipath.runtime import Workspace
+
+from uipath_langchain.agent.tools.structured_tool_with_argument_properties import (
+    StructuredToolWithArgumentProperties,
+)
+
+logger = logging.getLogger("uipath")
+
+_TOOL_NAME = "uipath_cli"
+
+_TOOL_DESCRIPTION = (
+    "Run a single UiPath `uip` command in the agent workspace; returns "
+    "exit_code/stdout/stderr. Only the uip/uipath binaries, one command per call; "
+    "shell chaining is refused. A negative exit_code means the command was refused "
+    "before it ran and stderr explains why. Use `subdir` to target a scaffolded "
+    "project folder."
+)
+
+_ALLOWED_BINARIES: tuple[str, ...] = ("uip", "uipath")
+
+_REJECTED_EXIT_CODE = -1000
+
+_COMMAND_TIMEOUT_SECONDS = 600
+
+_SHELL_OPERATORS: frozenset[str] = frozenset({"&&", "||", ";", "|"})
+
+_COMMAND_EXAMPLES: tuple[str, ...] = (
+    "solution pack",
+    "solution publish",
+    "codedagent init",
+)
+
+
+def _parse_uip_command(command: str) -> list[str]:
+    """Validate a single ``uip`` command and return its argv (binary excluded).
+
+    Args:
+        command: The command to run, e.g. ``"pack"`` or ``"solution publish"``.
+            An explicit ``uip``/``uipath`` prefix is tolerated and stripped.
+
+    Returns:
+        The argument tokens to pass after the resolved binary.
+
+    Raises:
+        ValueError: If the command is empty or chains multiple commands via a shell
+            operator.
+    """
+    try:
+        lexer = shlex.shlex(command, posix=True)
+        lexer.whitespace_split = True
+        lexer.commenters = ""
+        lexer.escape = ""
+        tokens = list(lexer)
+    except ValueError as exc:
+        raise ValueError(f"Could not parse command: {exc}") from exc
+
+    if not tokens:
+        raise ValueError("No command provided.")
+
+    if tokens[0] in _ALLOWED_BINARIES:
+        tokens = tokens[1:]
+    if not tokens:
+        raise ValueError("No `uip` subcommand provided.")
+
+    for token in tokens:
+        if token in _SHELL_OPERATORS:
+            raise ValueError(
+                f"Shell operator '{token}' is not allowed; run one command at a time."
+            )
+
+    return tokens
+
+
+class UiPathCliInput(BaseModel):
+    """Input schema for the ``uipath_cli`` tool."""
+
+    command: str = Field(
+        description=(
+            "A single `uip` command without the binary prefix, e.g. `solution pack`, "
+            "`solution publish`. One command only; no shell operators."
+        ),
+        examples=list(_COMMAND_EXAMPLES),
+    )
+    subdir: str = Field(
+        default="",
+        description="Workspace-relative directory to run in; defaults to the root.",
+    )
+
+
+class UiPathCliOutput(BaseModel):
+    """Output schema for the ``uipath_cli`` tool."""
+
+    command: str = Field(description="The argv actually executed (or attempted).")
+    exit_code: int = Field(
+        description=(
+            f"Process exit code, or {_REJECTED_EXIT_CODE} when the command was "
+            "rejected before running (see stderr for the reason)."
+        )
+    )
+    stdout: str = Field(description="Captured standard output.")
+    stderr: str = Field(description="Captured standard error, or the failure reason.")
+
+
+def _rejected(command: str, reason: str) -> dict[str, Any]:
+    """Build a 'rejected before running' result carrying the sentinel and reason."""
+    logger.warning("uipath_cli rejected command %r: %s", command, reason)
+    return UiPathCliOutput(
+        command=command,
+        exit_code=_REJECTED_EXIT_CODE,
+        stdout="",
+        stderr=reason,
+    ).model_dump()
+
+
+def _resolve_run_dir_within_workspace(workspace_root: Path, subdir: str) -> Path | None:
+    """Resolve ``subdir`` under the workspace, or ``None`` if it escapes.
+
+    ``.resolve()`` collapses ``..`` and follows symlinks, so absolute paths, parent
+    traversal, and symlink escapes are all rejected.
+    """
+    run_dir = (workspace_root / subdir).resolve()
+    if run_dir == workspace_root or workspace_root in run_dir.parents:
+        return run_dir
+    return None
+
+
+async def _run_uip_subprocess(
+    binary: str, args: list[str], run_dir: Path, echoed: str
+) -> dict[str, Any]:
+    """Run the resolved ``uip`` command in ``run_dir`` and map it to output."""
+    logger.info("uipath_cli running %s (cwd=%s)", echoed, run_dir)
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            binary,
+            *args,
+            cwd=run_dir,
+            stdin=asyncio.subprocess.DEVNULL,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+    except OSError as exc:
+        return _rejected(echoed, str(exc))
+
+    try:
+        stdout, stderr = await asyncio.wait_for(
+            proc.communicate(), timeout=_COMMAND_TIMEOUT_SECONDS
+        )
+    except asyncio.TimeoutError:
+        try:
+            proc.kill()
+        except ProcessLookupError:
+            pass
+        await proc.wait()
+        return _rejected(
+            echoed, f"Command timed out after {_COMMAND_TIMEOUT_SECONDS}s."
+        )
+
+    logger.info("uipath_cli command %s exited with code %s", echoed, proc.returncode)
+    exit_code = proc.returncode if proc.returncode is not None else _REJECTED_EXIT_CODE
+    return UiPathCliOutput(
+        command=echoed,
+        exit_code=exit_code,
+        stdout=stdout.decode(errors="replace"),
+        stderr=stderr.decode(errors="replace"),
+    ).model_dump()
+
+
+def create_uipath_cli_tool(workspace: Workspace) -> StructuredTool:
+    """Create the sandboxed ``uipath_cli`` tool bound to ``workspace``.
+
+    The returned tool runs one ``uip`` command per call inside the workspace
+    (optionally under ``subdir``) and returns a :class:`UiPathCliOutput` dict. It is
+    injected by the agent graph builder rather than selected as a resource, and
+    enforces that only the ``uip``/``uipath`` binaries run, one command at a time.
+
+    Args:
+        workspace: The agent's workspace; commands run within its directory. Must be a
+            real on-disk workspace (inject only for a ``FilesystemBackend``).
+
+    Returns:
+        A ``StructuredTool`` named ``uipath_cli``.
+    """
+
+    async def run_uipath_command(command: str, subdir: str = "") -> dict[str, Any]:
+        try:
+            args = _parse_uip_command(command)
+        except ValueError as exc:
+            return _rejected(command.strip() or "uip", str(exc))
+
+        @mockable(
+            name=_TOOL_NAME,
+            description=_TOOL_DESCRIPTION,
+            input_schema=UiPathCliInput.model_json_schema(),
+            output_schema=UiPathCliOutput.model_json_schema(),
+            example_calls=[],
+        )
+        async def _execute(**_tool_kwargs: Any) -> dict[str, Any]:
+            binary = shutil.which("uip") or shutil.which("uipath")
+            if binary is None:
+                return _rejected(
+                    shlex.join(["uip", *args]),
+                    "No `uip` or `uipath` binary found on PATH.",
+                )
+            echoed = shlex.join([os.path.basename(binary), *args])
+
+            run_dir = _resolve_run_dir_within_workspace(
+                workspace.path.resolve(), subdir
+            )
+            if run_dir is None:
+                return _rejected(
+                    echoed, f"Invalid subdir '{subdir}': escapes the workspace."
+                )
+
+            return await _run_uip_subprocess(binary, args, run_dir, echoed)
+
+        return await _execute(command=command, subdir=subdir)
+
+    return StructuredToolWithArgumentProperties(
+        name=_TOOL_NAME,
+        description=_TOOL_DESCRIPTION,
+        args_schema=UiPathCliInput,
+        coroutine=run_uipath_command,
+        output_type=UiPathCliOutput,
+        argument_properties={},
+        metadata={
+            "tool_type": _TOOL_NAME,
+            "display_name": _TOOL_NAME,
+            "args_schema": UiPathCliInput,
+            "output_schema": UiPathCliOutput,
+        },
+    )

tests/agent/advanced/test_create_advanced_agent.py

@@ -69,3 +69,27 @@ def test_advanced_agent_converts_sequences_to_lists(
             _, kwargs = mock_upstream.call_args
             assert isinstance(kwargs["tools"], list)
             assert isinstance(kwargs["subagents"], list)
+
+    def test_advanced_agent_forwards_skills(self, mock_model: MagicMock) -> None:
+        """Non-empty skills are forwarded to the upstream builder as a list."""
+        with patch(
+            "uipath_langchain.agent.advanced.agent._create_deep_agent"
+        ) as mock_upstream:
+            mock_upstream.return_value = MagicMock(spec=CompiledStateGraph)
+            create_advanced_agent(
+                mock_model, system_prompt="test", skills=("/skills/",)
+            )
+            _, kwargs = mock_upstream.call_args
+            assert kwargs["skills"] == ["/skills/"]
+
+    def test_advanced_agent_empty_skills_becomes_none(
+        self, mock_model: MagicMock
+    ) -> None:
+        """An empty skills sequence collapses to None (disables the middleware)."""
+        with patch(
+            "uipath_langchain.agent.advanced.agent._create_deep_agent"
+        ) as mock_upstream:
+            mock_upstream.return_value = MagicMock(spec=CompiledStateGraph)
+            create_advanced_agent(mock_model, system_prompt="test")
+            _, kwargs = mock_upstream.call_args
+            assert kwargs["skills"] is None