Coverage for haystack/tools/skills/skill_toolset.py: 100%
61 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5from typing import Annotated, Any
7from haystack.core.serialization import generate_qualified_class_name
8from haystack.dataclasses.file_content import FileContent
9from haystack.dataclasses.image_content import ImageContent
10from haystack.dataclasses.skill_info import SkillInfo
11from haystack.skill_stores.types.protocol import SkillStore
12from haystack.tools.from_function import create_tool_from_function
13from haystack.tools.tool import Tool
14from haystack.tools.toolset import Toolset
15from haystack.utils.deserialization import deserialize_component_inplace
18class SkillToolset(Toolset):
19 """
20 A Toolset that lets an Agent discover and read skills via progressive disclosure.
22 A skill is a directory (or equivalent storage unit) containing a `SKILL.md` file with YAML frontmatter
23 (`name` and `description`) and a markdown body of instructions. Skills may bundle additional files
24 (reference docs, examples, templates).
26 - On `warm_up`, the name and description of every discovered skill are baked into the `load_skill` tool
27 description so the model knows which skills exist without any system prompt injection.
28 - `load_skill` returns a skill's full instructions on demand, plus a manifest of its bundled files.
29 - `read_skill_file` reads a bundled file on demand.
31 ### Usage example
33 <!-- test-concept -->
34 ```python
35 from haystack.components.agents import Agent
36 from haystack.components.generators.chat import OpenAIChatGenerator
37 from haystack.dataclasses import ChatMessage
38 from haystack.tools import SkillToolset
39 from haystack.skill_stores.file_system import FileSystemSkillStore
41 store = FileSystemSkillStore("skills/")
42 skills_toolset = SkillToolset(store)
43 agent = Agent(chat_generator=OpenAIChatGenerator(), tools=skills_toolset)
44 result = agent.run(messages=[ChatMessage.from_user("Fill in this PDF form for me.")])
45 ```
47 Expected filesystem layout:
49 ```
50 skills/
51 pdf-forms/
52 SKILL.md # frontmatter (name, description) + markdown instructions
53 reference/forms.md
54 ```
56 The tool names `load_skill` and `read_skill_file` are fixed, so an `Agent` can use at most one
57 `SkillToolset`. To serve skills from multiple sources, back a single toolset with a custom store that
58 merges them.
59 """
61 def __init__(self, store: SkillStore) -> None:
62 """
63 Initialize the SkillToolset.
65 Constructing the toolset does not read any skills. The store is queried for the available skills on
66 `warm_up()`, so stores that do I/O (reading a directory, connecting to a database) stay cheap to
67 construct.
69 The `load_skill` and `read_skill_file` tools are created right away, so the toolset can be used as a
70 collection (length, membership checks, iteration) immediately.
72 :param store: A `haystack.skill_stores.types.SkillStore` instance to back this toolset.
73 """
74 self._store = store
75 self._skills: dict[str, SkillInfo] = {}
76 self._is_warmed_up = False
78 # We create both tools now and dynamically update the `load_skill` description at warm-up with the discovered
79 # catalog
80 self._load_skill_tool = self._create_load_skill_tool()
81 super().__init__(tools=[self._load_skill_tool, self._create_read_skill_file_tool()])
83 @property
84 def skills(self) -> dict[str, SkillInfo]:
85 """Mapping of skill name to its metadata. Triggers `warm_up()` on first access if not already warmed up."""
86 if not self._is_warmed_up:
87 self.warm_up()
88 return self._skills
90 def warm_up(self) -> None:
91 """
92 Discover the available skills from the store and bake the catalog into the `load_skill` description.
94 Only the description content is dynamic, so the (static) tools created in `__init__` are reused; this
95 refreshes `load_skill`'s description once the catalog is known. Idempotent: repeated calls after the
96 first are no-ops.
97 """
98 if self._is_warmed_up:
99 return
100 if hasattr(self._store, "warm_up"):
101 self._store.warm_up()
102 self._skills = self._store.list_skills()
103 self._load_skill_tool.description = self._load_skill_description()
104 self._is_warmed_up = True
106 def add(self, tool: Tool | Toolset) -> None:
107 """Adding tools is not supported: a SkillToolset's tools are fixed and defined by its store."""
108 raise NotImplementedError(
109 "SkillToolset does not support adding tools. To combine it with other tools, pass it to the Agent "
110 "alongside them, e.g. tools=[skill_toolset, other_tool]."
111 )
113 def __add__(self, other: Tool | Toolset | list[Tool]) -> "Toolset":
114 """Concatenation is not supported for SkillToolset."""
115 raise NotImplementedError(
116 "SkillToolset does not support concatenation. To combine it with other tools, pass it to the Agent "
117 "alongside them, e.g. tools=[skill_toolset, other_tool]."
118 )
120 def _load_skill_description(self) -> str:
121 """
122 Build the `load_skill` tool description, including the catalog of discovered skills.
124 The available skills (name + description) are baked into the description so the model can see which skills
125 exist and decide when to load one, without relying on any system prompt injection.
127 :returns: The tool description text.
128 """
129 lines = [
130 "Load a skill's full instructions before doing a task it covers. Skills are specialized instruction "
131 "sets for specific task types; once loaded, follow them exactly (they override your general approach). "
132 "If a loaded skill references a bundled file, fetch it with `read_skill_file`."
133 ]
134 if self._skills:
135 lines += ["", "Available skills:"]
136 lines += [f"- {meta.name}: {meta.description}" for meta in self._skills.values()]
137 else:
138 lines += ["", "No skills are currently available."]
139 return "\n".join(lines)
141 def _create_load_skill_tool(self) -> Tool:
142 """Create the `load_skill` tool, closed over this toolset's store."""
144 def load_skill(name: Annotated[str, "Exact name of the skill to load, from the Available skills list."]) -> str:
145 # The store raises an actionable error (e.g. unknown skill) on failure. We let it propagate so the Agent
146 # applies its own tool-failure policy.
147 body, bundled = self._store.load_skill(name)
148 if bundled:
149 manifest = "\n".join(f"- {path}" for path in bundled)
150 body = f"{body}\n\nBundled files (read with `read_skill_file`):\n{manifest}"
151 return body
153 return create_tool_from_function(
154 function=load_skill, name="load_skill", description=self._load_skill_description()
155 )
157 def _create_read_skill_file_tool(self) -> Tool:
158 """Create the `read_skill_file` tool, closed over this toolset's store."""
160 def read_skill_file(
161 name: Annotated[str, "Name of the skill that owns the file."],
162 path: Annotated[str, "Path of the file relative to the skill directory, e.g. 'reference/forms.md'."],
163 ) -> str | list[ImageContent | FileContent]:
164 """Read a file bundled with a skill (reference docs, examples, templates, images, PDFs)."""
165 # The store raises an actionable error (e.g. unknown skill) on failure. We let it propagate so the Agent
166 # applies its own tool-failure policy.
167 content = self._store.read_skill_file(name, path)
168 # Text is returned as-is; images/PDFs are wrapped in a list so they ride back as multimodal tool-result
169 # content parts for the model to ingest directly.
170 return content if isinstance(content, str) else [content]
172 # raw_result keeps ImageContent/FileContent intact instead of stringifying them, so they reach the model as
173 # image/file content parts. This requires a multimodal-capable generator (e.g. OpenAIResponsesChatGenerator).
174 return create_tool_from_function(
175 function=read_skill_file, name="read_skill_file", outputs_to_string={"raw_result": True}
176 )
178 def to_dict(self) -> dict[str, Any]:
179 """
180 Serialize the toolset to a dictionary.
182 :returns: Dictionary representation of the toolset.
183 """
184 return {"type": generate_qualified_class_name(type(self)), "data": {"store": self._store.to_dict()}}
186 @classmethod
187 def from_dict(cls, data: dict[str, Any]) -> "SkillToolset":
188 """
189 Deserialize a toolset from a dictionary.
191 :param data: Dictionary representation of the toolset, as produced by `to_dict`.
192 :returns: A new SkillToolset instance.
193 """
194 inner_data = data["data"]
195 deserialize_component_inplace(inner_data, key="store")
196 return cls(**inner_data)