feat(skills): add shell command execution in frontmatter for dynamic context

openhands-sdk/openhands/sdk/context/skills/__init__.py

@@ -14,7 +14,7 @@
     KeywordTrigger,
     TaskTrigger,
 )
-from openhands.sdk.context.skills.types import SkillKnowledge
+from openhands.sdk.context.skills.types import CommandSpec, SkillKnowledge
 from openhands.sdk.context.skills.utils import (
     RESOURCE_DIRECTORIES,
     discover_skill_resources,
@@ -23,6 +23,7 @@
 
 
 __all__ = [
+    "CommandSpec",
     "Skill",
     "SkillResources",
     "BaseTrigger",

openhands-sdk/openhands/sdk/context/skills/execute.py

@@ -0,0 +1,95 @@
+"""Command execution for dynamic skill context injection."""
+
+from __future__ import annotations
+
+import re
+import subprocess
+from pathlib import Path
+from typing import Final
+
+from openhands.sdk.context.skills.exceptions import SkillError
+from openhands.sdk.context.skills.types import CommandSpec
+from openhands.sdk.logger import get_logger
+
+
+logger = get_logger(__name__)
+
+# 50KB per command
+MAX_OUTPUT_SIZE: Final[int] = 50 * 1024
+
+
+def _execute_command(spec: CommandSpec, working_dir: Path | None = None) -> str:
+    """Execute a single command and return its output."""
+    cwd = str(working_dir) if working_dir else None
+    try:
+        result = subprocess.run(
+            spec.command,
+            shell=True,
+            cwd=cwd,
+            capture_output=True,
+            text=True,
+            timeout=spec.timeout,
+        )
+        if result.returncode != 0:
+            return _handle_error(
+                spec, f"Command exited with code {result.returncode}: {result.stderr}"
+            )
+        output = result.stdout.strip()
+        if len(output.encode()) > MAX_OUTPUT_SIZE:
+            output = output.encode()[:MAX_OUTPUT_SIZE].decode("utf-8", errors="ignore")
+            output += "\n... [output truncated]"
+        return output
+
+    except subprocess.TimeoutExpired:
+        return _handle_error(spec, f"Command timed out after {spec.timeout}s")
+    except Exception as e:
+        return _handle_error(spec, f"Failed to execute command: {e}")
+
+
+def _handle_error(spec: CommandSpec, message: str) -> str:
+    """Handle command execution error based on on_error setting."""
+    if spec.on_error == "fail":
+        raise SkillError(message)
+    logger.warning("Skill command '%s' failed: %s", spec.name, message)
+    if spec.on_error == "empty":
+        return ""
+    return f"[Error: {message}]"
+
+
+def _execute_commands(
+    commands: list[CommandSpec],
+    working_dir: Path | None = None,
+) -> dict[str, str]:
+    """Execute all commands and return name->output mapping."""
+    return {spec.name: _execute_command(spec, working_dir) for spec in commands}
+
+
+def render_content_with_commands(
+    content: str,
+    commands: list[CommandSpec],
+    working_dir: Path | None = None,
+    extra_vars: dict[str, str] | None = None,
+) -> str:
+    """Execute commands and substitute {{var_name}} patterns in content."""
+    if not commands and not extra_vars:
+        return content
+
+    # Execute commands
+    variables = _execute_commands(commands, working_dir) if commands else {}
+    if extra_vars:
+        collisions = set(variables) & set(extra_vars)
+        if collisions:
+            logger.warning("extra_vars overriding command outputs: %s", collisions)
+        variables.update(extra_vars)
+
+    if not variables:
+        return content
+
+    # Substitute {{var_name}} patterns
+    def replace_var(match: re.Match[str]) -> str:
+        var_name = match.group(1)
+        if var_name in variables:
+            return variables[var_name]
+        return match.group(0)
+
+    return re.sub(r"\{\{(\w+)\}\}", replace_var, content)

openhands-sdk/openhands/sdk/context/skills/skill.py

@@ -15,7 +15,7 @@
     KeywordTrigger,
     TaskTrigger,
 )
-from openhands.sdk.context.skills.types import InputMetadata
+from openhands.sdk.context.skills.types import CommandSpec, InputMetadata
 from openhands.sdk.context.skills.utils import (
     discover_skill_resources,
     find_mcp_config,
@@ -207,6 +207,13 @@ class Skill(BaseModel):
             "AgentSkills standard field. Only populated for SKILL.md directory format."
         ),
     )
+    commands: list[CommandSpec] = Field(
+        default_factory=list,
+        description=(
+            "Shell commands to execute for dynamic context injection. "
+            "Command outputs are available as {{name}} template variables in content."
+        ),
+    )
 
     _DESCRIPTION_TRUNCATE_NOTICE = (
         "<response clipped><NOTE>Due to the max output limit, only part of "
@@ -421,6 +428,14 @@ def _create_skill_from_metadata(
             k: v for k, v in agentskills_fields.items() if v is not None
         }
 
+        # Parse commands for dynamic context injection
+        commands: list[CommandSpec] = []
+        commands_raw = metadata_dict.get("commands", [])
+        if commands_raw:
+            if not isinstance(commands_raw, list):
+                raise SkillValidationError("commands must be a list")
+            commands = [CommandSpec.model_validate(c) for c in commands_raw]
+
         # Get trigger keywords from metadata
         keywords = metadata_dict.get("triggers", [])
         if not isinstance(keywords, list):
@@ -450,6 +465,7 @@ def _create_skill_from_metadata(
                 mcp_tools=mcp_tools,
                 resources=resources,
                 is_agentskills_format=is_agentskills_format,
+                commands=commands,
                 **agentskills_fields,
             )
 
@@ -462,6 +478,7 @@ def _create_skill_from_metadata(
                 mcp_tools=mcp_tools,
                 resources=resources,
                 is_agentskills_format=is_agentskills_format,
+                commands=commands,
                 **agentskills_fields,
             )
         else:
@@ -474,6 +491,7 @@ def _create_skill_from_metadata(
                 mcp_tools=mcp_tools,
                 resources=resources,
                 is_agentskills_format=is_agentskills_format,
+                commands=commands,
                 **agentskills_fields,
             )
 
@@ -625,6 +643,33 @@ def to_skill_info(self) -> SkillInfo:
             is_agentskills_format=self.is_agentskills_format,
         )
 
+    def render_content(
+        self,
+        working_dir: Path | None = None,
+        extra_vars: dict[str, str] | None = None,
+    ) -> str:
+        """Render skill content with command execution and variable substitution.
+
+        If commands are defined, executes them and substitutes {{name}} patterns
+        in content with command outputs.
+
+        Args:
+            working_dir: Directory to run commands in.
+            extra_vars: Additional variables for substitution.
+
+        Returns:
+            Processed content with variables substituted.
+        """
+        if not self.commands and not extra_vars:
+            return self.content
+
+        # Lazy import to avoid circular dependency: skill -> execute -> types -> skill
+        from openhands.sdk.context.skills.execute import render_content_with_commands
+
+        return render_content_with_commands(
+            self.content, self.commands, working_dir, extra_vars
+        )
+
 
 def load_skills_from_dir(
     skill_dir: str | Path,

openhands-sdk/openhands/sdk/context/skills/types.py

@@ -1,8 +1,34 @@
 from datetime import UTC, datetime
+from typing import Literal
 
 from pydantic import BaseModel, Field
 
 
+class CommandSpec(BaseModel):
+    """Specification for a shell command to execute for dynamic context.
+
+    **Security Warning**: Commands are executed via shell with full process
+    privileges. Only use with trusted skill sources. Avoid loading skills
+    with commands from untrusted third parties.
+    """
+
+    name: str = Field(description="Variable name for template substitution")
+    command: str = Field(description="Shell command to execute")
+    timeout: float = Field(default=10.0, gt=0.0, description="Timeout in seconds")
+    on_error: Literal["fail", "empty", "message"] = Field(
+        default="message",
+        description=(
+            "Error handling strategy when the command fails, i.e,"
+            "non-zero exit code or timeout."
+            "- 'fail' raises an exception, aborting skill loading. "
+            "- 'empty' silently returns an empty string, useful when the "
+            "command output is optional. "
+            "- 'message' returns a human-readable error description, "
+            "allowing the skill to proceed with diagnostic context."
+        ),
+    )
+
+
 class InputMetadata(BaseModel):
     """Metadata for task skill inputs."""
 

tests/sdk/context/skill/test_skill_commands.py

@@ -0,0 +1,103 @@
+"""Tests for shell command execution in skill frontmatter."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+from openhands.sdk.context.skills import CommandSpec, Skill, SkillValidationError
+from openhands.sdk.context.skills.exceptions import SkillError
+from openhands.sdk.context.skills.execute import (
+    _execute_command,
+    render_content_with_commands,
+)
+
+
+def test_command_spec_defaults():
+    """CommandSpec should have sensible defaults."""
+    spec = CommandSpec(name="test", command="echo hello")
+    assert spec.timeout == 10.0 and spec.on_error == "message"
+
+
+@pytest.mark.parametrize(
+    ("command", "on_error", "timeout", "check"),
+    [
+        ("echo hello", "message", 10.0, lambda r: r == "hello"),
+        ("exit 1", "message", 10.0, lambda r: "[Error:" in r),
+        ("exit 1", "empty", 10.0, lambda r: r == ""),
+        ("sleep 5", "message", 0.1, lambda r: "timed out" in r),
+    ],
+)
+def test_execute_command(
+    command: str, on_error: str, timeout: float, check: Callable[[str], bool]
+):
+    """_execute_command handles success, failure, and timeout correctly."""
+    spec = CommandSpec(name="t", command=command, on_error=on_error, timeout=timeout)  # type: ignore[arg-type]
+    assert check(_execute_command(spec))
+
+
+def test_execute_command_failure_raises():
+    """on_error='fail' should raise SkillError."""
+    with pytest.raises(SkillError):
+        _execute_command(CommandSpec(name="fail", command="exit 1", on_error="fail"))
+
+
+@pytest.mark.parametrize(
+    ("content", "commands", "extra_vars", "expected"),
+    [
+        ("Hi {{x}}", [], None, "Hi {{x}}"),
+        ("Hi {{x}}", [CommandSpec(name="x", command="echo world")], None, "Hi world"),
+        ("Hi {{x}}", [], {"x": "world"}, "Hi world"),
+        ("Hi {{x}}", [], {"y": "z"}, "Hi {{x}}"),
+    ],
+)
+def test_render_content(
+    content: str, commands: list, extra_vars: dict | None, expected: str
+):
+    """render_content_with_commands substitutes variables correctly."""
+    assert (
+        render_content_with_commands(content, commands, extra_vars=extra_vars)
+        == expected
+    )
+
+
+def test_skill_load_with_commands(tmp_path: Path):
+    """Skill.load parses commands from frontmatter."""
+    skill_md = tmp_path / "test-skill" / "SKILL.md"
+    skill_md.parent.mkdir()
+    skill_md.write_text(
+        "---\nname: test-skill\ncommands:\n"
+        "  - name: a\n    command: echo A\n"
+        "  - name: b\n    command: echo B\n    timeout: 5.0\n---\n{{a}} {{b}}"
+    )
+    skill = Skill.load(skill_md)
+    assert len(skill.commands) == 2 and skill.commands[1].timeout == 5.0
+
+
+def test_skill_load_commands_validation(tmp_path: Path):
+    """Skill.load rejects invalid commands field."""
+    path = tmp_path / "bad.md"
+    path.write_text("---\nname: s\ncommands: not-a-list\n---\n#")
+    with pytest.raises(SkillValidationError, match="commands must be a list"):
+        Skill.load(path)
+
+
+@pytest.mark.parametrize(
+    ("content", "commands", "expected"),
+    [
+        ("Hello", [], "Hello"),
+        ("Out: {{x}}", [CommandSpec(name="x", command="echo hi")], "Out: hi"),
+    ],
+)
+def test_skill_render_content(content: str, commands: list, expected: str):
+    """Skill.render_content executes commands and substitutes."""
+    assert (
+        Skill(name="t", content=content, commands=commands).render_content() == expected
+    )
+
+
+def test_execute_command_respects_working_dir(tmp_path: Path):
+    """Commands should run in the specified working directory."""
+    spec = CommandSpec(name="cwd", command="pwd")
+    result = _execute_command(spec, working_dir=tmp_path)
+    assert result == str(tmp_path.resolve())