Index

skills

Skill system — reusable multi-tool compositions.

Classes

SkillExecutor

SkillExecutor(tool_executor: ToolExecutor, *, bus: Optional[EventBus] = None)

Execute a skill manifest step-by-step.

Each step's arguments_template supports {key} placeholders that are resolved from the context dict (populated by prior step outputs).

Source code in src/openjarvis/skills/executor.py

def __init__(
    self,
    tool_executor: ToolExecutor,
    *,
    bus: Optional[EventBus] = None,
) -> None:
    self._tool_executor = tool_executor
    self._bus = bus

Functions

run

run(manifest: SkillManifest, *, initial_context: Optional[Dict[str, Any]] = None) -> SkillResult

Execute all steps in a skill manifest.

Source code in src/openjarvis/skills/executor.py

def run(
    self,
    manifest: SkillManifest,
    *,
    initial_context: Optional[Dict[str, Any]] = None,
) -> SkillResult:
    """Execute all steps in a skill manifest."""
    ctx: Dict[str, Any] = dict(initial_context or {})
    all_results: List[ToolResult] = []

    if self._bus:
        self._bus.publish(
            EventType.SKILL_EXECUTE_START,
            {"skill": manifest.name, "steps": len(manifest.steps)},
        )

    for i, step in enumerate(manifest.steps):
        # Render template
        try:
            rendered = self._render_template(step.arguments_template, ctx)
        except Exception as exc:
            result = ToolResult(
                tool_name=step.tool_name,
                content=f"Template rendering error: {exc}",
                success=False,
            )
            all_results.append(result)
            break

        # Execute
        tool_call = ToolCall(
            id=f"skill_{manifest.name}_{i}",
            name=step.tool_name,
            arguments=rendered,
        )
        result = self._tool_executor.execute(tool_call)
        all_results.append(result)

        if not result.success:
            break

        # Store output in context
        if step.output_key:
            ctx[step.output_key] = result.content

    success = all(r.success for r in all_results)

    if self._bus:
        self._bus.publish(
            EventType.SKILL_EXECUTE_END,
            {"skill": manifest.name, "success": success},
        )

    return SkillResult(
        skill_name=manifest.name,
        success=success,
        step_results=all_results,
        context=ctx,
    )

SkillTool

SkillTool(manifest: SkillManifest, executor: SkillExecutor)

Bases: BaseTool

Wraps a SkillManifest as a BaseTool that agents can invoke.

Follows the same adapter pattern as MCPToolAdapter.

Source code in src/openjarvis/skills/tool_adapter.py

def __init__(
    self,
    manifest: SkillManifest,
    executor: SkillExecutor,
) -> None:
    self._manifest = manifest
    self._executor = executor
    self.tool_id = f"skill_{manifest.name}"

SkillManifest dataclass

SkillManifest(name: str, version: str = '0.1.0', description: str = '', author: str = '', steps: List[SkillStep] = list(), required_capabilities: List[str] = list(), signature: str = '', metadata: Dict[str, Any] = dict())

Manifest describing a reusable skill.

Functions

manifest_bytes

manifest_bytes() -> bytes

Serialize the manifest (excluding signature) for signing/verification.

Source code in src/openjarvis/skills/types.py

def manifest_bytes(self) -> bytes:
    """Serialize the manifest (excluding signature) for signing/verification."""
    import json
    data = {
        "name": self.name,
        "version": self.version,
        "description": self.description,
        "author": self.author,
        "steps": [
            {
                "tool_name": s.tool_name,
                "arguments_template": s.arguments_template,
                "output_key": s.output_key,
            }
            for s in self.steps
        ],
        "required_capabilities": self.required_capabilities,
    }
    return json.dumps(data, sort_keys=True).encode()

SkillStep dataclass

SkillStep(tool_name: str, arguments_template: str = '{}', output_key: str = '')

A single step in a skill pipeline.

Functions

load_skill

load_skill(path: str | Path, *, verify_signature: bool = False, public_key: Optional[bytes] = None, scan_for_injection: bool = False) -> SkillManifest

Load a skill manifest from a TOML file.

Expected format:

[skill]
name = "research_and_summarize"
version = "0.1.0"
description = "Search web and summarize results"
author = "openjarvis"
required_capabilities = ["network:fetch"]
signature = ""

[[skill.steps]]
tool_name = "web_search"
arguments_template = '{"query": "{query}"}'
output_key = "search_results"

[[skill.steps]]
tool_name = "think"
arguments_template = '{"thought": "Summarize: {search_results}"}'
output_key = "summary"

Source code in src/openjarvis/skills/loader.py

def load_skill(
    path: str | Path,
    *,
    verify_signature: bool = False,
    public_key: Optional[bytes] = None,
    scan_for_injection: bool = False,
) -> SkillManifest:
    """Load a skill manifest from a TOML file.

    Expected format:
    ```toml
    [skill]
    name = "research_and_summarize"
    version = "0.1.0"
    description = "Search web and summarize results"
    author = "openjarvis"
    required_capabilities = ["network:fetch"]
    signature = ""

    [[skill.steps]]
    tool_name = "web_search"
    arguments_template = '{"query": "{query}"}'
    output_key = "search_results"

    [[skill.steps]]
    tool_name = "think"
    arguments_template = '{"thought": "Summarize: {search_results}"}'
    output_key = "summary"
    ```
    """
    path = Path(path)
    with open(path, "rb") as fh:
        data = tomllib.load(fh)

    skill_data = data.get("skill", {})

    steps = []
    for step_data in skill_data.get("steps", []):
        steps.append(SkillStep(
            tool_name=step_data["tool_name"],
            arguments_template=step_data.get("arguments_template", "{}"),
            output_key=step_data.get("output_key", ""),
        ))

    manifest = SkillManifest(
        name=skill_data.get("name", path.stem),
        version=skill_data.get("version", "0.1.0"),
        description=skill_data.get("description", ""),
        author=skill_data.get("author", ""),
        steps=steps,
        required_capabilities=skill_data.get("required_capabilities", []),
        signature=skill_data.get("signature", ""),
        metadata=skill_data.get("metadata", {}),
    )

    # Verify signature if requested
    if verify_signature and public_key and manifest.signature:
        try:
            from openjarvis.security.signing import verify_b64
            valid = verify_b64(
                manifest.manifest_bytes(),
                manifest.signature,
                public_key,
            )
            if not valid:
                raise ValueError(f"Invalid signature for skill '{manifest.name}'")
        except ImportError:
            raise ImportError(
                "Signature verification requires 'cryptography'. "
                "Install with: uv sync --extra security-signing"
            )

    # Scan for prompt injection if requested
    if scan_for_injection:
        try:
            from openjarvis.security.scanner import SecretScanner
            scanner = SecretScanner()
            for step in manifest.steps:
                scan_result = scanner.scan(step.arguments_template)
                if scan_result.findings:
                    raise ValueError(
                        f"Potential prompt injection in skill '{manifest.name}', "
                        f"step '{step.tool_name}': "
                        f"{scan_result.findings[0].description}"
                    )
        except ImportError:
            pass

    return manifest