fix: surface sub-agent RPC errors from AgentTool

When a sub-agent invoked through AgentTool fails mid-run (for example a
RemoteA2aAgent whose RPC errors out), the failing event carries an
error_message but no content. AgentTool.run_async only inspected event
content, so it discarded the error and returned an empty string, hiding
the failure from the calling agent and the model.

Track the most recent error_message while draining the sub-agent's event
stream and return it when there is no usable content, so callers see the
actual failure instead of a silent empty result.

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 952181505

Author: GWeale

src/google/adk/tools/agent_tool.py

@@ -286,6 +286,7 @@ async def run_async(
     )
 
     last_content = None
+    last_error_message = None
     last_grounding_metadata = None
     async with Aclosing(
         runner.run_async(
@@ -296,6 +297,8 @@ async def run_async(
         # Forward state delta to parent session.
         if event.actions.state_delta:
           tool_context.state.update(event.actions.state_delta)
+        if event.error_message:
+          last_error_message = event.error_message
         if event.content:
           last_content = event.content
           last_grounding_metadata = event.grounding_metadata
@@ -305,9 +308,11 @@ async def run_async(
     await runner.close()
 
     if last_content is None or last_content.parts is None:
-      return ''
+      return last_error_message or ''
     parts_text = (_part_to_text(p) for p in last_content.parts if not p.thought)
     merged_text = '\n'.join(t for t in parts_text if t)
+    if not merged_text and last_error_message:
+      return last_error_message
     output_schema = _get_output_schema(self.agent)
     if output_schema:
       tool_result = validate_schema(output_schema, merged_text)

tests/unittests/tools/test_agent_tool.py

@@ -1148,6 +1148,60 @@ async def test_run_async_skips_thought_parts():
   assert result == '42'
 
 
+async def _run_agent_tool_with_events(events: list[Event]) -> Any:
+  """Drives AgentTool with an inner agent that yields `events` in order."""
+
+  class _StaticAgent(BaseAgent):
+
+    async def _run_async_impl(self, ctx):
+      for event in events:
+        yield event
+
+  inner = _StaticAgent(name='inner_agent', description='static')
+  agent_tool = AgentTool(agent=inner)
+
+  session_service = InMemorySessionService()
+  session = await session_service.create_session(
+      app_name='test_app', user_id='test_user'
+  )
+  invocation_context = InvocationContext(
+      invocation_id='invocation_id',
+      agent=inner,
+      session=session,
+      session_service=session_service,
+  )
+  tool_context = ToolContext(invocation_context=invocation_context)
+
+  return await agent_tool.run_async(
+      args={'request': 'test request'}, tool_context=tool_context
+  )
+
+
+@mark.asyncio
+async def test_run_async_preserves_error_event_with_no_content():
+  """An error-only event with no content surfaces its error_message."""
+  result = await _run_agent_tool_with_events([
+      Event(author='inner_agent', error_message='A2A request failed: 503'),
+  ])
+  assert result == 'A2A request failed: 503'
+
+
+@mark.asyncio
+async def test_run_async_preserves_error_when_only_thought_parts():
+  """When the final content is all thought parts, the error_message wins."""
+  result = await _run_agent_tool_with_events([
+      Event(
+          author='inner_agent',
+          content=types.Content(
+              role='model',
+              parts=[types.Part(text='thinking', thought=True)],
+          ),
+      ),
+      Event(author='inner_agent', error_message='A2A request failed: 503'),
+  ])
+  assert result == 'A2A request failed: 503'
+
+
 class TestAgentToolWithCompositeAgents:
   """Tests for AgentTool wrapping composite agents (SequentialAgent, etc.)."""