Coverage for haystack/skill_stores/file_system/skill_store.py: 99%
95 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
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
17SKILL_FILE_NAME = "SKILL.md"
19# Non-text, non-image MIME types that are returned as `FileContent` so a multimodal LLM can ingest them directly.
20# Scoped to PDF, which is what the major providers accept as a file input; everything else falls back to text.
21_SUPPORTED_FILE_MIME_TYPES = {"application/pdf"}
24def _parse_frontmatter(text: str) -> tuple[dict[str, Any], str]:
25 """
26 Split a `SKILL.md` file into its YAML frontmatter and markdown body.
28 The frontmatter is the YAML block delimited by a leading and a trailing line containing exactly `---`.
29 If the first line is not `---`, no frontmatter is present and an empty mapping and the original text
30 are returned.
32 :param text: The full contents of a `SKILL.md` file.
33 :returns: A tuple of (frontmatter mapping, body).
34 :raises ValueError: If the frontmatter is opened with `---` but never closed, is not valid YAML, or is
35 not a YAML mapping.
36 """
37 lines = text.lstrip().split("\n")
38 if lines[0].rstrip() != "---":
39 return {}, text
41 # Find the closing delimiter: the next line containing exactly '---'.
42 closing_index = next((i for i, line in enumerate(lines[1:], start=1) if line.rstrip() == "---"), None)
43 if closing_index is None:
44 raise ValueError("Skill frontmatter is opened with '---' but never closed with a matching '---' line.")
46 frontmatter_block = "\n".join(lines[1:closing_index])
47 body = "\n".join(lines[closing_index + 1 :])
48 try:
49 loaded = yaml.safe_load(frontmatter_block) or {}
50 except yaml.YAMLError as e:
51 raise ValueError(f"Skill frontmatter is not valid YAML: {e}") from e
52 if not isinstance(loaded, dict):
53 raise ValueError("Skill frontmatter must be a YAML mapping.") # noqa: TRY004
54 return loaded, body.lstrip("\n")
57class FileSystemSkillStore:
58 """
59 SkillStore backed by a directory of skill sub-directories on the local filesystem.
61 Expected layout:
63 ```
64 skills/
65 pdf-forms/
66 SKILL.md # frontmatter (name, description) + markdown instructions
67 reference/forms.md # optional bundled file
68 ```
70 The skill catalog is built by reading the frontmatter of each `SKILL.md` on `warm_up`; bodies and bundled files
71 are read lazily when the agent calls the corresponding tool.
72 """
74 def __init__(self, skills_dir: str | Path) -> None:
75 """
76 Initialize the store with the root directory to scan.
78 No filesystem access happens here; the directory is scanned lazily on first use (see `warm_up`), so the store
79 can be constructed cheaply.
81 :param skills_dir: Root directory that contains one sub-directory per skill.
82 """
83 self.skills_dir = Path(skills_dir)
84 # Public metadata catalog returned by `list_skills`, populated on warm_up.
85 self._skills: dict[str, SkillInfo] = {}
86 # Private locator: maps each skill name to its directory, used to read content lazily.
87 self._skill_dirs: dict[str, Path] = {}
88 self._is_warmed_up = False
90 def warm_up(self) -> None:
91 """
92 Scan `skills_dir` and build the skill catalog by reading each skill's `SKILL.md` frontmatter.
94 Only the frontmatter is read here; bodies and bundled files are read lazily when the corresponding method is
95 called. Idempotent: repeated calls after the first are no-ops.
97 :raises ValueError: If `skills_dir` does not exist, is not a directory, a skill's frontmatter is missing,
98 malformed, or missing a required field, or two skills share the same name.
99 """
100 if self._is_warmed_up:
101 return
102 if not self.skills_dir.is_dir():
103 raise ValueError(f"Skills directory '{self.skills_dir}' does not exist or is not a directory.")
105 # Build into locals and swap at the end: if the scan fails halfway, no partial state is left behind (a retry
106 # after fixing the offending skill starts clean), and concurrent callers only ever observe either an empty or
107 # a complete catalog.
108 skills: dict[str, SkillInfo] = {}
109 skill_dirs: dict[str, Path] = {}
110 for skill_file in sorted(self.skills_dir.glob(f"*/{SKILL_FILE_NAME}")):
111 skill_dir = skill_file.parent
112 frontmatter, _ = _parse_frontmatter(skill_file.read_text(encoding="utf-8"))
114 name = frontmatter.get("name", skill_dir.name)
115 description = frontmatter.get("description")
116 if not description:
117 raise ValueError(f"Skill '{name}' ({skill_file}) is missing a 'description' in its frontmatter.")
118 if name in skills:
119 raise ValueError(f"Duplicate skill name '{name}' found in '{self.skills_dir}'.")
121 skills[name] = SkillInfo(name=name, description=description)
122 skill_dirs[name] = skill_dir
124 self._skills = skills
125 self._skill_dirs = skill_dirs
126 self._is_warmed_up = True
128 def _skill_dir(self, name: str) -> Path:
129 """
130 Return the directory of the named skill, warming up the store first if needed.
132 :param name: Skill name as returned by `list_skills`.
133 :returns: The skill's directory.
134 :raises KeyError: If no skill with `name` exists.
135 """
136 self.warm_up()
137 try:
138 return self._skill_dirs[name]
139 except KeyError:
140 available = ", ".join(self._skills) or "none"
141 # We suppress the original error since we are replacing it with a more informative error message
142 raise KeyError(f"Unknown skill '{name}'. Available skills: {available}.") from None
144 def _readable_files_hint(self, name: str) -> str:
145 """
146 Return a human-readable list of the files that can be read from the named skill.
148 Used to make `read_skill_file` errors actionable by telling the caller which paths are valid.
150 :param name: Skill name as returned by `list_skills`.
151 :returns: Comma-separated relative paths, or `"none"` if the skill bundles no readable files.
152 """
153 return ", ".join(self._list_skill_files(name)) or "none"
155 def list_skills(self) -> dict[str, SkillInfo]:
156 """
157 Return all skills discovered on disk, warming up the store first if needed.
159 :returns: Mapping of skill name to its metadata.
160 :raises ValueError: If the skills directory is invalid or a skill's frontmatter is malformed.
161 """
162 self.warm_up()
163 # We return a copy to prevent callers from mutating our internal state.
164 return dict(self._skills)
166 def load_skill(self, name: str) -> tuple[str, list[str]]:
167 """
168 Read the named skill's instruction body and the manifest of its bundled files.
170 :param name: Skill name as returned by `list_skills`.
171 :returns: A tuple of (markdown body of the skill's `SKILL.md` with frontmatter stripped, sorted list of
172 POSIX-style paths relative to the skill directory for any bundled files). The file list is empty when
173 the skill bundles no extras.
174 :raises KeyError: If no skill with `name` exists.
175 """
176 _, body = _parse_frontmatter((self._skill_dir(name) / SKILL_FILE_NAME).read_text(encoding="utf-8"))
177 return body, self._list_skill_files(name)
179 def _list_skill_files(self, name: str) -> list[str]:
180 """
181 Return the relative paths of all files bundled with the named skill, excluding its `SKILL.md`.
183 :param name: Skill name as returned by `list_skills`.
184 :returns: Sorted list of POSIX-style paths relative to the skill directory. Empty when there are none.
185 :raises KeyError: If no skill with `name` exists.
186 """
187 skill_dir = self._skill_dir(name)
188 return sorted(
189 p.relative_to(skill_dir).as_posix()
190 for p in skill_dir.rglob("*")
191 if p.is_file() and p.name != SKILL_FILE_NAME
192 )
194 def read_skill_file(self, name: str, path: str) -> str | ImageContent | FileContent:
195 """
196 Read a file bundled with the named skill, preventing path traversal outside the skill directory.
198 The return type depends on the file: text files are returned as a `str`, image files (PNG, JPEG, ...) as an
199 `ImageContent`, and PDFs as a `FileContent`, so a multimodal agent can pass them straight to the model.
201 :param name: Skill name as returned by `list_skills`.
202 :param path: Path of the file relative to the skill directory (e.g. `"reference/forms.md"`).
203 :returns: The file's text content (`str`), an `ImageContent` for images, or a `FileContent` for PDFs.
204 :raises KeyError: If no skill with `name` exists.
205 :raises PermissionError: If `path` resolves outside the skill's directory (path-traversal attempt). The
206 message lists the readable files so the caller can retry with a valid path.
207 :raises FileNotFoundError: If the file does not exist within the skill. The message lists the readable
208 files so the caller can retry with a valid path.
209 :raises ValueError: If the file is binary but not a supported image or PDF (i.e. not UTF-8 text either).
210 """
211 skill_dir = self._skill_dir(name).resolve()
212 target = (skill_dir / path).resolve()
213 if skill_dir != target and skill_dir not in target.parents:
214 raise PermissionError(
215 f"Cannot read '{path}' from skill '{name}': the path resolves outside the skill directory. "
216 f"Use a path relative to the skill root. Readable files: {self._readable_files_hint(name)}."
217 )
218 if not target.is_file():
219 raise FileNotFoundError(
220 f"File '{path}' not found in skill '{name}'. Readable files: {self._readable_files_hint(name)}."
221 )
223 # Check file types (PDF) before images: IMAGE_MIME_TYPES includes "application/pdf" (ImageContent can
224 # rasterize PDFs), but a skill's bundled PDF should reach the model as a FileContent, not an image.
225 mime_type, _ = mimetypes.guess_type(target.as_posix())
226 if mime_type in _SUPPORTED_FILE_MIME_TYPES:
227 encoded = base64.b64encode(target.read_bytes()).decode("utf-8")
228 return FileContent(base64_data=encoded, mime_type=mime_type, filename=target.name, validation=False)
229 if mime_type in IMAGE_MIME_TYPES:
230 encoded = base64.b64encode(target.read_bytes()).decode("utf-8")
231 return ImageContent(base64_image=encoded, mime_type=mime_type, validation=False)
232 try:
233 return target.read_text(encoding="utf-8")
234 except UnicodeDecodeError as e:
235 raise ValueError(
236 f"File '{path}' in skill '{name}' is not a readable asset. Only UTF-8 text, images, and PDFs "
237 f"are supported."
238 ) from e
240 def to_dict(self) -> dict[str, Any]:
241 """
242 Serialize this store to a dictionary for use with `from_dict`.
244 :returns: Dictionary representation of the store.
245 """
246 return default_to_dict(self, skills_dir=str(self.skills_dir))
248 @classmethod
249 def from_dict(cls, data: dict[str, Any]) -> "FileSystemSkillStore":
250 """
251 Deserialize a `FileSystemSkillStore` from its dictionary representation.
253 :param data: Dictionary representation of the store, as produced by `to_dict`.
254 :returns: A new `FileSystemSkillStore` instance.
255 """
256 return default_from_dict(cls, data)