feat: add skills feature to advanced agent with workspace to load skills

mariadhakalUipath · claude · mariadhakalUipath · commit 55a90313ef2e · 2026-07-13T15:25:15.000-07:00
Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/src/uipath_langchain/agent/advanced/__init__.py b/src/uipath_langchain/agent/advanced/__init__.py
@@ -14,13 +14,17 @@
     MEMORY_DIR_NAME,
     MEMORY_INDEX_FILENAME,
     MEMORY_INDEX_VIRTUAL_PATH,
+    SKILLS_DIR_NAME,
+    SKILLS_VIRTUAL_PATH,
     create_state_with_input,
 )
 
 __all__ = [
     "MEMORY_DIR_NAME",
     "MEMORY_INDEX_FILENAME",
     "MEMORY_INDEX_VIRTUAL_PATH",
+    "SKILLS_DIR_NAME",
+    "SKILLS_VIRTUAL_PATH",
     "AdvancedAgentGraphState",
     "BackendFactory",
     "BackendProtocol",
diff --git a/src/uipath_langchain/agent/advanced/agent.py b/src/uipath_langchain/agent/advanced/agent.py
@@ -24,6 +24,7 @@
     MEMORY_INDEX_VIRTUAL_PATH,
     create_state_with_input,
     resolve_input_attachments,
+    resolve_skill_sources,
 )
 
 
@@ -35,12 +36,14 @@ def create_advanced_agent(
     backend: BackendProtocol | BackendFactory | None = None,
     response_format: ResponseFormat[Any] | None = None,
     memory: Sequence[str] = (),
+    skills: Sequence[str] = (),
 ) -> 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.
+    ``memory`` (file paths) and ``skills`` (source directories) are loaded via
+    deepagents' ``MemoryMiddleware`` and ``SkillsMiddleware``; an empty sequence
+    disables each. Workspace tools like ``uipath_cli`` are supplied by the caller
+    via ``tools``.
     """
     return _create_deep_agent(
         model=model,
@@ -50,6 +53,7 @@ def create_advanced_agent(
         backend=backend,
         response_format=response_format,
         memory=list(memory) or None,
+        skills=list(skills) or None,
     )
 
 
@@ -62,14 +66,14 @@ 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.
 
     With a ``FilesystemBackend``, attachment-shaped inputs are downloaded into the
-    workspace and given a ``FilePath`` before the user message is built. A
-    ``FilesystemBackend`` also enables workspace memory: deepagents'
-    ``MemoryMiddleware`` reads ``/memory/MEMORY.md`` from the backend each turn.
-    Memory stays disabled for non-filesystem backends, which carry no workspace.
+    workspace, memory (``/memory/MEMORY.md``) is enabled, and the workspace
+    ``skills/`` dir is prepended to any declared ``skills``. Both stay disabled for
+    non-filesystem backends, which carry no workspace.
     """
     memory_sources = (
         [MEMORY_INDEX_VIRTUAL_PATH] if isinstance(backend, FilesystemBackend) else []
@@ -82,6 +86,7 @@ def create_advanced_agent_graph(
         backend=backend,
         response_format=response_format,
         memory=memory_sources,
+        skills=resolve_skill_sources(backend, skills),
     )
 
     wrapper_state = create_state_with_input(input_schema)
diff --git a/src/uipath_langchain/agent/advanced/utils.py b/src/uipath_langchain/agent/advanced/utils.py
@@ -4,6 +4,7 @@
 import copy
 import logging
 import uuid
+from collections.abc import Sequence
 from pathlib import Path
 from typing import Any, NamedTuple, cast
 
@@ -30,6 +31,28 @@
 # FilesystemBackend resolves it under the workspace root.
 MEMORY_INDEX_VIRTUAL_PATH = f"/{MEMORY_DIR_NAME}/{MEMORY_INDEX_FILENAME}"
 
+# Skills live under <workspace>/skills/
+SKILLS_DIR_NAME = "skills"
+
+SKILLS_VIRTUAL_PATH = f"/{SKILLS_DIR_NAME}/"
+
+
+def resolve_skill_sources(
+    backend: BackendProtocol | BackendFactory | None,
+    skills: Sequence[str] | None,
+) -> list[str]:
+    """Resolve the skill sources for ``SkillsMiddleware``.
+
+    With a ``FilesystemBackend`` the workspace ``skills/`` dir is prepended to the
+    declared ``skills``; other backends carry no workspace, so only ``skills`` is used.
+    """
+    if not skills:
+        return []
+    sources = list(skills)
+    if isinstance(backend, FilesystemBackend) and SKILLS_VIRTUAL_PATH not in sources:
+        sources.insert(0, SKILLS_VIRTUAL_PATH)
+    return sources
+
 
 def create_state_with_input(
     input_schema: type[BaseModel] | None,
diff --git a/src/uipath_langchain/agent/tools/internal_tools/__init__.py b/src/uipath_langchain/agent/tools/internal_tools/__init__.py
@@ -1,5 +1,9 @@
 """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",
+]
diff --git a/src/uipath_langchain/agent/tools/internal_tools/uipath_cli_tool.py b/src/uipath_langchain/agent/tools/internal_tools/uipath_cli_tool.py
@@ -0,0 +1,175 @@
+"""Workspace CLI tool injected when skills are active."""
+
+from __future__ import annotations
+
+import os
+import shlex
+import subprocess
+from collections.abc import Iterable
+from pathlib import Path
+
+from langchain_core.tools import BaseTool
+from pydantic import BaseModel, Field
+from uipath.runtime import Workspace
+
+from uipath_langchain.agent.tools.base_uipath_structured_tool import (
+    BaseUiPathStructuredTool,
+)
+
+ALLOWED_BINARIES = {"uipath", "uip"}
+MAX_LOG_BYTES = 4000
+
+# Classify as "internal" so traces label it (untyped default is "Integration").
+UIPATH_CLI_TOOL_NAME = "uipath_cli"
+UIPATH_CLI_TOOL_TYPE = "internal"
+
+# Destructive verbs blocked by default.
+DEFAULT_DENIED_SUBCOMMANDS = frozenset(
+    {
+        "delete",
+        "remove",
+        "cancel",
+        "uninstall",
+        "unassign",
+        "unlink",
+        "unset",
+        "clear",
+        "rm",
+        "del",
+        "destroy",
+        "purge",
+        "drop",
+        "revoke",
+        "deactivate",
+        "undeploy",
+        "unpublish",
+        "unregister",
+        "unimport",
+        "prune",
+        "wipe",
+        "reset",
+    }
+)
+
+__all__ = [
+    "create_uipath_cli_tool",
+    "DEFAULT_DENIED_SUBCOMMANDS",
+    "UIPATH_CLI_TOOL_NAME",
+    "UIPATH_CLI_TOOL_TYPE",
+]
+
+_TOOL_DESCRIPTION = (
+    "Run a `uipath` / `uip` command in a workspace subdirectory. Returns combined "
+    "stdout/stderr. Only `uipath` and `uip` are allowed. Inherits the agent process "
+    "environment, including `UIPATH_ACCESS_TOKEN` / `UIPATH_BASE_URL` for tenant auth."
+)
+
+
+class UiPathCliArgs(BaseModel):
+    """Arguments for ``uipath_cli``."""
+
+    command: str = Field(
+        description=(
+            "Full CLI command starting with `uipath` or `uip`, e.g. 'uipath pack' "
+            "or 'uip solution publish dev'."
+        ),
+    )
+    subdir: str = Field(
+        default="",
+        description=(
+            "Workspace-relative subdirectory to run in. Give each scaffolded "
+            "project its own subdir. Defaults to the workspace root."
+        ),
+    )
+
+
+def create_uipath_cli_tool(
+    workspace: Workspace,
+    timeout_sec: int = 120,
+    denied_subcommands: Iterable[str] | None = None,
+) -> BaseTool:
+    """Build a ``uipath_cli`` tool jailed to ``workspace`` for advanced agent builders with UiPath skills.
+    The tool only reads ``workspace.path``; the workspace lifecycle (creation, cleanup) stays owned by the runtime.
+    """
+    workspace_path = Path(workspace.path).expanduser().resolve()
+    workspace_path.mkdir(parents=True, exist_ok=True)
+    denied = frozenset(
+        s.lower()
+        for s in (
+            DEFAULT_DENIED_SUBCOMMANDS
+            if denied_subcommands is None
+            else denied_subcommands
+        )
+    )
+
+    def run_uipath_cli(command: str, subdir: str = "") -> str:
+        run_dir = workspace_path
+        if subdir:
+            candidate = (workspace_path / subdir).resolve()
+            if not candidate.is_relative_to(workspace_path):
+                return f"ERROR: subdir '{subdir}' escapes the workspace directory."
+            candidate.mkdir(parents=True, exist_ok=True)
+            run_dir = candidate
+
+        try:
+            argv = shlex.split(command)
+        except ValueError as e:
+            return f"ERROR: could not parse command: {e}"
+
+        if not argv:
+            return "ERROR: empty command"
+
+        binary = argv[0]
+        # Reject any path-qualified binary (e.g. ./uip, /tmp/uip) so only the
+        # UiPath CLI resolved from PATH runs, not a lookalike in the workspace.
+        if os.path.basename(binary) != binary or binary not in ALLOWED_BINARIES:
+            return (
+                f"ERROR: '{binary}' is not allowed. Only these binaries are "
+                f"permitted (by name, without a path): {sorted(ALLOWED_BINARIES)}"
+            )
+
+        def is_denied(tok: str) -> bool:
+            low = tok.lower()
+            return any(low == d or low.startswith(f"{d}-") for d in denied)
+
+        blocked = next(
+            (tok for tok in argv[1:] if not tok.startswith("-") and is_denied(tok)),
+            None,
+        )
+        if blocked is not None:
+            return (
+                f"ERROR: subcommand '{blocked}' is not allowed. These destructive "
+                f"subcommands are blocked: {sorted(denied)}"
+            )
+
+        try:
+            proc = subprocess.run(
+                argv,
+                cwd=str(run_dir),
+                capture_output=True,
+                text=True,
+                timeout=timeout_sec,
+                check=False,
+            )
+        except subprocess.TimeoutExpired:
+            return f"ERROR: command timed out after {timeout_sec}s"
+        except FileNotFoundError:
+            return f"ERROR: '{binary}' not found on PATH. Install the UiPath CLI."
+
+        return (
+            f"exit_code: {proc.returncode}\n"
+            f"--- stdout ---\n{proc.stdout[:MAX_LOG_BYTES]}\n"
+            f"--- stderr ---\n{proc.stderr[:MAX_LOG_BYTES]}"
+        )
+
+    return BaseUiPathStructuredTool(
+        name=UIPATH_CLI_TOOL_NAME,
+        description=_TOOL_DESCRIPTION,
+        args_schema=UiPathCliArgs,
+        func=run_uipath_cli,
+        metadata={
+            "tool_type": UIPATH_CLI_TOOL_TYPE,
+            "display_name": UIPATH_CLI_TOOL_NAME,
+            "args_schema": UiPathCliArgs,
+        },
+    )
diff --git a/tests/agent/advanced/test_skills_injection.py b/tests/agent/advanced/test_skills_injection.py
@@ -0,0 +1,80 @@
+"""Tests for deepagents skills based on the backend wrapper."""
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from deepagents.backends.filesystem import FilesystemBackend
+from langchain_core.language_models import BaseChatModel
+from pydantic import BaseModel
+
+from uipath_langchain.agent.advanced.agent import (
+    create_advanced_agent_graph,
+)
+from uipath_langchain.agent.advanced.utils import (
+    SKILLS_VIRTUAL_PATH,
+)
+
+
+class _Input(BaseModel):
+    """Minimal input schema for the test."""
+
+    task: str
+
+
+class _Output(BaseModel):
+    """Minimal output schema for the test."""
+
+    result: str = ""
+
+
+def _build_user_message(args: dict[str, Any]) -> str:
+    return args.get("task", "")
+
+
+def _skills_kwarg(backend: Any, **overrides: Any) -> Any:
+    """Build the wrapper graph and return the ``skills`` kwarg handed to deepagents."""
+    with patch(
+        "uipath_langchain.agent.advanced.agent._create_deep_agent",
+        return_value=MagicMock(),
+    ) as mock_create:
+        create_advanced_agent_graph(
+            model=MagicMock(spec=BaseChatModel),
+            tools=[],
+            system_prompt="",
+            backend=backend,
+            response_format=None,
+            input_schema=_Input,
+            output_schema=_Output,
+            build_user_message=_build_user_message,
+            **overrides,
+        )
+    return mock_create.call_args.kwargs["skills"]
+
+
+class TestWorkspaceSkillsWiring:
+    """Skills are opt-in: enabled only when the caller declares them."""
+
+    def test_disabled_by_default_on_filesystem_backend(self, tmp_path: Any) -> None:
+        backend = FilesystemBackend(root_dir=tmp_path, virtual_mode=True)
+        assert _skills_kwarg(backend) is None
+
+    def test_empty_skills_sequence_disables(self, tmp_path: Any) -> None:
+        # An empty sequence disables skills, matching the ``memory`` parameter.
+        backend = FilesystemBackend(root_dir=tmp_path, virtual_mode=True)
+        assert _skills_kwarg(backend, skills=[]) is None
+
+    def test_declared_skills_enable_and_prepend_workspace(self, tmp_path: Any) -> None:
+        backend = FilesystemBackend(root_dir=tmp_path, virtual_mode=True)
+        assert _skills_kwarg(backend, skills=["/extra/"]) == [
+            SKILLS_VIRTUAL_PATH,
+            "/extra/",
+        ]
+
+    def test_no_duplicate_workspace_source(self, tmp_path: Any) -> None:
+        backend = FilesystemBackend(root_dir=tmp_path, virtual_mode=True)
+        assert _skills_kwarg(backend, skills=[SKILLS_VIRTUAL_PATH]) == [
+            SKILLS_VIRTUAL_PATH
+        ]
+
+    def test_explicit_sources_only_for_non_filesystem_backend(self) -> None:
+        assert _skills_kwarg(None, skills=["/extra/"]) == ["/extra/"]
diff --git a/tests/agent/tools/internal_tools/test_uipath_cli_tool.py b/tests/agent/tools/internal_tools/test_uipath_cli_tool.py
