check if transfer target is a sibling agent

Author: AriYusa

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

@@ -723,6 +723,15 @@ def _get_agent_to_run(
     agent_to_run = root_agent.find_agent(agent_name)
     if not agent_to_run:
       raise ValueError(f'Agent {agent_name} not found in the agent tree.')
+
+    from ...agents.llm_agent import LlmAgent
+
+    if all([
+        isinstance(invocation_context.agent, LlmAgent),
+        invocation_context.agent.disallow_transfer_to_peers,
+        agent_to_run.parent_agent == invocation_context.agent.parent_agent,
+    ]):
+      raise ValueError(f'Transfer to sibling agent {agent_name} is disallowed.')
     return agent_to_run
 
   async def _call_llm_async(

tests/unittests/flows/llm_flows/test_base_llm_flow_transfer.py

@@ -0,0 +1,62 @@
+"""Tests for BaseLlmFlow._get_agent_to_run transfer-to-peers behavior."""
+
+from __future__ import annotations
+
+from google.adk.agents.llm_agent import LlmAgent
+from google.adk.flows.llm_flows.base_llm_flow import BaseLlmFlow
+import pytest
+
+from ... import testing_utils
+
+
+def make_agent_tree():
+  root = LlmAgent(name='root')
+  child1 = LlmAgent(name='child1')
+  child2 = LlmAgent(name='child2')
+
+  child1.parent_agent = root
+  child2.parent_agent = root
+  root.sub_agents = [child1, child2]
+  return root, child1, child2
+
+
+@pytest.mark.asyncio
+async def test_transfer_to_sibling_disallowed_raises():
+  root, child1, child2 = make_agent_tree()
+
+  caller = child1
+  caller.disallow_transfer_to_peers = True
+
+  ctx = await testing_utils.create_invocation_context(caller)
+
+  flow = BaseLlmFlow()
+
+  with pytest.raises(ValueError) as exc:
+    flow._get_agent_to_run(ctx, 'child2')
+
+
+@pytest.mark.asyncio
+async def test_transfer_to_sibling_allowed_returns_agent():
+  root, child1, child2 = make_agent_tree()
+
+  caller = child1
+  caller.disallow_transfer_to_peers = False
+
+  ctx = await testing_utils.create_invocation_context(caller)
+
+  flow = BaseLlmFlow()
+  agent = flow._get_agent_to_run(ctx, 'child2')
+  assert agent is not None
+  assert agent.name == 'child2'
+
+
+@pytest.mark.asyncio
+async def test_transfer_to_unknown_agent_raises():
+  root, child1, child2 = make_agent_tree()
+
+  caller = child1
+  ctx = await testing_utils.create_invocation_context(caller)
+  flow = BaseLlmFlow()
+
+  with pytest.raises(ValueError) as exc:
+    flow._get_agent_to_run(ctx, 'not_in_tree')