fix: Notify agent of failed agent transfer in function response

src/google/adk/flows/llm_flows/base_llm_flow.py

@@ -439,15 +439,29 @@ async def _postprocess_live(
       function_response_event = await functions.handle_function_calls_live(
           invocation_context, model_response_event, llm_request.tools_dict
       )
-      yield function_response_event
 
       transfer_to_agent = function_response_event.actions.transfer_to_agent
       if transfer_to_agent:
         agent_to_run = self._get_agent_to_run(
             invocation_context, transfer_to_agent
         )
-        async for item in agent_to_run.run_live(invocation_context):
-          yield item
+        transfer_successful = False
+        try:
+          async for item in agent_to_run.run_live(invocation_context):
+            if not transfer_successful:
+              yield function_response_event
+              transfer_successful = True
+            yield item
+        finally:
+          if not transfer_successful:
+            function_response_event.content.parts[
+                0
+            ].function_response.response = {
+                'result': f'Error transferring to {transfer_to_agent}'
+            }
+            yield function_response_event
+      else:
+        yield function_response_event
 
   async def _postprocess_run_processors_async(
       self, invocation_context: InvocationContext, llm_response: LlmResponse
@@ -471,14 +485,28 @@ async def _postprocess_handle_function_calls_async(
       if auth_event:
         yield auth_event
 
-      yield function_response_event
       transfer_to_agent = function_response_event.actions.transfer_to_agent
       if transfer_to_agent:
         agent_to_run = self._get_agent_to_run(
             invocation_context, transfer_to_agent
         )
-        async for event in agent_to_run.run_async(invocation_context):
-          yield event
+        transfer_successful = False
+        try:
+          async for event in agent_to_run.run_async(invocation_context):
+            if not transfer_successful:
+              yield function_response_event
+              transfer_successful = True
+            yield event
+        finally:
+          if not transfer_successful:
+            function_response_event.content.parts[
+                0
+            ].function_response.response = {
+                'result': f'Error transferring to {transfer_to_agent}'
+            }
+            yield function_response_event
+      else:
+        yield function_response_event
 
   def _get_agent_to_run(
       self, invocation_context: InvocationContext, agent_name: str

tests/unittests/flows/llm_flows/test_agent_transfer.py

@@ -11,13 +11,16 @@
 # 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 typing import Optional
 
 from google.adk.agents.llm_agent import Agent
 from google.adk.agents.loop_agent import LoopAgent
+from google.adk.agents.readonly_context import ReadonlyContext
 from google.adk.agents.sequential_agent import SequentialAgent
-from google.adk.tools import exit_loop
+from google.adk.tools import BaseTool, FunctionTool, exit_loop
 from google.genai.types import Part
 
+from google.adk.tools.base_toolset import BaseToolset
 from ... import testing_utils
 
 
@@ -311,3 +314,61 @@ def test_auto_to_loop():
   assert testing_utils.simplify_events(runner.run('test2')) == [
       ('root_agent', 'response5'),
   ]
+
+
+class FakeError(Exception):
+  pass
+
+
+class BrokenToolset(BaseToolset):
+
+  async def get_tools(
+      self,
+      readonly_context: Optional[ReadonlyContext] = None,
+  ) -> list[BaseTool]:
+    raise FakeError()
+
+  async def close(self):
+    pass
+
+
+def test_transfer_error_handling():
+  response = [
+      transfer_call_part('sub_agent_1'),
+      'response1',
+      'response2',
+  ]
+  mockModel = testing_utils.MockModel.create(responses=response)
+  # root (auto) - sub_agent_1 (auto)
+
+  sub_agent_1 = Agent(
+      name='sub_agent_1',
+      model=mockModel,
+      tools=[
+          BrokenToolset(),
+      ],
+  )
+  root_agent = Agent(
+      name='root_agent',
+      model=mockModel,
+      sub_agents=[sub_agent_1],
+  )
+
+  runner = testing_utils.InMemoryRunner(root_agent)
+
+  # Asserts the transfer.
+  assert testing_utils.simplify_events(runner.run('test1')) == [
+      ('root_agent', transfer_call_part('sub_agent_1')),
+      (
+          'root_agent',
+          Part.from_function_response(
+              name='transfer_to_agent',
+              response={'result': 'Error transferring to sub_agent_1'},
+          ),
+      ),
+  ]
+
+  # root_agent should still be the current agent as the transfer failed.
+  assert testing_utils.simplify_events(runner.run('test2')) == [
+      ('root_agent', 'response1'),
+  ]