mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Support list[str] arguments for file-based skill scripts (#5850)
Port of .NET PR #5475. Broadens the args type from dict[str, Any] | None to dict[str, Any] | list[str] | None across the skill script API surface, enabling CLI-style argv forwarding to subprocess scripts. Changes: - SkillScript.run(), InlineSkillScript.run(), FileSkillScript.run(): widen args type; InlineSkillScript rejects list with TypeError - FileSkillScript.parameters_schema: returns array-of-strings schema - FileSkill.content: appends <scripts> block with parameters_schema - SkillScriptRunner protocol: widen args type - SkillsProvider._run_skill_script: widen args type - run_skill_script tool schema: accept object, array, or null - subprocess_script_runner sample: accept list[str], reject dict - class_based_skill sample: fix missing SkillFrontmatter wrapper - Standardize 'folder' to 'directory' in docstrings (#5712) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
committed by
GitHub
Unverified
parent
3256550c55
commit
7432105ebe
@@ -289,13 +289,15 @@ class SkillScript(ABC):
|
||||
return None
|
||||
|
||||
@abstractmethod
|
||||
async def run(self, skill: Skill, args: dict[str, Any] | None = None, **kwargs: Any) -> Any:
|
||||
async def run(self, skill: Skill, args: dict[str, Any] | list[str] | None = None, **kwargs: Any) -> Any:
|
||||
"""Run this script.
|
||||
|
||||
Args:
|
||||
skill: The skill that owns this script.
|
||||
args: Optional keyword arguments for the script, provided by the
|
||||
agent/LLM.
|
||||
args: Optional arguments for the script, provided by the
|
||||
agent/LLM. May be a ``dict`` (named keyword arguments
|
||||
for inline scripts) or a ``list[str]`` (positional CLI
|
||||
arguments for file-based scripts).
|
||||
**kwargs: Runtime keyword arguments forwarded only to script
|
||||
functions that accept ``**kwargs``.
|
||||
|
||||
@@ -361,19 +363,31 @@ class InlineSkillScript(SkillScript):
|
||||
self._parameters_schema_resolved = True
|
||||
return self._parameters_schema
|
||||
|
||||
async def run(self, skill: Skill, args: dict[str, Any] | None = None, **kwargs: Any) -> Any:
|
||||
async def run(self, skill: Skill, args: dict[str, Any] | list[str] | None = None, **kwargs: Any) -> Any:
|
||||
"""Run the script by invoking the callable in-process.
|
||||
|
||||
Args:
|
||||
skill: The skill that owns this script.
|
||||
args: Optional keyword arguments for the script, provided by the
|
||||
agent/LLM.
|
||||
agent/LLM. Must be a ``dict`` or ``None``; passing a
|
||||
``list`` raises :class:`TypeError` because inline scripts
|
||||
bind arguments by keyword name.
|
||||
**kwargs: Runtime keyword arguments forwarded only to script
|
||||
functions that accept ``**kwargs``.
|
||||
|
||||
Returns:
|
||||
The script execution result.
|
||||
|
||||
Raises:
|
||||
TypeError: If ``args`` is a ``list`` (array-style arguments
|
||||
are only supported for file-based scripts).
|
||||
"""
|
||||
if isinstance(args, list):
|
||||
raise TypeError(
|
||||
f"Inline script '{self.name}' requires keyword arguments (dict), "
|
||||
f"but received a list. Array-style arguments are only supported "
|
||||
f"for file-based scripts."
|
||||
)
|
||||
if self._accepts_kwargs: # noqa: SIM108
|
||||
result = self.function(**(args or {}), **kwargs)
|
||||
else:
|
||||
@@ -431,13 +445,23 @@ class FileSkillScript(SkillScript):
|
||||
self.full_path = full_path
|
||||
self._runner = runner
|
||||
|
||||
async def run(self, skill: Skill, args: dict[str, Any] | None = None, **kwargs: Any) -> Any:
|
||||
@property
|
||||
def parameters_schema(self) -> dict[str, Any] | None:
|
||||
"""JSON Schema advertising that file scripts accept a string array.
|
||||
|
||||
Returns a fixed schema ``{"type": "array", "items": {"type": "string"}}``
|
||||
so that the LLM knows to pass positional CLI arguments as a JSON array
|
||||
of strings.
|
||||
"""
|
||||
return {"type": "array", "items": {"type": "string"}}
|
||||
|
||||
async def run(self, skill: Skill, args: dict[str, Any] | list[str] | None = None, **kwargs: Any) -> Any:
|
||||
"""Run the script by delegating to the configured runner.
|
||||
|
||||
Args:
|
||||
skill: The skill that owns this script. Must be a
|
||||
:class:`FileSkill`.
|
||||
args: Optional keyword arguments for the script.
|
||||
args: Optional arguments for the script.
|
||||
**kwargs: Additional runtime keyword arguments (unused).
|
||||
|
||||
Returns:
|
||||
@@ -1348,6 +1372,7 @@ class FileSkill(Skill):
|
||||
self.path = path
|
||||
self._resources: list[SkillResource] = list(resources) if resources is not None else []
|
||||
self._scripts: list[SkillScript] = list(scripts) if scripts is not None else []
|
||||
self._cached_content: str | None = None
|
||||
|
||||
@property
|
||||
def frontmatter(self) -> SkillFrontmatter:
|
||||
@@ -1356,8 +1381,23 @@ class FileSkill(Skill):
|
||||
|
||||
@property
|
||||
def content(self) -> str:
|
||||
"""The skill content provided at construction time."""
|
||||
return self._content
|
||||
"""The skill content with appended scripts block.
|
||||
|
||||
When scripts are present, a ``<scripts>`` XML block is appended
|
||||
to the raw SKILL.md content so that the LLM can discover each
|
||||
script's ``<parameters_schema>``.
|
||||
|
||||
The result is cached after the first access. Adding scripts
|
||||
after the first access will not be reflected.
|
||||
"""
|
||||
if self._cached_content is not None:
|
||||
return self._cached_content
|
||||
if not self._scripts:
|
||||
self._cached_content = self._content
|
||||
else:
|
||||
script_lines = "\n".join(_create_script_element(s) for s in self._scripts)
|
||||
self._cached_content = f"{self._content}\n\n<scripts>\n{script_lines}\n</scripts>"
|
||||
return self._cached_content
|
||||
|
||||
@property
|
||||
def resources(self) -> list[SkillResource]:
|
||||
@@ -1392,7 +1432,9 @@ class SkillScriptRunner(Protocol):
|
||||
satisfies this protocol.
|
||||
"""
|
||||
|
||||
def __call__(self, skill: FileSkill, script: FileSkillScript, args: dict[str, Any] | None = None) -> Any:
|
||||
def __call__(
|
||||
self, skill: FileSkill, script: FileSkillScript, args: dict[str, Any] | list[str] | None = None
|
||||
) -> Any:
|
||||
"""Run a skill script.
|
||||
|
||||
The :class:`SkillsProvider` resolves skill and script names
|
||||
@@ -1402,7 +1444,7 @@ class SkillScriptRunner(Protocol):
|
||||
Args:
|
||||
skill: The file-based skill that owns the script.
|
||||
script: The file-based script to run.
|
||||
args: Optional keyword arguments for the script.
|
||||
args: Optional arguments for the script.
|
||||
|
||||
Returns:
|
||||
The result. May be any type; the framework
|
||||
@@ -1982,7 +2024,7 @@ class SkillsProvider(ContextProvider):
|
||||
if include_script_runner_tool:
|
||||
|
||||
async def _run_script(
|
||||
skill_name: str, script_name: str, args: dict[str, Any] | None = None, **kwargs: Any
|
||||
skill_name: str, script_name: str, args: dict[str, Any] | list[str] | None = None, **kwargs: Any
|
||||
) -> Any:
|
||||
return await self._run_skill_script(skills, skill_name, script_name, args, **kwargs)
|
||||
|
||||
@@ -2005,12 +2047,31 @@ class SkillsProvider(ContextProvider):
|
||||
),
|
||||
},
|
||||
"args": {
|
||||
"type": ["object", "null"],
|
||||
"additionalProperties": True,
|
||||
"oneOf": [
|
||||
{
|
||||
"type": "object",
|
||||
"additionalProperties": True,
|
||||
"description": (
|
||||
"Named arguments as key-value pairs "
|
||||
'(e.g. {"length": 24, "uppercase": true}).'
|
||||
),
|
||||
},
|
||||
{
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": (
|
||||
"Positional CLI arguments as a string array "
|
||||
'(e.g. ["input.docx", "--output", "result.idx"]).'
|
||||
),
|
||||
},
|
||||
{"type": "null"},
|
||||
],
|
||||
"default": None,
|
||||
"description": (
|
||||
"Arguments to pass to the script as key-value pairs. "
|
||||
"Use parameter names as keys without leading dashes "
|
||||
"Arguments to pass to the script. "
|
||||
"Use an array of strings for CLI-style positional arguments "
|
||||
'(e.g. ["input.docx", "--output", "result.idx"]), '
|
||||
"or an object for named parameters "
|
||||
'(e.g. {"length": 24, "uppercase": true}). '
|
||||
"How these values are mapped to the underlying script "
|
||||
"is determined by the script implementation or configured runner."
|
||||
@@ -2060,7 +2121,7 @@ class SkillsProvider(ContextProvider):
|
||||
skills: Sequence[Skill],
|
||||
skill_name: str,
|
||||
script_name: str,
|
||||
args: dict[str, Any] | None = None,
|
||||
args: dict[str, Any] | list[str] | None = None,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
"""Run a named script from a skill.
|
||||
@@ -2072,9 +2133,8 @@ class SkillsProvider(ContextProvider):
|
||||
skills: The skills to look up the skill from.
|
||||
skill_name: The name of the owning skill.
|
||||
script_name: The script name to look up (case-insensitive).
|
||||
args: Optional keyword arguments for the script, provided by the
|
||||
agent/LLM. These are mapped to the function's declared
|
||||
parameters.
|
||||
args: Optional arguments for the script, provided by the
|
||||
agent/LLM.
|
||||
**kwargs: Runtime keyword arguments forwarded only to script
|
||||
functions that accept ``**kwargs`` (e.g. arguments passed via
|
||||
``agent.run(user_id="123")``).
|
||||
@@ -2254,7 +2314,7 @@ class FileSkillsSource(SkillsSource):
|
||||
|
||||
Args:
|
||||
skill_paths: One or more directory paths to search for file-based
|
||||
skills. Each path may point to an individual skill folder
|
||||
skills. Each path may point to an individual skill directory
|
||||
(containing ``SKILL.md``) or to a parent that contains skill
|
||||
subdirectories.
|
||||
|
||||
|
||||
@@ -3518,7 +3518,6 @@ class TestSkillsProviderFactories:
|
||||
await _init_provider(provider)
|
||||
run_tool = next(t for t in _ctx(provider)[2] if hasattr(t, "name") and t.name == "run_skill_script")
|
||||
args_desc = run_tool.parameters()["properties"]["args"]["description"]
|
||||
assert "without leading dashes" in args_desc
|
||||
assert "script implementation or configured runner" in args_desc
|
||||
|
||||
async def test_require_script_approval_sets_approval_mode(self) -> None:
|
||||
@@ -4744,12 +4743,16 @@ class TestCreateScriptElement:
|
||||
def test_name_only(self) -> None:
|
||||
s = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/scripts/run.py")
|
||||
elem = _create_script_element(s)
|
||||
assert elem == ' <script name="run.py"/>'
|
||||
assert 'name="run.py"' in elem
|
||||
assert "<parameters_schema>" in elem
|
||||
assert '"type": "array"' in elem
|
||||
|
||||
def test_with_description(self) -> None:
|
||||
s = FileSkillScript(name="run.py", description="Execute script.", full_path=f"{_ABS}/test/scripts/run.py")
|
||||
elem = _create_script_element(s)
|
||||
assert elem == ' <script name="run.py" description="Execute script."/>'
|
||||
assert 'name="run.py"' in elem
|
||||
assert 'description="Execute script."' in elem
|
||||
assert "<parameters_schema>" in elem
|
||||
|
||||
def test_xml_escapes_name(self) -> None:
|
||||
s = FileSkillScript(name='script"special', full_path=f"{_ABS}/test/scripts/s.py")
|
||||
@@ -4776,10 +4779,12 @@ class TestCreateScriptElement:
|
||||
assert "query" in elem
|
||||
assert """ not in elem
|
||||
|
||||
def test_no_parameters_for_file_script(self) -> None:
|
||||
def test_file_script_includes_array_parameters(self) -> None:
|
||||
s = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/scripts/run.py")
|
||||
elem = _create_script_element(s)
|
||||
assert "<parameters_schema>" not in elem
|
||||
assert "<parameters_schema>" in elem
|
||||
assert '"type": "array"' in elem
|
||||
assert '"type": "string"' in elem
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -4800,7 +4805,7 @@ class TestSkillScriptParametersSchema:
|
||||
|
||||
def test_none_for_file_based_script(self) -> None:
|
||||
script = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/scripts/run.py")
|
||||
assert script.parameters_schema is None
|
||||
assert script.parameters_schema == {"type": "array", "items": {"type": "string"}}
|
||||
|
||||
def test_no_params_function_returns_none(self) -> None:
|
||||
def noop() -> None:
|
||||
@@ -5407,3 +5412,169 @@ class TestInlineSkillContentCaching:
|
||||
second = skill.content
|
||||
assert first is second # Same object (cached)
|
||||
assert "<name>test-skill</name>" in first
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: Array-style (list[str]) script arguments
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestArrayStyleScriptArgs:
|
||||
"""Tests for list[str] arguments on skill scripts (port of .NET PR #5475)."""
|
||||
|
||||
async def test_inline_script_rejects_list_args(self) -> None:
|
||||
"""InlineSkillScript.run() raises TypeError when args is a list."""
|
||||
script = InlineSkillScript(name="greet", function=lambda name="world": f"hello {name}")
|
||||
skill = InlineSkill(frontmatter=SkillFrontmatter(name="s", description="d"), instructions="c")
|
||||
with pytest.raises(TypeError, match="requires keyword arguments"):
|
||||
await script.run(skill, args=["hello", "--name", "Alice"])
|
||||
|
||||
async def test_inline_script_error_message_mentions_script_name(self) -> None:
|
||||
"""The TypeError message includes the script name for debugging."""
|
||||
script = InlineSkillScript(name="my-script", function=lambda: None)
|
||||
skill = InlineSkill(frontmatter=SkillFrontmatter(name="s", description="d"), instructions="c")
|
||||
with pytest.raises(TypeError, match="my-script"):
|
||||
await script.run(skill, args=["arg1"])
|
||||
|
||||
async def test_file_script_passes_list_to_runner(self) -> None:
|
||||
"""FileSkillScript.run() passes list[str] args through to the runner."""
|
||||
captured: dict[str, Any] = {}
|
||||
|
||||
def runner(skill: Any, script: Any, args: Any = None) -> str:
|
||||
captured["args"] = args
|
||||
return "ok"
|
||||
|
||||
script = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/run.py", runner=runner)
|
||||
skill = FileSkill(
|
||||
frontmatter=SkillFrontmatter(name="my-skill", description="d"), content="c", path=f"{_ABS}/test"
|
||||
)
|
||||
result = await script.run(skill, args=["input.docx", "--output", "result.idx"])
|
||||
assert result == "ok"
|
||||
assert captured["args"] == ["input.docx", "--output", "result.idx"]
|
||||
|
||||
async def test_file_script_passes_dict_to_runner(self) -> None:
|
||||
"""FileSkillScript.run() still passes dict args through to the runner."""
|
||||
captured: dict[str, Any] = {}
|
||||
|
||||
def runner(skill: Any, script: Any, args: Any = None) -> str:
|
||||
captured["args"] = args
|
||||
return "ok"
|
||||
|
||||
script = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/run.py", runner=runner)
|
||||
skill = FileSkill(
|
||||
frontmatter=SkillFrontmatter(name="my-skill", description="d"), content="c", path=f"{_ABS}/test"
|
||||
)
|
||||
result = await script.run(skill, args={"key": "val"})
|
||||
assert result == "ok"
|
||||
assert captured["args"] == {"key": "val"}
|
||||
|
||||
async def test_file_script_passes_none_to_runner(self) -> None:
|
||||
"""FileSkillScript.run() passes None args through to the runner."""
|
||||
captured: dict[str, Any] = {}
|
||||
|
||||
def runner(skill: Any, script: Any, args: Any = None) -> str:
|
||||
captured["args"] = args
|
||||
return "ok"
|
||||
|
||||
script = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/run.py", runner=runner)
|
||||
skill = FileSkill(
|
||||
frontmatter=SkillFrontmatter(name="my-skill", description="d"), content="c", path=f"{_ABS}/test"
|
||||
)
|
||||
result = await script.run(skill)
|
||||
assert result == "ok"
|
||||
assert captured["args"] is None
|
||||
|
||||
def test_file_script_parameters_schema_returns_array(self) -> None:
|
||||
"""FileSkillScript.parameters_schema returns the string-array JSON schema."""
|
||||
script = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/run.py")
|
||||
assert script.parameters_schema == {"type": "array", "items": {"type": "string"}}
|
||||
|
||||
async def test_runner_protocol_accepts_list_args(self) -> None:
|
||||
"""A runner accepting list[str] args satisfies the SkillScriptRunner protocol."""
|
||||
captured: dict[str, Any] = {}
|
||||
|
||||
def my_runner(skill: Any, script: Any, args: Any = None) -> str:
|
||||
captured["args"] = args
|
||||
return "ok"
|
||||
|
||||
assert isinstance(my_runner, SkillScriptRunner)
|
||||
skill = FileSkill(
|
||||
frontmatter=SkillFrontmatter(name="s", description="d"), content="c", path=f"{_ABS}/test"
|
||||
)
|
||||
script = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/run.py")
|
||||
result = my_runner(skill, script, args=["--flag", "value"])
|
||||
assert result == "ok"
|
||||
assert captured["args"] == ["--flag", "value"]
|
||||
|
||||
async def test_tool_schema_accepts_array_args(self) -> None:
|
||||
"""The run_skill_script tool schema accepts array-style args via oneOf."""
|
||||
skill = InlineSkill(frontmatter=SkillFrontmatter(name="my-skill", description="test"), instructions="body")
|
||||
skill.scripts.append(InlineSkillScript(name="s1", function=lambda: None))
|
||||
|
||||
provider = SkillsProvider([skill])
|
||||
await _init_provider(provider)
|
||||
run_tool = next(t for t in _ctx(provider)[2] if hasattr(t, "name") and t.name == "run_skill_script")
|
||||
args_schema = run_tool.parameters()["properties"]["args"]
|
||||
assert "oneOf" in args_schema
|
||||
types = [s.get("type") for s in args_schema["oneOf"]]
|
||||
assert "object" in types
|
||||
assert "array" in types
|
||||
assert "null" in types
|
||||
|
||||
async def test_run_skill_script_with_list_args_via_provider(self) -> None:
|
||||
"""End-to-end: list args flow through provider to file-based script runner."""
|
||||
captured: dict[str, Any] = {}
|
||||
|
||||
def runner(skill: Any, script: Any, args: Any = None) -> str:
|
||||
captured["args"] = args
|
||||
return "list_result"
|
||||
|
||||
script = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/run.py", runner=runner)
|
||||
skill = FileSkill(
|
||||
frontmatter=SkillFrontmatter(name="my-skill", description="test"),
|
||||
content="Body",
|
||||
path=f"{_ABS}/test",
|
||||
scripts=[script],
|
||||
)
|
||||
|
||||
provider = SkillsProvider([skill])
|
||||
await _init_provider(provider)
|
||||
run_tool = next(t for t in _ctx(provider)[2] if hasattr(t, "name") and t.name == "run_skill_script")
|
||||
result = await run_tool.func(skill_name="my-skill", script_name="run.py", args=["input.docx", "--verbose"])
|
||||
assert result == "list_result"
|
||||
assert captured["args"] == ["input.docx", "--verbose"]
|
||||
|
||||
async def test_run_skill_script_inline_with_list_args_returns_error(self) -> None:
|
||||
"""Inline script called with list args through provider returns error (TypeError caught)."""
|
||||
skill = InlineSkill(frontmatter=SkillFrontmatter(name="my-skill", description="test"), instructions="body")
|
||||
skill.scripts.append(InlineSkillScript(name="s1", function=lambda: "ok"))
|
||||
|
||||
provider = SkillsProvider([skill])
|
||||
await _init_provider(provider)
|
||||
run_tool = next(t for t in _ctx(provider)[2] if hasattr(t, "name") and t.name == "run_skill_script")
|
||||
result = await run_tool.func(skill_name="my-skill", script_name="s1", args=["arg1"])
|
||||
assert "Error" in result
|
||||
assert "Failed to run" in result
|
||||
|
||||
def test_file_skill_content_includes_scripts_block(self) -> None:
|
||||
"""FileSkill.content appends a <scripts> block when scripts are present."""
|
||||
script = FileSkillScript(name="run.py", full_path=f"{_ABS}/test/run.py")
|
||||
skill = FileSkill(
|
||||
frontmatter=SkillFrontmatter(name="my-skill", description="test"),
|
||||
content="---\nname: my-skill\n---\nBody",
|
||||
path=f"{_ABS}/test",
|
||||
scripts=[script],
|
||||
)
|
||||
assert "<scripts>" in skill.content
|
||||
assert 'name="run.py"' in skill.content
|
||||
assert "<parameters_schema>" in skill.content
|
||||
assert '"type": "array"' in skill.content
|
||||
|
||||
def test_file_skill_content_no_scripts_no_block(self) -> None:
|
||||
"""FileSkill.content does not append a <scripts> block when no scripts."""
|
||||
skill = FileSkill(
|
||||
frontmatter=SkillFrontmatter(name="my-skill", description="test"),
|
||||
content="---\nname: my-skill\n---\nBody",
|
||||
path=f"{_ABS}/test",
|
||||
)
|
||||
assert "<scripts>" not in skill.content
|
||||
|
||||
@@ -10,7 +10,7 @@ import os
|
||||
# warnings.filterwarnings("ignore", message=r"\[SKILLS\].*", category=FutureWarning)
|
||||
from textwrap import dedent
|
||||
|
||||
from agent_framework import Agent, ClassSkill, SkillsProvider
|
||||
from agent_framework import Agent, ClassSkill, SkillFrontmatter, SkillsProvider
|
||||
from agent_framework.foundry import FoundryChatClient
|
||||
from azure.identity import AzureCliCredential
|
||||
from dotenv import load_dotenv
|
||||
@@ -49,10 +49,12 @@ class UnitConverterSkill(ClassSkill):
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(
|
||||
name="unit-converter",
|
||||
description=(
|
||||
"Convert between common units using a multiplication factor. "
|
||||
"Use when asked to convert miles, kilometers, pounds, or kilograms."
|
||||
frontmatter=SkillFrontmatter(
|
||||
name="unit-converter",
|
||||
description=(
|
||||
"Convert between common units using a multiplication factor. "
|
||||
"Use when asked to convert miles, kilometers, pounds, or kilograms."
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@@ -20,30 +20,43 @@ from typing import Any
|
||||
from agent_framework import FileSkill, FileSkillScript
|
||||
|
||||
|
||||
def subprocess_script_runner(skill: FileSkill, script: FileSkillScript, args: dict[str, Any] | None = None) -> str:
|
||||
def subprocess_script_runner(
|
||||
skill: FileSkill, script: FileSkillScript, args: dict[str, Any] | list[str] | None = None
|
||||
) -> str:
|
||||
"""Run a skill script as a local Python subprocess.
|
||||
Uses ``FileSkillScript.full_path`` as the script path, converts the
|
||||
``args`` dict to CLI flags, and returns captured output.
|
||||
``args`` to CLI arguments, and returns captured output.
|
||||
Args:
|
||||
skill: The file-based skill that owns the script.
|
||||
script: The file-based script to run.
|
||||
args: Optional arguments forwarded as CLI flags.
|
||||
args: Optional arguments. A ``list[str]`` is forwarded as
|
||||
positional CLI arguments. Passing a ``dict`` or any other
|
||||
type raises :class:`TypeError` — file-based scripts expect
|
||||
positional arguments as a JSON array of strings.
|
||||
Returns:
|
||||
The combined stdout/stderr output, or an error message.
|
||||
Raises:
|
||||
TypeError: If ``args`` is not a ``list[str]`` or ``None``, or if
|
||||
any list element is not a string.
|
||||
"""
|
||||
script_path = Path(script.full_path)
|
||||
if not script_path.is_file():
|
||||
return f"Error: Script file not found: {script_path}"
|
||||
cmd = [sys.executable, str(script_path)]
|
||||
# Convert args dict to CLI flags
|
||||
if args:
|
||||
for key, value in args.items():
|
||||
if isinstance(value, bool):
|
||||
if value:
|
||||
cmd.append(f"--{key}")
|
||||
elif value is not None:
|
||||
cmd.append(f"--{key}")
|
||||
cmd.append(str(value))
|
||||
if isinstance(args, list):
|
||||
for item in args:
|
||||
if not isinstance(item, str):
|
||||
raise TypeError(
|
||||
f"File-based skill scripts only accept string CLI arguments "
|
||||
f"but received a {type(item).__name__}. "
|
||||
f"All array elements must be strings."
|
||||
)
|
||||
cmd.extend(args)
|
||||
elif args is not None:
|
||||
raise TypeError(
|
||||
f"Expected a list of CLI arguments but received {type(args).__name__}. "
|
||||
f"File-based skill scripts expect positional arguments as a list of strings."
|
||||
)
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
|
||||
Reference in New Issue
Block a user