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 
31
32
33
34
35
36
37
38
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 
40
41
42
43
44
45
46
47
48
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
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,
    )