Skip to content

executor

executor

SkillExecutor — runs skill steps sequentially through ToolExecutor.

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 
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    tool_executor: ToolExecutor,
    *,
    bus: Optional[EventBus] = None,
) -> None:
    self._tool_executor = tool_executor
    self._bus = bus
    self._skill_resolver: Optional[SkillResolver] = None
Functions
set_skill_resolver 
set_skill_resolver(resolver: SkillResolver) -> None

Register a callback used to delegate skill_name steps.

Source code in src/openjarvis/skills/executor.py 
45
46
47
def set_skill_resolver(self, resolver: SkillResolver) -> None:
    """Register a callback used to delegate ``skill_name`` steps."""
    self._skill_resolver = resolver
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 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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):
        step_id = step.tool_name or step.skill_name

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

        if step.skill_name:
            # Delegate to sub-skill resolver
            result = self._run_sub_skill(
                step.skill_name, rendered, ctx, manifest.name, i
            )
        else:
            # Execute via tool executor
            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,
    )