tool_adapter

tool_adapter

SkillTool — wraps a skill as a tool that agents can invoke.

Classes

SkillTool

SkillTool(manifest: SkillManifest, executor: SkillExecutor, *, skill_manager: Optional[Any] = None)

Bases: BaseTool

Wraps a SkillManifest as a BaseTool that agents can invoke.

Follows the same adapter pattern as MCPToolAdapter.

PARAMETER	DESCRIPTION
manifest	The skill manifest to wrap. TYPE: SkillManifest
executor	A :class:SkillExecutor used to run the skill pipeline. TYPE: SkillExecutor
skill_manager	Optional skill manager (reserved for sub-skill delegation). TYPE: Optional[Any] DEFAULT: None

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

def __init__(
    self,
    manifest: SkillManifest,
    executor: SkillExecutor,
    *,
    skill_manager: Optional[Any] = None,
) -> None:
    self._manifest = manifest
    self._executor = executor
    self._skill_manager = skill_manager
    self.tool_id = f"skill_{manifest.name}"
    self._parameters = self._build_parameters()

Functions

execute

execute(**params: Any) -> ToolResult

Execute the skill.

If the manifest has pipeline steps, run them via the executor and collect the last step's output. If markdown_content is present, append it to the content. Returns a combined :class:ToolResult.

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

def execute(self, **params: Any) -> ToolResult:
    """Execute the skill.

    If the manifest has pipeline steps, run them via the executor and
    collect the last step's output.  If ``markdown_content`` is present,
    append it to the content.  Returns a combined :class:`ToolResult`.
    """
    tool_name = self.spec.name
    content_parts: List[str] = []

    if self._manifest.steps:
        # Build initial context from all supplied params
        initial_ctx: Dict[str, Any] = {k: v for k, v in params.items()}

        result = self._executor.run(self._manifest, initial_context=initial_ctx)

        if not result.success:
            # Propagate failure immediately
            error_content = (
                result.step_results[-1].content
                if result.step_results
                else "Pipeline failed with no step results."
            )
            return ToolResult(
                tool_name=tool_name,
                content=error_content,
                success=False,
                metadata=self._build_result_metadata(
                    steps_run=len(result.step_results)
                ),
            )

        # Use the last step's output as the primary content
        if result.step_results:
            last_step = result.step_results[-1]
            last_output = (
                result.context.get(
                    # Prefer the keyed output if available
                    self._manifest.steps[-1].output_key,
                    last_step.content,
                )
                if self._manifest.steps[-1].output_key
                else last_step.content
            )
            content_parts.append(str(last_output))

    # Append markdown content if present (hybrid or instruction-only)
    if self._manifest.markdown_content:
        content_parts.append(self._manifest.markdown_content)

    combined = "\n\n".join(filter(None, content_parts))

    return ToolResult(
        tool_name=tool_name,
        content=combined,
        success=True,
        metadata=self._build_result_metadata(steps_run=len(self._manifest.steps)),
    )