feat: Implement GCPSkillRegistry in ADK

This CL implements the  class in the integrations folder, used specifically for the Skill Registry API.

Co-authored-by: Kathy Wu <wukathy@google.com>
PiperOrigin-RevId: 915627057

Author: wukath

contributing/samples/gcp_skill_registry_agent/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from . import agent

contributing/samples/gcp_skill_registry_agent/agent.py

@@ -0,0 +1,40 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Sample agent demonstrating the use of GCPSkillRegistry."""
+
+from google.adk import Agent
+from google.adk.integrations.skill_registry import GCPSkillRegistry
+from google.adk.tools.skill_toolset import SkillToolset
+
+# Initialize GCP Skill Registry
+registry = GCPSkillRegistry(
+    project_id="your-project-id", location="us-central1"
+)
+
+# Initialize SkillToolset with registry
+skill_toolset = SkillToolset(skills=[], registry=registry)
+
+root_agent = Agent(
+    model="gemini-2.5-flash",
+    name="skill_registry_agent",
+    description=(
+        "An agent that can discover and load skills from GCP Skill Registry."
+    ),
+    instruction=(
+        "Use search_skills to find skills and load_skill to load them if"
+        " needed."
+    ),
+    tools=[skill_toolset],
+)

src/google/adk/integrations/skill_registry/__init__.py

@@ -0,0 +1,19 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Skill Registry integrations."""
+
+from .gcp_skill_registry import GCPSkillRegistry
+
+__all__ = ["GCPSkillRegistry"]

src/google/adk/integrations/skill_registry/gcp_skill_registry.py

@@ -0,0 +1,93 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""GCP Skill Registry implementation."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import os
+
+from google.adk.skills import _utils
+from google.adk.skills import models
+from google.adk.skills.skill_registry import SkillRegistry
+import vertexai
+
+
+class GCPSkillRegistry(SkillRegistry):
+  """GCP implementation of SkillRegistry using GCP Skill Registry API."""
+
+  def __init__(
+      self, *, project_id: str | None = None, location: str | None = None
+  ):
+    """Initializes the GCP Skill Registry.
+
+    Args:
+      project_id: Optional GCP project ID. If omitted, loads from environment.
+      location: Optional GCP location. If omitted, loads from environment.
+    """
+    self.project_id = project_id or os.environ.get("GOOGLE_CLOUD_PROJECT")
+    self.location = location or os.environ.get("GOOGLE_CLOUD_LOCATION")
+    self._client = vertexai.Client(
+        project=self.project_id,
+        location=self.location,
+        http_options={
+            "api_version": "v1beta1",
+        },
+    ).aio
+
+  async def get_skill(self, *, name: str) -> models.Skill:
+    """Fetches a skill from the registry.
+
+    Args:
+      name: The name of the skill.
+
+    Returns:
+      A Skill object.
+    """
+    full_name = (
+        f"projects/{self.project_id}/locations/{self.location}/skills/{name}"
+    )
+    skill_resource = await self._client.skills.get(name=full_name)
+
+    zip_bytes_base64 = skill_resource.zipped_filesystem
+    if not zip_bytes_base64:
+      raise ValueError(f"Skill '{name}' does not contain zipped filesystem.")
+
+    zip_bytes = base64.b64decode(zip_bytes_base64)
+
+    return await asyncio.to_thread(_utils._load_skill_from_zip_bytes, zip_bytes)
+
+  async def search_skills(self, *, query: str) -> list[models.Frontmatter]:
+    """Searches for skills in the registry.
+
+    Args:
+      query: The search query.
+
+    Returns:
+      A list of Frontmatter objects for discovery.
+    """
+    response = await self._client.skills.retrieve(query=query)
+
+    results = []
+    if response.retrieved_skills:
+      for s in response.retrieved_skills:
+        results.append(
+            models.Frontmatter(
+                name=s.skill_name.split("/")[-1] if s.skill_name else "",
+                description=s.description or "",
+            )
+        )
+    return results

src/google/adk/skills/_utils.py

@@ -16,9 +16,12 @@
 
 from __future__ import annotations
 
+import io
 import logging
 import pathlib
+from typing import Dict
 from typing import Union
+import zipfile
 
 from google.auth import credentials as auth
 from google.cloud import storage
@@ -177,6 +180,96 @@ def _load_skill_from_dir(skill_dir: Union[str, pathlib.Path]) -> models.Skill:
   )
 
 
+def _load_skill_from_zip_bytes(zip_bytes: bytes) -> models.Skill:
+  """Load a complete skill directly from in-memory zip file bytes.
+
+  Args:
+    zip_bytes: The raw bytes of the zip file containing the skill.
+
+  Returns:
+    Skill object with all components loaded.
+
+  Raises:
+    FileNotFoundError: If SKILL.md is not found in the archive.
+    ValueError: If SKILL.md is invalid or contains dangerous paths.
+  """
+  with zipfile.ZipFile(io.BytesIO(zip_bytes)) as z:
+    # Security check for zip slip
+    for member in z.infolist():
+      filename = member.filename
+      if (
+          filename.startswith("/")
+          or filename.startswith("../")
+          or "/../" in filename
+      ):
+        raise ValueError(f"Dangerous zip entry ignored: {filename}")
+
+    # Find SKILL.md or skill.md
+    skill_md_content = None
+    for name in ("SKILL.md", "skill.md"):
+      try:
+        skill_md_content = z.read(name).decode("utf-8")
+        break
+      except KeyError:
+        continue
+
+    if skill_md_content is None:
+      raise FileNotFoundError("SKILL.md not found in zipped filesystem.")
+
+    parsed, body = _parse_skill_md_content(skill_md_content)
+    skill_name = parsed.get("name")
+    if not skill_name:
+      raise ValueError("SKILL.md frontmatter must contain 'name'")
+    if (
+        not isinstance(skill_name, str)
+        or pathlib.Path(skill_name).name != skill_name
+    ):
+      raise ValueError(f"Invalid skill name in SKILL.md: {skill_name}")
+
+    frontmatter = models.Frontmatter.model_validate(parsed)
+
+    # Helper to load files under a directory prefix inside the zip
+    def _load_zip_dir(prefix: str) -> dict[str, str]:
+      result = {}
+      if not prefix.endswith("/"):
+        prefix += "/"
+      for info in z.infolist():
+        if info.is_dir():
+          continue
+        if info.filename.startswith(prefix):
+          # Avoid cache files or similar
+          if "__pycache__" in info.filename:
+            continue
+          relative_path = info.filename[len(prefix) :]
+          if not relative_path:
+            continue
+          try:
+            result[relative_path] = z.read(info).decode("utf-8")
+          except UnicodeDecodeError:
+            continue
+      return result
+
+    references = _load_zip_dir("references")
+    assets = _load_zip_dir("assets")
+    raw_scripts = _load_zip_dir("scripts")
+    scripts = {
+        name: models.Script(src=content)
+        for name, content in raw_scripts.items()
+    }
+
+    resources = models.Resources(
+        references=references,
+        assets=assets,
+        scripts=scripts,
+    )
+
+    return models.Skill(
+        frontmatter=frontmatter,
+        instructions=body,
+        resources=resources,
+    )
+
+
 def _validate_skill_dir(
     skill_dir: Union[str, pathlib.Path],
 ) -> list[str]: