# SPDX-FileCopyrightText: 2022-present deepset GmbH # # SPDX-License-Identifier: Apache-2.0 from typing import Annotated, Any from haystack.core.serialization import generate_qualified_class_name from haystack.dataclasses.file_content import FileContent from haystack.dataclasses.image_content import ImageContent from haystack.dataclasses.skill_info import SkillInfo from haystack.skill_stores.types.protocol import SkillStore from haystack.tools.from_function import create_tool_from_function from haystack.tools.tool import Tool from haystack.tools.toolset import Toolset from haystack.utils.deserialization import deserialize_component_inplace class SkillToolset(Toolset): """ A Toolset that lets an Agent discover and read skills via progressive disclosure. A skill is a directory (or equivalent storage unit) containing a `SKILL.md` file with YAML frontmatter (`name` and `description`) and a markdown body of instructions. Skills may bundle additional files (reference docs, examples, templates). - On `warm_up`, the name and description of every discovered skill are baked into the `load_skill` tool description so the model knows which skills exist without any system prompt injection. - `load_skill` returns a skill's full instructions on demand, plus a manifest of its bundled files. - `read_skill_file` reads a bundled file on demand. ### Usage example ```python from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.tools import SkillToolset from haystack.skill_stores.file_system import FileSystemSkillStore store = FileSystemSkillStore("skills/") skills_toolset = SkillToolset(store) agent = Agent(chat_generator=OpenAIChatGenerator(), tools=skills_toolset) result = agent.run(messages=[ChatMessage.from_user("Fill in this PDF form for me.")]) ``` Expected filesystem layout: ``` skills/ pdf-forms/ SKILL.md # frontmatter (name, description) + markdown instructions reference/forms.md ``` The tool names `load_skill` and `read_skill_file` are fixed, so an `Agent` can use at most one `SkillToolset`. To serve skills from multiple sources, back a single toolset with a custom store that merges them. """ def __init__(self, store: SkillStore) -> None: """ Initialize the SkillToolset. Constructing the toolset does not read any skills. The store is queried for the available skills on `warm_up()`, so stores that do I/O (reading a directory, connecting to a database) stay cheap to construct. The `load_skill` and `read_skill_file` tools are created right away, so the toolset can be used as a collection (length, membership checks, iteration) immediately. :param store: A `haystack.skill_stores.types.SkillStore` instance to back this toolset. """ self._store = store self._skills: dict[str, SkillInfo] = {} self._is_warmed_up = False # We create both tools now and dynamically update the `load_skill` description at warm-up with the discovered # catalog self._load_skill_tool = self._create_load_skill_tool() super().__init__(tools=[self._load_skill_tool, self._create_read_skill_file_tool()]) @property def skills(self) -> dict[str, SkillInfo]: """Mapping of skill name to its metadata. Triggers `warm_up()` on first access if not already warmed up.""" if not self._is_warmed_up: self.warm_up() return self._skills def warm_up(self) -> None: """ Discover the available skills from the store and bake the catalog into the `load_skill` description. Only the description content is dynamic, so the (static) tools created in `__init__` are reused; this refreshes `load_skill`'s description once the catalog is known. Idempotent: repeated calls after the first are no-ops. """ if self._is_warmed_up: return if hasattr(self._store, "warm_up"): self._store.warm_up() self._skills = self._store.list_skills() self._load_skill_tool.description = self._load_skill_description() self._is_warmed_up = True def add(self, tool: Tool | Toolset) -> None: """Adding tools is not supported: a SkillToolset's tools are fixed and defined by its store.""" raise NotImplementedError( "SkillToolset does not support adding tools. To combine it with other tools, pass it to the Agent " "alongside them, e.g. tools=[skill_toolset, other_tool]." ) def __add__(self, other: Tool | Toolset | list[Tool]) -> "Toolset": """Concatenation is not supported for SearchableToolset.""" raise NotImplementedError( "SkillToolset does not support concatenation. To combine it with other tools, pass it to the Agent " "alongside them, e.g. tools=[skill_toolset, other_tool]." ) def _load_skill_description(self) -> str: """ Build the `load_skill` tool description, including the catalog of discovered skills. The available skills (name + description) are baked into the description so the model can see which skills exist and decide when to load one, without relying on any system prompt injection. :returns: The tool description text. """ lines = [ "Load a skill's full instructions before doing a task it covers. Skills are specialized instruction " "sets for specific task types; once loaded, follow them exactly (they override your general approach). " "If a loaded skill references a bundled file, fetch it with `read_skill_file`." ] if self._skills: lines += ["", "Available skills:"] lines += [f"- {meta.name}: {meta.description}" for meta in self._skills.values()] else: lines += ["", "No skills are currently available."] return "\n".join(lines) def _create_load_skill_tool(self) -> Tool: """Create the `load_skill` tool, closed over this toolset's store.""" def load_skill(name: Annotated[str, "Exact name of the skill to load, from the Available skills list."]) -> str: # The store raises an actionable error (e.g. unknown skill) on failure. We let it propagate so the Agent # applies its own tool-failure policy. body, bundled = self._store.load_skill(name) if bundled: manifest = "\n".join(f"- {path}" for path in bundled) body = f"{body}\n\nBundled files (read with `read_skill_file`):\n{manifest}" return body return create_tool_from_function( function=load_skill, name="load_skill", description=self._load_skill_description() ) def _create_read_skill_file_tool(self) -> Tool: """Create the `read_skill_file` tool, closed over this toolset's store.""" def read_skill_file( name: Annotated[str, "Name of the skill that owns the file."], path: Annotated[str, "Path of the file relative to the skill directory, e.g. 'reference/forms.md'."], ) -> str | list[ImageContent | FileContent]: """Read a file bundled with a skill (reference docs, examples, templates, images, PDFs).""" # The store raises an actionable error (e.g. unknown skill) on failure. We let it propagate so the Agent # applies its own tool-failure policy. content = self._store.read_skill_file(name, path) # Text is returned as-is; images/PDFs are wrapped in a list so they ride back as multimodal tool-result # content parts for the model to ingest directly. return content if isinstance(content, str) else [content] # raw_result keeps ImageContent/FileContent intact instead of stringifying them, so they reach the model as # image/file content parts. This requires a multimodal-capable generator (e.g. OpenAIResponsesChatGenerator). return create_tool_from_function( function=read_skill_file, name="read_skill_file", outputs_to_string={"raw_result": True} ) def to_dict(self) -> dict[str, Any]: """ Serialize the toolset to a dictionary. :returns: Dictionary representation of the toolset. """ return {"type": generate_qualified_class_name(type(self)), "data": {"store": self._store.to_dict()}} @classmethod def from_dict(cls, data: dict[str, Any]) -> "SkillToolset": """ Deserialize a toolset from a dictionary. :param data: Dictionary representation of the toolset, as produced by `to_dict`. :returns: A new SkillToolset instance. """ inner_data = data["data"] deserialize_component_inplace(inner_data, key="store") return cls(**inner_data)