manager

manager

SkillManager — coordinates skill discovery, catalog, tool wrapping, and execution.

Classes

SkillManager

SkillManager(bus: EventBus, *, capability_policy: Optional[Any] = None, overlay_dir: Optional[Path] = None)

Coordinate skill discovery, resolution, catalog generation, and execution.

PARAMETER	DESCRIPTION
bus	Event bus for publishing lifecycle events. TYPE: EventBus
capability_policy	Optional capability policy passed through to tool executors. TYPE: Optional[Any] DEFAULT: None

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

def __init__(
    self,
    bus: EventBus,
    *,
    capability_policy: Optional[Any] = None,
    overlay_dir: Optional[Path] = None,
) -> None:
    self._bus = bus
    self._capability_policy = capability_policy
    self._skills: Dict[str, SkillManifest] = {}
    self._tool_executor: Optional[ToolExecutor] = None
    if overlay_dir is None:
        # Try to read from config first; fall back to the default
        # ~/.openjarvis/learning/skills/ if config can't be loaded.
        try:
            from openjarvis.core.config import load_config

            cfg = load_config()
            cfg_dir = getattr(
                getattr(cfg.learning, "skills", None),
                "overlay_dir",
                None,
            )
            if cfg_dir:
                overlay_dir = Path(cfg_dir).expanduser()
        except Exception:
            pass
        if overlay_dir is None:
            overlay_dir = Path("~/.openjarvis/learning/skills/").expanduser()
    self._overlay_dir = Path(overlay_dir).expanduser()

Functions

discover

discover(paths: Optional[List[Path]] = None) -> None

Scan directories in order and register skills.

First-seen name wins (workspace path listed first = highest precedence). After loading, the full dependency graph is validated and any sidecar overlays in overlay_dir are applied to discovered skills.

PARAMETER	DESCRIPTION
paths	Directories to scan. If None or empty, no skills are loaded from disk — but _load_overlays() still runs (in case the caller had previously seeded self._skills directly). TYPE: Optional[List[Path]] DEFAULT: None

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

def discover(self, paths: Optional[List[Path]] = None) -> None:
    """Scan directories in order and register skills.

    First-seen name wins (workspace path listed first = highest precedence).
    After loading, the full dependency graph is validated and any
    sidecar overlays in ``overlay_dir`` are applied to discovered skills.

    Parameters
    ----------
    paths:
        Directories to scan.  If *None* or empty, no skills are loaded
        from disk — but ``_load_overlays()`` still runs (in case the
        caller had previously seeded ``self._skills`` directly).
    """
    if paths:
        for directory in paths:
            manifests = discover_skills(directory)
            for manifest in manifests:
                # First-seen wins: do not overwrite an already-registered skill
                if manifest.name not in self._skills:
                    self._skills[manifest.name] = manifest

        # Validate the dependency graph after loading skills
        if self._skills:
            validate_dependencies(self._skills)

    # Load optimization overlays (Plan 2A) — always runs, even when no
    # paths are provided, so callers can apply overlays to skills loaded
    # via other means.
    self._load_overlays()

resolve

resolve(name: str) -> SkillManifest

Return the manifest for a skill by name.

RAISES	DESCRIPTION
KeyError	If name is not registered.

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

def resolve(self, name: str) -> SkillManifest:
    """Return the manifest for a skill by name.

    Raises
    ------
    KeyError
        If *name* is not registered.
    """
    if name not in self._skills:
        raise KeyError(f"Skill '{name}' not found")
    return self._skills[name]

skill_names

skill_names() -> List[str]

Return the names of all registered skills.

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

def skill_names(self) -> List[str]:
    """Return the names of all registered skills."""
    return list(self._skills.keys())

get_skill_tools

get_skill_tools(*, tool_executor: Optional[ToolExecutor] = None) -> List[BaseTool]

Wrap each registered skill as a :class:SkillTool (a :class:BaseTool).

PARAMETER	DESCRIPTION
tool_executor	Tool executor to use when running skill pipelines. Falls back to the one set via :meth:set_tool_executor if not provided here. TYPE: Optional[ToolExecutor] DEFAULT: None

RETURNS	DESCRIPTION
list[BaseTool]	One :class:SkillTool per registered skill.

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

def get_skill_tools(
    self, *, tool_executor: Optional[ToolExecutor] = None
) -> List[BaseTool]:
    """Wrap each registered skill as a :class:`SkillTool` (a :class:`BaseTool`).

    Parameters
    ----------
    tool_executor:
        Tool executor to use when running skill pipelines.  Falls back to
        the one set via :meth:`set_tool_executor` if not provided here.

    Returns
    -------
    list[BaseTool]
        One :class:`SkillTool` per registered skill.
    """
    executor = tool_executor or self._tool_executor
    tools: List[BaseTool] = []

    for manifest in self._skills.values():
        real_executor = executor or _NullToolExecutor()
        skill_exec = SkillExecutor(real_executor, bus=self._bus)

        # Wire sub-skill resolver so nested skill_name steps can delegate back
        skill_exec.set_skill_resolver(self._make_resolver())

        skill_tool = SkillTool(manifest, skill_exec, skill_manager=self)
        tools.append(skill_tool)

    return tools

get_catalog_xml

get_catalog_xml() -> str

Generate an <available_skills> XML catalog.

Skills with disable_model_invocation=True are excluded so that internal or automation-only skills are not surfaced to the model.

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

def get_catalog_xml(self) -> str:
    """Generate an ``<available_skills>`` XML catalog.

    Skills with ``disable_model_invocation=True`` are excluded so that
    internal or automation-only skills are not surfaced to the model.
    """
    lines: List[str] = ["<available_skills>"]

    for manifest in self._skills.values():
        if manifest.disable_model_invocation:
            continue
        escaped_name = html.escape(manifest.name)
        escaped_desc = html.escape(manifest.description or manifest.name)
        lines.append(
            f"  <skill name={escaped_name!r} description={escaped_desc!r} />"
        )

    lines.append("</available_skills>")
    return "\n".join(lines)

get_few_shot_examples

get_few_shot_examples() -> List[str]

Return formatted few-shot example strings ready for system prompt.

Pulls from manifest.metadata.openjarvis.few_shot for every registered skill. Returns one formatted string per example.

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

def get_few_shot_examples(self) -> List[str]:
    """Return formatted few-shot example strings ready for system prompt.

    Pulls from ``manifest.metadata.openjarvis.few_shot`` for every
    registered skill.  Returns one formatted string per example.
    """
    examples: List[str] = []
    for name, manifest in self._skills.items():
        oj = manifest.metadata.get("openjarvis", {}) if manifest.metadata else {}
        few_shot = oj.get("few_shot", []) or []
        for ex in few_shot:
            if not isinstance(ex, dict):
                continue
            inp = str(ex.get("input", ""))
            out = str(ex.get("output", ""))
            if inp or out:
                examples.append(f"### {name}\nInput: {inp}\nOutput: {out}")
    return examples

discover_from_traces

discover_from_traces(trace_store: Any, *, min_frequency: int = 3, min_outcome: float = 0.5, output_dir: Optional[Path] = None) -> List[Dict[str, Any]]

Mine the trace store for recurring tool sequences.

For each recurring sequence found by :class:SkillDiscovery, write a TOML skill manifest into output_dir (default ~/.openjarvis/skills/discovered/). Returns a list of dicts with name and path for each manifest written.

Names are normalized to spec-compliant kebab-case (lowercase with hyphens, no underscores) so the resulting manifests load cleanly through the discovery walker.

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

def discover_from_traces(
    self,
    trace_store: Any,
    *,
    min_frequency: int = 3,
    min_outcome: float = 0.5,
    output_dir: Optional[Path] = None,
) -> List[Dict[str, Any]]:
    """Mine the trace store for recurring tool sequences.

    For each recurring sequence found by :class:`SkillDiscovery`, write
    a TOML skill manifest into *output_dir* (default
    ``~/.openjarvis/skills/discovered/``).  Returns a list of dicts with
    ``name`` and ``path`` for each manifest written.

    Names are normalized to spec-compliant kebab-case (lowercase with
    hyphens, no underscores) so the resulting manifests load cleanly
    through the discovery walker.
    """
    from openjarvis.learning.agents.skill_discovery import SkillDiscovery

    traces = trace_store.list_traces(limit=10000)
    discovery = SkillDiscovery(
        min_frequency=min_frequency,
        min_outcome=min_outcome,
    )
    discovered = discovery.analyze_traces(traces)

    if output_dir is None:
        output_dir = Path("~/.openjarvis/skills/discovered/").expanduser()
    output_dir = Path(output_dir).expanduser()
    output_dir.mkdir(parents=True, exist_ok=True)

    written: List[Dict[str, Any]] = []
    for skill in discovered:
        name = self._normalize_skill_name(skill.name)
        skill_subdir = output_dir / name
        skill_subdir.mkdir(parents=True, exist_ok=True)
        toml_path = skill_subdir / "skill.toml"
        toml_path.write_text(
            self._serialize_discovered_skill(name, skill),
            encoding="utf-8",
        )
        written.append({"name": name, "path": str(toml_path)})

    return written

execute

execute(name: str, context: Optional[Dict[str, Any]] = None) -> SkillResult

Resolve and execute a skill by name.

PARAMETER	DESCRIPTION
name	Skill name to execute. TYPE: str
context	Initial context dict passed to the executor. TYPE: Optional[Dict[str, Any]] DEFAULT: None

RETURNS	DESCRIPTION
SkillResult

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

def execute(
    self,
    name: str,
    context: Optional[Dict[str, Any]] = None,
) -> SkillResult:
    """Resolve and execute a skill by name.

    Parameters
    ----------
    name:
        Skill name to execute.
    context:
        Initial context dict passed to the executor.

    Returns
    -------
    SkillResult
    """
    manifest = self.resolve(name)
    executor = SkillExecutor(
        self._tool_executor or _NullToolExecutor(),
        bus=self._bus,
    )
    executor.set_skill_resolver(self._make_resolver())
    return executor.run(manifest, initial_context=context)

set_tool_executor

set_tool_executor(tool_executor: ToolExecutor) -> None

Attach a :class:ToolExecutor for running tool steps in skill pipelines.

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

def set_tool_executor(self, tool_executor: ToolExecutor) -> None:
    """Attach a :class:`ToolExecutor` for running tool steps in skill pipelines."""
    self._tool_executor = tool_executor

find_installed_paths

find_installed_paths(name: str, *, roots: Optional[List[Path]] = None) -> List[Path]

Return on-disk skill directories matching name.

A directory matches when it contains skill.toml or SKILL.md and either the directory name equals name or its parsed manifest's name field equals name.

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

def find_installed_paths(
    self, name: str, *, roots: Optional[List[Path]] = None
) -> List[Path]:
    """Return on-disk skill directories matching ``name``.

    A directory matches when it contains ``skill.toml`` or ``SKILL.md``
    and either the directory name equals ``name`` or its parsed
    manifest's ``name`` field equals ``name``.
    """
    if roots is None:
        roots = [Path("~/.openjarvis/skills/").expanduser(), Path("./skills")]

    matches: List[Path] = []
    for root in roots:
        if not root.exists():
            continue
        for candidate in root.rglob("*"):
            if not candidate.is_dir():
                continue
            toml = candidate / "skill.toml"
            md = candidate / "SKILL.md"
            if not (toml.exists() or md.exists()):
                continue
            if candidate.name == name:
                matches.append(candidate)
                continue
            # Fall back to parsed manifest name
            try:
                from openjarvis.skills.loader import load_skill_directory

                manifest = load_skill_directory(candidate)
                if manifest is not None and manifest.name == name:
                    matches.append(candidate)
            except Exception:
                continue
    return matches

remove

remove(name: str, *, roots: Optional[List[Path]] = None) -> List[Path]

Remove an installed skill by name.

Returns the list of directories that were removed. Raises :class:FileNotFoundError when no matching skill exists on disk.

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

def remove(self, name: str, *, roots: Optional[List[Path]] = None) -> List[Path]:
    """Remove an installed skill by name.

    Returns the list of directories that were removed.  Raises
    :class:`FileNotFoundError` when no matching skill exists on disk.
    """
    import shutil

    paths = self.find_installed_paths(name, roots=roots)
    if not paths:
        raise FileNotFoundError(f"No installed skill named {name!r}")
    for p in paths:
        shutil.rmtree(p)
    # Drop from in-memory catalog
    self._skills.pop(name, None)
    return paths

Functions