feat(tools): let toolsets replace a failed tools listing

Apps fronting per-user OAuth MCP servers need a seam to contribute
placeholder tools on HTTP 401 instead of the framework silently dropping
the toolset. Add BaseToolset.on_tools_listing_error and invoke it from
the agent and SkillToolset isolate-and-continue paths.

Author: a2105z

src/google/adk/agents/llm_agent.py

@@ -210,11 +210,11 @@ async def _convert_tool_union_to_tools(
   try:
     return await tool_union.get_tools_with_prefix(ctx)
   except Exception as e:
-    # The agent still runs, just without this toolset's tools, and the model
-    # will answer as though it never had them. That is a lost capability
-    # rather than a degraded one, so report it at error level, name which
-    # toolset was lost, and keep the traceback: str(e) is empty for several
-    # of the exceptions raised by transport clients.
+    # The agent still runs after a listing failure. Report at error level,
+    # name which toolset was lost, and keep the traceback: str(e) is empty
+    # for several of the exceptions raised by transport clients. Then let the
+    # toolset decide what to contribute (default: nothing) so apps can return
+    # placeholder tools for expected failures such as per-user OAuth 401s.
     logger.error(
         'Agent %s will run without the tools from toolset %s%s, which failed'
         ' to load: %s',
@@ -228,7 +228,7 @@ async def _convert_tool_union_to_tools(
         e,
         exc_info=True,
     )
-    return []
+    return await tool_union.on_tools_listing_error(e, ctx)
 
 
 # TODO: drop the explicit abc.ABC base once BaseNode surfaces ABCMeta to

src/google/adk/tools/base_toolset.py

@@ -99,6 +99,35 @@ async def get_tools(
       list[BaseTool]: A list of tools available under the specified context.
     """
 
+  async def on_tools_listing_error(
+      self,
+      error: Exception,
+      readonly_context: Optional[ReadonlyContext] = None,
+  ) -> list[BaseTool]:
+    """Decide what this toolset contributes when ``get_tools`` fails.
+
+    Framework call sites that load toolsets for an agent (or for skill
+    additional tools) catch listing failures so one broken toolset cannot take
+    down the rest. By default those call sites then contribute nothing from the
+    failed toolset. Override this hook to contribute a replacement tool list
+    instead — for example placeholder tools that prompt the user to complete
+    per-user OAuth for an MCP server that returned HTTP 401.
+
+    Direct callers of ``get_tools`` / ``get_tools_with_prefix`` still see the
+    original exception; this hook is only invoked by the framework's
+    isolate-and-continue paths.
+
+    Args:
+      error: The exception raised by ``get_tools`` / ``get_tools_with_prefix``.
+      readonly_context: The same context passed to the failed listing call.
+
+    Returns:
+      Tools to use in place of the failed listing. The default returns an
+      empty list (contribute nothing), matching historical framework behavior.
+    """
+    del error, readonly_context
+    return []
+
   @final
   async def get_tools_with_prefix(
       self,

src/google/adk/tools/skill_toolset.py

@@ -1314,7 +1314,11 @@ async def _resolve_additional_tools_from_state(
               ts_tools,
               exc_info=ts_tools,
           )
-          continue
+          # Let the toolset replace a failed listing (default: contribute
+          # nothing) so apps can surface expected auth failures as tools.
+          ts_tools = await toolset.on_tools_listing_error(
+              ts_tools, readonly_context
+          )
         if isinstance(ts_tools, BaseException):
           raise ts_tools
         for t in ts_tools:

tests/unittests/agents/test_llm_agent_fields.py

@@ -643,6 +643,38 @@ async def get_tools(self, readonly_context=None):
     # The traceback is what identifies where inside the toolset it broke.
     assert record.exc_info is not None
 
+  async def test_canonical_tools_uses_on_tools_listing_error_hook(self, caplog):
+    """A toolset can contribute placeholder tools after a listing failure."""
+    from google.adk.tools.base_tool import BaseTool
+    from google.adk.tools.base_toolset import BaseToolset
+
+    class AuthPromptToolset(BaseToolset):
+
+      async def get_tools(self, readonly_context=None):
+        raise ConnectionError('HTTP 401 Unauthorized')
+
+      async def on_tools_listing_error(self, error, readonly_context=None):
+        del readonly_context
+        tool = mock.MagicMock(spec=BaseTool)
+        tool.name = 'connect_mcp_server'
+        tool.description = f'Authorize access ({error})'
+        tool._get_declaration = mock.MagicMock(return_value=None)
+        return [tool]
+
+    agent = LlmAgent(
+        name='test_agent',
+        model='gemini-pro',
+        tools=[AuthPromptToolset()],
+    )
+    ctx = await _create_readonly_context(agent)
+
+    with caplog.at_level(logging.ERROR, logger='google_adk'):
+      tools = await agent.canonical_tools(ctx)
+
+    assert len(tools) == 1
+    assert tools[0].name == 'connect_mcp_server'
+    assert any('failed to load' in r.getMessage() for r in caplog.records)
+
 
 # Tests for multi-provider model support via string model names
 @pytest.mark.parametrize(

tests/unittests/tools/test_base_toolset.py

@@ -448,3 +448,40 @@ async def test_get_tools_with_prefix_caching():
       readonly_context=readonly_context2
   )
   assert tools4 is not tools5
+
+
+@pytest.mark.asyncio
+async def test_on_tools_listing_error_default_contributes_nothing():
+  """Default hook returns an empty list so failed listings contribute nothing."""
+  toolset = _TestingToolset()
+  tools = await toolset.on_tools_listing_error(
+      ConnectionError('MCP unauthorized'), readonly_context=None
+  )
+  assert tools == []
+
+
+@pytest.mark.asyncio
+async def test_on_tools_listing_error_can_be_overridden():
+  """Subclasses can return placeholder tools when listing fails."""
+
+  class _AuthPromptToolset(_TestingToolset):
+
+    async def on_tools_listing_error(
+        self,
+        error: Exception,
+        readonly_context: Optional[ReadonlyContext] = None,
+    ) -> list[BaseTool]:
+      del readonly_context
+      return [
+          _TestingTool(
+              name='connect_mcp',
+              description=f'Authorize MCP access ({error})',
+          )
+      ]
+
+  toolset = _AuthPromptToolset()
+  tools = await toolset.on_tools_listing_error(
+      ConnectionError('HTTP 401'), readonly_context=None
+  )
+  assert len(tools) == 1
+  assert tools[0].name == 'connect_mcp'

tests/unittests/tools/test_skill_toolset.py

@@ -2403,6 +2403,8 @@ async def test_skill_toolset_resolution_isolates_failing_toolset(
   failing_toolset.get_tools_with_prefix.side_effect = RuntimeError(
       "MCP server unreachable"
   )
+  # Default hook contributes nothing — same as historical skip behavior.
+  failing_toolset.on_tools_listing_error = mock.AsyncMock(return_value=[])
 
   toolset = skill_toolset.SkillToolset(
       [mock_skill1],
@@ -2428,6 +2430,46 @@ async def test_skill_toolset_resolution_isolates_failing_toolset(
   assert "Skipping toolset" in caplog.text
 
 
+@pytest.mark.asyncio
+async def test_skill_toolset_resolution_uses_on_tools_listing_error_hook(
+    mock_skill1, caplog
+):
+  """A failing toolset can contribute placeholder tools via the hook."""
+  mock_skill1.frontmatter.metadata = {
+      "adk_additional_tools": ["connect_mcp_server"]
+  }
+  mock_skill1.name = "skill1"
+
+  placeholder = mock.create_autospec(skill_toolset.BaseTool, instance=True)
+  placeholder.name = "connect_mcp_server"
+
+  failing_toolset = mock.create_autospec(
+      skill_toolset.BaseToolset, instance=True
+  )
+  failing_toolset.get_tools_with_prefix.side_effect = ConnectionError(
+      "HTTP 401 Unauthorized"
+  )
+  failing_toolset.on_tools_listing_error = mock.AsyncMock(
+      return_value=[placeholder]
+  )
+
+  toolset = skill_toolset.SkillToolset(
+      [mock_skill1],
+      additional_tools=[failing_toolset],
+  )
+  ctx = _make_tool_context_with_agent()
+
+  load_tool = skill_toolset.LoadSkillTool(toolset)
+  await load_tool.run_async(args={"skill_name": "skill1"}, tool_context=ctx)
+
+  with caplog.at_level(logging.WARNING):
+    tools = await toolset.get_tools(readonly_context=ctx)
+
+  assert "connect_mcp_server" in {t.name for t in tools}
+  failing_toolset.on_tools_listing_error.assert_awaited_once()
+  assert "Skipping toolset" in caplog.text
+
+
 @pytest.mark.asyncio
 async def test_skill_toolset_resolution_propagates_system_exceptions(
     mock_skill1,