Coverage for haystack/skill_stores/file_system/skill_store.py: 98%
105 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-09 14:34 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import base64
6import mimetypes
7from pathlib import Path
8from typing import Any
10import yaml
12from haystack.core.serialization import default_from_dict, default_to_dict
13from haystack.dataclasses.file_content import FileContent
14from haystack.dataclasses.image_content import IMAGE_MIME_TYPES, ImageContent
15from haystack.dataclasses.skill_info import SkillInfo
16from haystack.utils import Secret
18SKILL_FILE_NAME = "SKILL.md"
20# Non-text, non-image MIME types that are returned as `FileContent` so a multimodal LLM can ingest them directly.
21# Scoped to PDF, which is what the major providers accept as a file input; everything else falls back to text.
22_SUPPORTED_FILE_MIME_TYPES = {"application/pdf"}
25def _parse_frontmatter(text: str) -> tuple[dict[str, Any], str]:
26 """
27 Split a `SKILL.md` file into its YAML frontmatter and markdown body.
29 The frontmatter is the YAML block delimited by a leading and a trailing line containing exactly `---`.
30 If the first line is not `---`, no frontmatter is present and an empty mapping and the original text
31 are returned.
33 :param text: The full contents of a `SKILL.md` file.
34 :returns: A tuple of (frontmatter mapping, body).
35 :raises ValueError: If the frontmatter is opened with `---` but never closed, is not valid YAML, or is
36 not a YAML mapping.
37 """
38 lines = text.lstrip().split("\n")
39 if lines[0].rstrip() != "---":
40 return {}, text
42 # Find the closing delimiter: the next line containing exactly '---'.
43 closing_index = next((i for i, line in enumerate(lines[1:], start=1) if line.rstrip() == "---"), None)
44 if closing_index is None:
45 raise ValueError("Skill frontmatter is opened with '---' but never closed with a matching '---' line.")
47 frontmatter_block = "\n".join(lines[1:closing_index])
48 body = "\n".join(lines[closing_index + 1 :])
49 try:
50 loaded = yaml.safe_load(frontmatter_block) or {}
51 except yaml.YAMLError as e:
52 raise ValueError(f"Skill frontmatter is not valid YAML: {e}") from e
53 if not isinstance(loaded, dict):
54 raise ValueError("Skill frontmatter must be a YAML mapping.") # noqa: TRY004
55 return loaded, body.lstrip("\n")
58class FileSystemSkillStore:
59 """
60 SkillStore backed by a directory of skill sub-directories on the local filesystem.
62 Expected layout:
64 ```
65 skills/
66 pdf-forms/
67 SKILL.md # frontmatter (name, description) + markdown instructions
68 reference/forms.md # optional bundled file
69 ```
71 The skill catalog is built by reading the frontmatter of each `SKILL.md` on `warm_up`; bodies and bundled files
72 are read lazily when the agent calls the corresponding tool.
73 """
75 def __init__(self, skills_dir: str | Path | Secret) -> None:
76 """
77 Initialize the store with the root directory to scan.
79 No filesystem access happens here; the directory is scanned lazily on first use (see `warm_up`), so the store
80 can be constructed cheaply.
82 :param skills_dir: Root directory that contains one sub-directory per skill. Can also be a `Secret`
83 (e.g. `Secret.from_env_var("SKILLS_DIR")`) to source the path from an environment variable rather than
84 hard-coding it.
85 """
86 self.skills_dir = skills_dir
87 # Public metadata catalog returned by `list_skills`, populated on warm_up.
88 self._skills: dict[str, SkillInfo] = {}
89 # Private locator: maps each skill name to its directory, used to read content lazily.
90 self._skill_dirs: dict[str, Path] = {}
91 self._is_warmed_up = False
93 def _resolve_skills_dir(self) -> Path:
94 """
95 Resolve `skills_dir` to a concrete filesystem path, reading the `Secret`'s value if one was passed.
97 :returns: The root directory that contains one sub-directory per skill.
98 """
99 if isinstance(self.skills_dir, Secret):
100 resolved = self.skills_dir.resolve_value()
101 if resolved is None:
102 raise ValueError("The 'skills_dir' secret did not resolve to a value.")
103 return Path(resolved)
104 return Path(self.skills_dir)
106 def warm_up(self) -> None:
107 """
108 Scan `skills_dir` and build the skill catalog by reading each skill's `SKILL.md` frontmatter.
110 Only the frontmatter is read here; bodies and bundled files are read lazily when the corresponding method is
111 called. Idempotent: repeated calls after the first are no-ops.
113 :raises ValueError: If `skills_dir` does not exist, is not a directory, a skill's frontmatter is missing,
114 malformed, or missing a required field, or two skills share the same name.
115 """
116 if self._is_warmed_up:
117 return
118 skills_dir = self._resolve_skills_dir()
119 if not skills_dir.is_dir():
120 raise ValueError(f"Skills directory '{skills_dir}' does not exist or is not a directory.")
122 # Build into locals and swap at the end: if the scan fails halfway, no partial state is left behind (a retry
123 # after fixing the offending skill starts clean), and concurrent callers only ever observe either an empty or
124 # a complete catalog.
125 skills: dict[str, SkillInfo] = {}
126 skill_dirs: dict[str, Path] = {}
127 for skill_file in sorted(skills_dir.glob(f"*/{SKILL_FILE_NAME}")):
128 skill_dir = skill_file.parent
129 frontmatter, _ = _parse_frontmatter(skill_file.read_text(encoding="utf-8"))
131 name = frontmatter.get("name", skill_dir.name)
132 description = frontmatter.get("description")
133 if not description:
134 raise ValueError(f"Skill '{name}' ({skill_file}) is missing a 'description' in its frontmatter.")
135 if name in skills:
136 raise ValueError(f"Duplicate skill name '{name}' found in '{skills_dir}'.")
138 skills[name] = SkillInfo(name=name, description=description)
139 skill_dirs[name] = skill_dir
141 self._skills = skills
142 self._skill_dirs = skill_dirs
143 self._is_warmed_up = True
145 def _skill_dir(self, name: str) -> Path:
146 """
147 Return the directory of the named skill, warming up the store first if needed.
149 :param name: Skill name as returned by `list_skills`.
150 :returns: The skill's directory.
151 :raises KeyError: If no skill with `name` exists.
152 """
153 self.warm_up()
154 try:
155 return self._skill_dirs[name]
156 except KeyError:
157 available = ", ".join(self._skills) or "none"
158 # We suppress the original error since we are replacing it with a more informative error message
159 raise KeyError(f"Unknown skill '{name}'. Available skills: {available}.") from None
161 def _readable_files_hint(self, name: str) -> str:
162 """
163 Return a human-readable list of the files that can be read from the named skill.
165 Used to make `read_skill_file` errors actionable by telling the caller which paths are valid.
167 :param name: Skill name as returned by `list_skills`.
168 :returns: Comma-separated relative paths, or `"none"` if the skill bundles no readable files.
169 """
170 return ", ".join(self._list_skill_files(name)) or "none"
172 def list_skills(self) -> dict[str, SkillInfo]:
173 """
174 Return all skills discovered on disk, warming up the store first if needed.
176 :returns: Mapping of skill name to its metadata.
177 :raises ValueError: If the skills directory is invalid or a skill's frontmatter is malformed.
178 """
179 self.warm_up()
180 # We return a copy to prevent callers from mutating our internal state.
181 return dict(self._skills)
183 def load_skill(self, name: str) -> tuple[str, list[str]]:
184 """
185 Read the named skill's instruction body and the manifest of its bundled files.
187 :param name: Skill name as returned by `list_skills`.
188 :returns: A tuple of (markdown body of the skill's `SKILL.md` with frontmatter stripped, sorted list of
189 POSIX-style paths relative to the skill directory for any bundled files). The file list is empty when
190 the skill bundles no extras.
191 :raises KeyError: If no skill with `name` exists.
192 """
193 _, body = _parse_frontmatter((self._skill_dir(name) / SKILL_FILE_NAME).read_text(encoding="utf-8"))
194 return body, self._list_skill_files(name)
196 def _list_skill_files(self, name: str) -> list[str]:
197 """
198 Return the relative paths of all files bundled with the named skill, excluding its `SKILL.md`.
200 :param name: Skill name as returned by `list_skills`.
201 :returns: Sorted list of POSIX-style paths relative to the skill directory. Empty when there are none.
202 :raises KeyError: If no skill with `name` exists.
203 """
204 skill_dir = self._skill_dir(name).resolve()
205 return sorted(
206 p.relative_to(skill_dir).as_posix()
207 for p in skill_dir.rglob("*")
208 if p.is_file() and p.name != SKILL_FILE_NAME and p.resolve().is_relative_to(skill_dir)
209 )
211 def read_skill_file(self, name: str, path: str) -> str | ImageContent | FileContent:
212 """
213 Read a file bundled with the named skill, preventing path traversal outside the skill directory.
215 The return type depends on the file: text files are returned as a `str`, image files (PNG, JPEG, ...) as an
216 `ImageContent`, and PDFs as a `FileContent`, so a multimodal agent can pass them straight to the model.
218 :param name: Skill name as returned by `list_skills`.
219 :param path: Path of the file relative to the skill directory (e.g. `"reference/forms.md"`).
220 :returns: The file's text content (`str`), an `ImageContent` for images, or a `FileContent` for PDFs.
221 :raises KeyError: If no skill with `name` exists.
222 :raises PermissionError: If `path` resolves outside the skill's directory (path-traversal attempt). The
223 message lists the readable files so the caller can retry with a valid path.
224 :raises FileNotFoundError: If the file does not exist within the skill. The message lists the readable
225 files so the caller can retry with a valid path.
226 :raises ValueError: If the file is binary but not a supported image or PDF (i.e. not UTF-8 text either).
227 """
228 skill_dir = self._skill_dir(name).resolve()
229 target = (skill_dir / path).resolve()
230 if skill_dir != target and skill_dir not in target.parents:
231 raise PermissionError(
232 f"Cannot read '{path}' from skill '{name}': the path resolves outside the skill directory. "
233 f"Use a path relative to the skill root. Readable files: {self._readable_files_hint(name)}."
234 )
235 if not target.is_file():
236 raise FileNotFoundError(
237 f"File '{path}' not found in skill '{name}'. Readable files: {self._readable_files_hint(name)}."
238 )
240 # Check file types (PDF) before images: IMAGE_MIME_TYPES includes "application/pdf" (ImageContent can
241 # rasterize PDFs), but a skill's bundled PDF should reach the model as a FileContent, not an image.
242 mime_type, _ = mimetypes.guess_type(target.as_posix())
243 if mime_type in _SUPPORTED_FILE_MIME_TYPES:
244 encoded = base64.b64encode(target.read_bytes()).decode("utf-8")
245 return FileContent(base64_data=encoded, mime_type=mime_type, filename=target.name, validation=False)
246 if mime_type in IMAGE_MIME_TYPES:
247 encoded = base64.b64encode(target.read_bytes()).decode("utf-8")
248 return ImageContent(base64_image=encoded, mime_type=mime_type, validation=False)
249 try:
250 return target.read_text(encoding="utf-8")
251 except UnicodeDecodeError as e:
252 raise ValueError(
253 f"File '{path}' in skill '{name}' is not a readable asset. Only UTF-8 text, images, and PDFs "
254 f"are supported."
255 ) from e
257 def to_dict(self) -> dict[str, Any]:
258 """
259 Serialize this store to a dictionary for use with `from_dict`.
261 :returns: Dictionary representation of the store.
262 """
263 skills_dir = self.skills_dir if isinstance(self.skills_dir, Secret) else str(self.skills_dir)
264 return default_to_dict(self, skills_dir=skills_dir)
266 @classmethod
267 def from_dict(cls, data: dict[str, Any]) -> "FileSystemSkillStore":
268 """
269 Deserialize a `FileSystemSkillStore` from its dictionary representation.
271 :param data: Dictionary representation of the store, as produced by `to_dict`.
272 :returns: A new `FileSystemSkillStore` instance.
273 """
274 return default_from_dict(cls, data)