feat: Implement Skill Registry interface in ADK

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

Author: wukath

src/google/adk/skills/skill_registry.py

@@ -0,0 +1,83 @@
+# 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.
+
+"""Interface for skill registry."""
+
+from __future__ import annotations
+
+import abc
+from typing import Any
+from typing import Dict
+from typing import List
+
+from . import models
+
+
+class SkillRegistry(abc.ABC):
+  """Interface for a skill registry."""
+
+  @abc.abstractmethod
+  async def get_skill(
+      self, *, name: str, version: str | None = None
+  ) -> models.Skill:
+    """Fetches a skill from the registry.
+
+    Args:
+        name: The name of the skill.
+        version: Optional version of the skill.
+
+    Returns:
+        A Skill object.
+    """
+    pass
+
+  @abc.abstractmethod
+  async def search_skills(
+      self,
+      *,
+      query: str,
+      filters: Dict[str, Any] | None = None,
+      **kwargs,
+  ) -> List[models.Frontmatter]:
+    """Searches for skills in the registry.
+
+    Args:
+        query: The search query.
+        filters: Optional filters.
+        **kwargs: Additional implementation-specific arguments.
+
+    Returns:
+        A list of Frontmatter objects for discovery.
+    """
+    pass
+
+  @abc.abstractmethod
+  def get_filter_schema(self) -> Dict[str, Any] | None:
+    """Returns the JSON schema for the filters supported by this registry.
+
+    Returns:
+        A JSON schema dict or None if filters are not supported
+    """
+    pass
+
+  def get_search_description(self) -> str:
+    """Returns the description for the search_skills tool.
+
+    Registries can override this to provide specialized instructions to the
+    model on how to use their specific search capabilities.
+    """
+    return (
+        "Searches for relevant skills in the registry based on a semantic or"
+        " keyword query."
+    )

src/google/adk/tools/skill_toolset.py

@@ -37,6 +37,7 @@
 from ..features import FeatureName
 from ..skills import models
 from ..skills import prompt
+from ..skills.skill_registry import SkillRegistry
 from .base_tool import BaseTool
 from .base_toolset import BaseToolset
 from .function_tool import FunctionTool
@@ -155,6 +156,20 @@ async def run_async(
       }
 
     skill = self._toolset._get_skill(skill_name)
+    if not skill and self._toolset._registry:
+      try:
+        skill = await self._toolset._registry.get_skill(name=skill_name)
+        if skill:
+          self._toolset._skills[skill_name] = skill
+      except Exception as e:
+        logger.exception(
+            "Failed to fetch skill '%s' from registry.", skill_name
+        )
+        return {
+            "error": f"Failed to fetch skill '{skill_name}' from registry: {e}",
+            "error_code": "REGISTRY_ERROR",
+        }
+
     if not skill:
       return {
           "error": f"Skill '{skill_name}' not found.",
@@ -771,14 +786,75 @@ async def run_async(
     )
 
 
+@experimental(FeatureName.SKILL_TOOLSET)
+class SearchSkillsTool(BaseTool):
+  """Tool to search for relevant skills in the registry."""
+
+  def __init__(self, toolset: "SkillToolset"):
+    super().__init__(
+        name="search_skills",
+        description=toolset._registry.get_search_description(),
+    )
+    self._toolset = toolset
+
+  def _get_declaration(self) -> types.FunctionDeclaration | None:
+    properties = {
+        "query": {
+            "type": "string",
+            "description": "Semantic or keyword search query.",
+        },
+    }
+    filter_schema = self._toolset._registry.get_filter_schema()
+    if filter_schema:
+      properties["filters"] = filter_schema
+    return types.FunctionDeclaration(
+        name=self.name,
+        description=self.description,
+        parameters_json_schema={
+            "type": "object",
+            "properties": properties,
+            "required": ["query"],
+        },
+    )
+
+  async def run_async(
+      self, *, args: dict[str, Any], tool_context: ToolContext
+  ) -> Any:
+    query = args.get("query")
+    filters = args.get("filters")
+
+    if not query:
+      return {
+          "error": "Argument 'query' is required.",
+          "error_code": "INVALID_ARGUMENTS",
+      }
+
+    results = await self._toolset._registry.search_skills(
+        query=query, filters=filters
+    )
+
+    formatted_results = []
+    for r in results:
+      if r.name in self._toolset._skills:
+        logger.warning(
+            "Naming conflict detected: Skill '%s' exists both locally and in"
+            " the registry. Filtering out the registry skill.",
+            r.name,
+        )
+        continue
+      formatted_results.append(r.model_dump())
+    return formatted_results
+
+
 @experimental(FeatureName.SKILL_TOOLSET)
 class SkillToolset(BaseToolset):
   """A toolset for managing and interacting with agent skills."""
 
   def __init__(
       self,
-      skills: list[models.Skill],
+      skills: list[models.Skill] | None = None,
       *,
+      registry: Optional[SkillRegistry] = None,
       code_executor: Optional[BaseCodeExecutor] = None,
       script_timeout: int = _DEFAULT_SCRIPT_TIMEOUT,
       additional_tools: list[ToolUnion] | None = None,
@@ -787,6 +863,7 @@ def __init__(
 
     Args:
       skills: List of skills to register.
+      registry: Optional skill registry for dynamic discovery.
       code_executor: Optional code executor for script execution.
       script_timeout: Timeout in seconds for shell script execution via
         subprocess.run. Defaults to 300 seconds. Does not apply to Python
@@ -796,12 +873,13 @@ def __init__(
 
     # Check for duplicate skill names
     seen: set[str] = set()
-    for skill in skills:
+    for skill in skills or []:
       if skill.name in seen:
         raise ValueError(f"Duplicate skill name '{skill.name}'.")
       seen.add(skill.name)
 
-    self._skills = {skill.name: skill for skill in skills}
+    self._skills = {skill.name: skill for skill in skills or []}
+    self._registry = registry
     self._code_executor = code_executor
     self._script_timeout = script_timeout
     self._use_invocation_cache = False
@@ -824,6 +902,8 @@ def __init__(
         LoadSkillResourceTool(self),
         RunSkillScriptTool(self),
     ]
+    if self._registry:
+      self._tools.append(SearchSkillsTool(self))
 
   async def get_tools(
       self, readonly_context: ReadonlyContext | None = None
@@ -904,6 +984,12 @@ async def process_llm_request(
     skills_xml = prompt.format_skills_as_xml(skills)
     instructions = []
     instructions.append(_DEFAULT_SKILL_SYSTEM_INSTRUCTION)
+    if self._registry:
+      instructions.append(
+          "\nYou can also use the `search_skills` tool to discover additional"
+          " skills in the registry if the available skills listed below are"
+          " not sufficient.\n"
+      )
     instructions.append(skills_xml)
     llm_request.append_instructions(instructions)
 

tests/unittests/tools/test_skill_toolset.py

@@ -22,6 +22,7 @@
 from google.adk.code_executors.unsafe_local_code_executor import UnsafeLocalCodeExecutor
 from google.adk.models import llm_request as llm_request_model
 from google.adk.skills import models
+from google.adk.skills.skill_registry import SkillRegistry
 from google.adk.tools import skill_toolset
 from google.adk.tools import tool_context
 from google.genai import types
@@ -473,6 +474,128 @@ async def test_scripts_resource_not_found(mock_skill1, tool_context_instance):
   assert result["error_code"] == "RESOURCE_NOT_FOUND"
 
 
+class MockSkillRegistry(SkillRegistry):
+
+  def __init__(self):
+    self.skills = {}
+    self.search_results = []
+
+  async def get_skill(self, *, name, version=None):
+    return self.skills.get(name)
+
+  async def search_skills(self, *, query, filters=None, **kwargs):
+    return self.search_results
+
+  def get_filter_schema(self):
+    return None
+
+
+@pytest.mark.asyncio
+async def test_skill_toolset_init_with_registry(mock_skill1):
+  registry = MockSkillRegistry()
+  toolset = skill_toolset.SkillToolset([mock_skill1], registry=registry)
+  assert toolset._registry == registry
+  tools = await toolset.get_tools()
+  assert len(tools) == 5  # 4 default + SearchSkillsTool
+  assert isinstance(tools[4], skill_toolset.SearchSkillsTool)
+
+
+@pytest.mark.asyncio
+async def test_search_skills_tool_run_async(mock_skill1, tool_context_instance):
+  registry = MockSkillRegistry()
+  frontmatter = mock.create_autospec(models.Frontmatter, instance=True)
+  frontmatter.name = "remote-skill"
+  frontmatter.model_dump.return_value = {"name": "remote-skill"}
+  registry.search_results = [frontmatter]
+
+  toolset = skill_toolset.SkillToolset([mock_skill1], registry=registry)
+  tool = skill_toolset.SearchSkillsTool(toolset)
+
+  result = await tool.run_async(
+      args={"query": "test"}, tool_context=tool_context_instance
+  )
+  assert result == [{"name": "remote-skill"}]
+
+
+@pytest.mark.asyncio
+async def test_search_skills_tool_collision(
+    mock_skill1, tool_context_instance, caplog
+):
+  registry = MockSkillRegistry()
+  frontmatter = mock.create_autospec(models.Frontmatter, instance=True)
+  frontmatter.name = "skill1"  # Same name as mock_skill1
+  frontmatter.model_dump.return_value = {"name": "skill1"}
+
+  frontmatter2 = mock.create_autospec(models.Frontmatter, instance=True)
+  frontmatter2.name = "remote-skill"
+  frontmatter2.model_dump.return_value = {"name": "remote-skill"}
+
+  registry.search_results = [frontmatter, frontmatter2]
+
+  toolset = skill_toolset.SkillToolset([mock_skill1], registry=registry)
+  tool = skill_toolset.SearchSkillsTool(toolset)
+
+  with caplog.at_level(logging.WARNING):
+    result = await tool.run_async(
+        args={"query": "test"}, tool_context=tool_context_instance
+    )
+  assert result == [{"name": "remote-skill"}]
+  assert "Naming conflict detected" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_load_skill_tool_fetches_from_registry(
+    tool_context_instance, mock_skill1
+):
+  registry = MockSkillRegistry()
+  registry.skills["my-skill"] = mock_skill1
+
+  toolset = skill_toolset.SkillToolset([], registry=registry)
+  tool = skill_toolset.LoadSkillTool(toolset)
+
+  result = await tool.run_async(
+      args={"skill_name": "my-skill"}, tool_context=tool_context_instance
+  )
+  assert result["skill_name"] == "my-skill"
+  assert toolset._skills["my-skill"] == mock_skill1
+
+
+@pytest.mark.asyncio
+async def test_load_skill_tool_registry_error(tool_context_instance):
+  registry = MockSkillRegistry()
+  registry.get_skill = mock.AsyncMock(
+      side_effect=Exception("Test registry error")
+  )
+
+  toolset = skill_toolset.SkillToolset([], registry=registry)
+  tool = skill_toolset.LoadSkillTool(toolset)
+
+  result = await tool.run_async(
+      args={"skill_name": "my-skill"}, tool_context=tool_context_instance
+  )
+  assert result["error_code"] == "REGISTRY_ERROR"
+  assert "Failed to fetch skill 'my-skill' from registry" in result["error"]
+
+
+@pytest.mark.asyncio
+async def test_process_llm_request_with_registry(
+    mock_skill1, tool_context_instance
+):
+  registry = MockSkillRegistry()
+  toolset = skill_toolset.SkillToolset([mock_skill1], registry=registry)
+  llm_req = mock.create_autospec(llm_request_model.LlmRequest, instance=True)
+
+  await toolset.process_llm_request(
+      tool_context=tool_context_instance, llm_request=llm_req
+  )
+
+  llm_req.append_instructions.assert_called_once()
+  args, _ = llm_req.append_instructions.call_args
+  instructions = args[0]
+  assert len(instructions) == 3  # default + search instruction + skills xml
+  assert "search_skills" in instructions[1]
+
+
 # RunSkillScriptTool tests