feat(skills): lazily load L2 instructions and L3 resources from disk

src/google/adk/skills/_utils.py

@@ -20,6 +20,7 @@
 import io
 import logging
 import pathlib
+from typing import Any
 from typing import Callable
 from typing import Dict
 from typing import Union
@@ -83,6 +84,19 @@ def _validate_relative_resource_path(path: str) -> str:
   return normalized
 
 
+def _scan_dir(directory: pathlib.Path) -> dict[str, pathlib.Path]:
+  """Recursively collect file paths from a directory without reading them."""
+  files: dict[str, pathlib.Path] = {}
+  if directory.exists() and directory.is_dir():
+    for file_path in directory.rglob("*"):
+      if "__pycache__" in file_path.parts:
+        continue
+      if file_path.is_file():
+        relative_path = file_path.relative_to(directory)
+        files[relative_path.as_posix()] = file_path
+  return files
+
+
 def _load_dir(directory: pathlib.Path) -> dict[str, Union[str, bytes]]:
   """Recursively load files from a directory into a dictionary.
 
@@ -96,20 +110,10 @@ def _load_dir(directory: pathlib.Path) -> dict[str, Union[str, bytes]]:
     Dictionary mapping relative file paths to their content: a `str` for
     UTF-8 text files, or raw `bytes` for binary files.
   """
-  files: dict[str, Union[str, bytes]] = {}
-  if directory.exists() and directory.is_dir():
-    for file_path in directory.rglob("*"):
-      if "__pycache__" in file_path.parts:
-        continue
-      if file_path.is_file():
-        relative_path = file_path.relative_to(directory)
-        try:
-          files[relative_path.as_posix()] = file_path.read_text(
-              encoding="utf-8"
-          )
-        except UnicodeDecodeError:
-          files[relative_path.as_posix()] = file_path.read_bytes()
-  return files
+  return {
+      key: models._read_resource_file(path)
+      for key, path in _scan_dir(directory).items()
+  }
 
 
 def _build_scripts(
@@ -169,6 +173,36 @@ def _parse_skill_md_content(content: str) -> tuple[dict, str]:
   return parsed, body
 
 
+def _find_skill_md(skill_dir: pathlib.Path) -> pathlib.Path:
+  """Locate SKILL.md or skill.md in a skill directory."""
+  if not skill_dir.is_dir():
+    raise FileNotFoundError(f"Skill directory '{skill_dir}' not found.")
+
+  for name in ("SKILL.md", "skill.md"):
+    path = skill_dir / name
+    if path.exists():
+      return path
+
+  raise FileNotFoundError(f"SKILL.md not found in '{skill_dir}'.")
+
+
+def _parse_skill_md_frontmatter(skill_md: pathlib.Path) -> dict[str, Any]:
+  """Parse only the YAML frontmatter from a SKILL.md file."""
+  with skill_md.open(encoding="utf-8") as f:
+    buf = ""
+    while True:
+      chunk = f.read(4096)
+      if not chunk:
+        break
+      buf += chunk
+      if not buf.startswith("---"):
+        raise ValueError("SKILL.md must start with YAML frontmatter (---)")
+      if "---" in buf[3:]:
+        break
+  parsed, _ = _parse_skill_md_content(buf)
+  return parsed
+
+
 def _parse_skill_md(
     skill_dir: pathlib.Path,
 ) -> tuple[dict, str, pathlib.Path]:
@@ -184,19 +218,7 @@ def _parse_skill_md(
     FileNotFoundError: If the directory or SKILL.md is not found.
     ValueError: If SKILL.md is invalid.
   """
-  if not skill_dir.is_dir():
-    raise FileNotFoundError(f"Skill directory '{skill_dir}' not found.")
-
-  skill_md = None
-  for name in ("SKILL.md", "skill.md"):
-    path = skill_dir / name
-    if path.exists():
-      skill_md = path
-      break
-
-  if skill_md is None:
-    raise FileNotFoundError(f"SKILL.md not found in '{skill_dir}'.")
-
+  skill_md = _find_skill_md(skill_dir)
   content = skill_md.read_text(encoding="utf-8")
   parsed, body = _parse_skill_md_content(content)
 
@@ -210,7 +232,8 @@ def _load_skill_from_dir(skill_dir: Union[str, pathlib.Path]) -> models.Skill:
     skill_dir: Path to the skill directory.
 
   Returns:
-    Skill object with all components loaded.
+    Skill object with frontmatter loaded and instructions/resources loaded on
+    demand.
 
   Raises:
     FileNotFoundError: If the skill directory or SKILL.md is not found.
@@ -219,7 +242,8 @@ def _load_skill_from_dir(skill_dir: Union[str, pathlib.Path]) -> models.Skill:
   """
   skill_dir = pathlib.Path(skill_dir).resolve()
 
-  parsed, body, skill_md = _parse_skill_md(skill_dir)
+  skill_md = _find_skill_md(skill_dir)
+  parsed = _parse_skill_md_frontmatter(skill_md)
 
   # Use model_validate to handle aliases like allowed-tools
   frontmatter = models.Frontmatter.model_validate(parsed)
@@ -231,22 +255,20 @@ def _load_skill_from_dir(skill_dir: Union[str, pathlib.Path]) -> models.Skill:
         f" name '{skill_dir.name}'."
     )
 
-  references = _load_dir(skill_dir / "references")
-  assets = _load_dir(skill_dir / "assets")
-  scripts = _build_scripts(_load_dir(skill_dir / "scripts"))
-
-  resources = models.Resources(
-      references=references,
-      assets=assets,
-      scripts=scripts,
-  )
+  resources = models.Resources()
+  resources._reference_paths = _scan_dir(skill_dir / "references")
+  resources._asset_paths = _scan_dir(skill_dir / "assets")
+  resources._script_paths = _scan_dir(skill_dir / "scripts")
 
   skill = models.Skill(
       frontmatter=frontmatter,
-      instructions=body,
+      instructions="",
       resources=resources,
   )
   skill._uri = skill_dir.as_uri()
+  skill._instructions_loader = lambda: _parse_skill_md_content(
+      skill_md.read_text(encoding="utf-8")
+  )[1]
   return skill
 
 
@@ -487,7 +509,7 @@ def _validate_skill_dir(
     return [f"SKILL.md not found in '{skill_dir}'."]
 
   try:
-    parsed, _, _ = _parse_skill_md(skill_dir)
+    parsed = _parse_skill_md_frontmatter(skill_md)
   except (FileNotFoundError, ValueError) as e:
     return [str(e)]
 
@@ -529,7 +551,7 @@ def _read_skill_properties(
     ValueError: If the frontmatter is invalid.
   """
   skill_dir = pathlib.Path(skill_dir).resolve()
-  parsed, _, _ = _parse_skill_md(skill_dir)
+  parsed = _parse_skill_md_frontmatter(_find_skill_md(skill_dir))
   return models.Frontmatter.model_validate(parsed)
 
 

src/google/adk/skills/models.py

@@ -16,8 +16,11 @@
 
 from __future__ import annotations
 
+import logging
+import pathlib
 import re
 from typing import Any
+from typing import Callable
 from typing import Optional
 import unicodedata
 
@@ -146,6 +149,14 @@ def __str__(self) -> str:
     return self.src
 
 
+def _read_resource_file(path: pathlib.Path) -> str | bytes:
+  """Reads a resource file as UTF-8 text or falls back to raw bytes."""
+  try:
+    return path.read_text(encoding="utf-8")
+  except UnicodeDecodeError:
+    return path.read_bytes()
+
+
 class Resources(BaseModel):
   """L3 skill content: additional instructions, assets, and scripts.
 
@@ -161,6 +172,24 @@ class Resources(BaseModel):
   assets: dict[str, str | bytes] = {}
   scripts: dict[str, Script] = {}
 
+  _reference_paths: dict[str, pathlib.Path] = {}
+  _asset_paths: dict[str, pathlib.Path] = {}
+  _script_paths: dict[str, pathlib.Path] = {}
+
+  def __getattribute__(self, name: str) -> Any:
+    if name in ("references", "assets", "scripts"):
+      private = object.__getattribute__(self, "__pydantic_private__") or {}
+      if name == "references" and private.get("_reference_paths"):
+        for key in list(private["_reference_paths"]):
+          self.get_reference(key)
+      elif name == "assets" and private.get("_asset_paths"):
+        for key in list(private["_asset_paths"]):
+          self.get_asset(key)
+      elif name == "scripts" and private.get("_script_paths"):
+        for key in list(private["_script_paths"]):
+          self.get_script(key)
+    return super().__getattribute__(name)
+
   def get_reference(self, reference_id: str) -> Optional[str | bytes]:
     """Get content of a reference file.
 
@@ -170,7 +199,13 @@ def get_reference(self, reference_id: str) -> Optional[str | bytes]:
     Returns:
         Reference content as string or bytes, or None if not found
     """
-    return self.references.get(reference_id)
+    refs = object.__getattribute__(self, "__dict__")["references"]
+    if reference_id in refs:
+      return refs[reference_id]
+    path = self._reference_paths.pop(reference_id, None)
+    if path is not None:
+      refs[reference_id] = _read_resource_file(path)
+    return refs.get(reference_id)
 
   def get_asset(self, asset_id: str) -> Optional[str | bytes]:
     """Get content of an asset file.
@@ -181,7 +216,13 @@ def get_asset(self, asset_id: str) -> Optional[str | bytes]:
     Returns:
         Asset content as string or bytes, or None if not found
     """
-    return self.assets.get(asset_id)
+    assets = object.__getattribute__(self, "__dict__")["assets"]
+    if asset_id in assets:
+      return assets[asset_id]
+    path = self._asset_paths.pop(asset_id, None)
+    if path is not None:
+      assets[asset_id] = _read_resource_file(path)
+    return assets.get(asset_id)
 
   def get_script(self, script_id: str) -> Optional[Script]:
     """Get content of a script file.
@@ -192,19 +233,33 @@ def get_script(self, script_id: str) -> Optional[Script]:
     Returns:
         Script object, or None if not found
     """
-    return self.scripts.get(script_id)
+    scripts = object.__getattribute__(self, "__dict__")["scripts"]
+    if script_id in scripts:
+      return scripts[script_id]
+    path = self._script_paths.pop(script_id, None)
+    if path is not None:
+      try:
+        scripts[script_id] = Script(src=path.read_text(encoding="utf-8"))
+      except UnicodeDecodeError:
+        logging.warning("Skipping non-UTF-8 skill script '%s'.", script_id)
+        return None
+    return scripts.get(script_id)
 
   def list_references(self) -> list[str]:
     """List all available reference paths."""
-    return list(self.references.keys())
+    refs = object.__getattribute__(self, "__dict__")["references"]
+    return list(dict.fromkeys([*refs, *self._reference_paths]))
 
   def list_assets(self) -> list[str]:
     """List all available asset paths."""
-    return list(self.assets.keys())
+    assets = object.__getattribute__(self, "__dict__")["assets"]
+    return list(dict.fromkeys([*assets, *self._asset_paths]))
 
   def list_scripts(self) -> list[str]:
     """List all available script paths."""
-    return list(self.scripts.keys())
+    for key in list(self._script_paths):
+      self.get_script(key)
+    return list(object.__getattribute__(self, "__dict__")["scripts"].keys())
 
 
 class Skill(BaseModel):
@@ -230,6 +285,18 @@ class Skill(BaseModel):
   """Location the skill was loaded from, used for telemetry.
   Should be compliant with RFC 3986.
   """
+  _instructions_loader: Optional[Callable[[], str]] = None
+
+  def __getattribute__(self, name: str) -> Any:
+    if name == "instructions":
+      private = object.__getattribute__(self, "__pydantic_private__") or {}
+      loader = private.get("_instructions_loader")
+      if loader is not None:
+        d = object.__getattribute__(self, "__dict__")
+        if not d.get("instructions"):
+          d["instructions"] = loader()
+        private["_instructions_loader"] = None
+    return super().__getattribute__(name)
 
   @property
   def name(self) -> str:

tests/unittests/skills/test__utils.py

@@ -1400,3 +1400,52 @@ def test__validate_relative_resource_path_rejects_dot_prefixed_windows_drive():
   for unsafe_path in ("./C:/evil.py", ".\\C:\\evil.py", "./c:evil.py"):
     with pytest.raises(ValueError, match="Unsafe path"):
       _validate_relative_resource_path(unsafe_path)
+
+
+def test__load_skill_from_dir_defers_instructions_and_resources_until_accessed(
+    tmp_path,
+):
+  """L2 instructions and L3 resources are read on demand, not at load time."""
+  skill_dir = tmp_path / "lazy-skill"
+  skill_dir.mkdir()
+  (skill_dir / "SKILL.md").write_text(
+      "---\nname: lazy-skill\ndescription: Lazy desc\n---\nLazy instructions"
+  )
+  ref_dir = skill_dir / "references"
+  ref_dir.mkdir()
+  (ref_dir / "ref1.md").write_text("ref1 content")
+  (ref_dir / "ref2.md").write_text("ref2 content")
+  assets_dir = skill_dir / "assets"
+  assets_dir.mkdir()
+  (assets_dir / "asset1.txt").write_text("asset1 content")
+  scripts_dir = skill_dir / "scripts"
+  scripts_dir.mkdir()
+  (scripts_dir / "script1.sh").write_text("echo lazy")
+
+  read_files: list[str] = []
+  real_read_text = pathlib.Path.read_text
+
+  def tracking_read_text(self, *args, **kwargs):
+    read_files.append(self.name)
+    return real_read_text(self, *args, **kwargs)
+
+  with mock.patch.object(pathlib.Path, "read_text", tracking_read_text):
+    skill = _load_skill_from_dir(skill_dir)
+    assert skill.name == "lazy-skill"
+    assert skill.description == "Lazy desc"
+    assert read_files == []
+
+    # Listing references and assets does not read file contents.
+    assert sorted(skill.resources.list_references()) == ["ref1.md", "ref2.md"]
+    assert skill.resources.list_assets() == ["asset1.txt"]
+    assert read_files == []
+
+    # Fetching a single reference reads only that file and caches it.
+    assert skill.resources.get_reference("ref1.md") == "ref1 content"
+    assert skill.resources.get_reference("ref1.md") == "ref1 content"
+    assert read_files == ["ref1.md"]
+
+    # Accessing instructions reads SKILL.md on demand and caches it.
+    assert skill.instructions == "Lazy instructions"
+    assert skill.instructions == "Lazy instructions"
+    assert read_files == ["ref1.md", "SKILL.md"]