feat(plugins): add TRANSFER_A2A classification for TransferToAgentTool

When RemoteA2aAgent is used via sub_agents, the framework creates a
TransferToAgentTool that only stores agent names. The analytics plugin
returned TRANSFER_AGENT unconditionally for all transfer tools, losing
the A2A distinction. This preserves transfer-target origin metadata so
analytics can distinguish remote A2A transfers from local ones.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: caohy1988

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

@@ -48,8 +48,23 @@ async def run_async(
     if not transfer_targets:
       return
 
+    try:
+      from ...agents.remote_a2a_agent import RemoteA2aAgent
+    except ImportError:
+      RemoteA2aAgent = None
+
+    target_origin_by_name = {
+        agent.name: (
+            'TRANSFER_A2A'
+            if RemoteA2aAgent is not None and isinstance(agent, RemoteA2aAgent)
+            else 'TRANSFER_AGENT'
+        )
+        for agent in transfer_targets
+    }
+
     transfer_to_agent_tool = TransferToAgentTool(
-        agent_names=[agent.name for agent in transfer_targets]
+        agent_names=[agent.name for agent in transfer_targets],
+        target_origin_by_name=target_origin_by_name,
     )
 
     llm_request.append_instructions([

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -208,6 +208,30 @@ def _get_tool_origin(tool: "BaseTool") -> str:
   return "UNKNOWN"
 
 
+def _resolve_transfer_origin(
+    tool: "BaseTool",
+    tool_origin: str,
+    tool_args: Optional[dict[str, Any]],
+) -> str:
+  """Refine transfer-tool origin using the selected target name.
+
+  For ``TransferToAgentTool`` calls, resolves the per-call origin
+  (e.g. ``TRANSFER_A2A``) from the tool's ``_target_origin_by_name``
+  metadata and the ``agent_name`` in ``tool_args``.
+
+  Returns ``tool_origin`` unchanged for non-transfer tools or when
+  the selected target is not in the metadata map.
+  """
+  if (
+      tool_origin == "TRANSFER_AGENT"
+      and tool_args
+      and "agent_name" in tool_args
+  ):
+    origin_map = getattr(tool, "_target_origin_by_name", {})
+    return origin_map.get(tool_args["agent_name"], tool_origin)
+  return tool_origin
+
+
 def _recursive_smart_truncate(
     obj: Any, max_len: int, seen: Optional[set[int]] = None
 ) -> tuple[Any, bool]:
@@ -3175,7 +3199,9 @@ async def before_tool_callback(
     args_truncated, is_truncated = _recursive_smart_truncate(
         tool_args, self.config.max_content_length
     )
-    tool_origin = _get_tool_origin(tool)
+    tool_origin = _resolve_transfer_origin(
+        tool, _get_tool_origin(tool), tool_args
+    )
     content_dict = {
         "tool": tool.name,
         "args": args_truncated,
@@ -3209,7 +3235,9 @@ async def after_tool_callback(
     resp_truncated, is_truncated = _recursive_smart_truncate(
         result, self.config.max_content_length
     )
-    tool_origin = _get_tool_origin(tool)
+    tool_origin = _resolve_transfer_origin(
+        tool, _get_tool_origin(tool), tool_args
+    )
     content_dict = {
         "tool": tool.name,
         "result": resp_truncated,
@@ -3254,7 +3282,9 @@ async def on_tool_error_callback(
     args_truncated, is_truncated = _recursive_smart_truncate(
         tool_args, self.config.max_content_length
     )
-    tool_origin = _get_tool_origin(tool)
+    tool_origin = _resolve_transfer_origin(
+        tool, _get_tool_origin(tool), tool_args
+    )
     content_dict = {
         "tool": tool.name,
         "args": args_truncated,

src/google/adk/tools/transfer_to_agent_tool.py

@@ -49,19 +49,30 @@ class TransferToAgentTool(FunctionTool):
 
   Attributes:
     agent_names: List of valid agent names that can be transferred to.
+    target_origin_by_name: Optional mapping from agent name to its origin
+        category (e.g. ``"TRANSFER_AGENT"`` or ``"TRANSFER_A2A"``).
+        Used by analytics plugins to distinguish remote A2A transfers
+        from local ones at call time.
   """
 
   def __init__(
       self,
       agent_names: list[str],
+      *,
+      target_origin_by_name: Optional[dict[str, str]] = None,
   ):
     """Initialize the TransferToAgentTool.
 
     Args:
       agent_names: List of valid agent names that can be transferred to.
+      target_origin_by_name: Optional mapping from agent name to its
+          origin category. When provided, analytics plugins can resolve
+          per-call origin (e.g. ``"TRANSFER_A2A"``) from the selected
+          target name.
     """
     super().__init__(func=transfer_to_agent)
     self._agent_names = agent_names
+    self._target_origin_by_name = target_origin_by_name or {}
 
   @override
   def _get_declaration(self) -> Optional[types.FunctionDeclaration]:

tests/unittests/flows/llm_flows/test_agent_transfer_system_instructions.py

@@ -296,3 +296,60 @@ async def test_agent_transfer_no_instructions_when_no_transfer_targets():
   instructions = llm_request.config.system_instruction or ''
   assert '**NOTE**:' not in instructions
   assert 'transfer_to_agent' not in instructions
+
+
+@pytest.mark.asyncio
+async def test_transfer_tool_preserves_a2a_target_origin():
+  """Test that TransferToAgentTool metadata distinguishes A2A from local."""
+  try:
+    from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
+  except ImportError:
+    pytest.skip('RemoteA2aAgent not available')
+
+  mockModel = testing_utils.MockModel.create(responses=[])
+
+  local_agent = Agent(
+      name='local_agent',
+      model=mockModel,
+      description='Local agent',
+  )
+
+  # Minimal subclass that satisfies isinstance() without needing
+  # full A2A dependencies (agent_card, httpx, etc.).
+  class _StubRemoteA2aAgent(RemoteA2aAgent):
+
+    model_config = {'arbitrary_types_allowed': True}
+
+    def __init__(self, **kwargs):
+      # Bypass RemoteA2aAgent.__init__; call BaseAgent directly.
+      from google.adk.agents.base_agent import BaseAgent
+
+      BaseAgent.__init__(self, **kwargs)
+
+  remote_agent = _StubRemoteA2aAgent(
+      name='remote_agent',
+      description='Remote A2A agent',
+  )
+
+  main_agent = Agent(
+      name='main_agent',
+      model=mockModel,
+      sub_agents=[local_agent, remote_agent],
+      description='Main agent',
+  )
+
+  invocation_context = await create_test_invocation_context(main_agent)
+  llm_request = LlmRequest()
+
+  async for _ in agent_transfer.request_processor.run_async(
+      invocation_context, llm_request
+  ):
+    pass
+
+  # Retrieve the TransferToAgentTool that was added to the request
+  tool = llm_request.tools_dict.get('transfer_to_agent')
+  assert tool is not None, 'transfer_to_agent tool not found in llm_request'
+
+  origin_map = tool._target_origin_by_name
+  assert origin_map['local_agent'] == 'TRANSFER_AGENT'
+  assert origin_map['remote_agent'] == 'TRANSFER_A2A'

tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py

@@ -4371,6 +4371,72 @@ def test_unknown_tool_returns_unknown(self):
     result = bigquery_agent_analytics_plugin._get_tool_origin(tool)
     assert result == "UNKNOWN"
 
+  def test_transfer_tool_remote_target_returns_transfer_a2a(self):
+    from google.adk.tools.transfer_to_agent_tool import TransferToAgentTool
+
+    tool = TransferToAgentTool(
+        agent_names=["local", "remote"],
+        target_origin_by_name={
+            "local": "TRANSFER_AGENT",
+            "remote": "TRANSFER_A2A",
+        },
+    )
+    result = bigquery_agent_analytics_plugin._resolve_transfer_origin(
+        tool, "TRANSFER_AGENT", {"agent_name": "remote"}
+    )
+    assert result == "TRANSFER_A2A"
+
+  def test_transfer_tool_local_target_returns_transfer_agent(self):
+    from google.adk.tools.transfer_to_agent_tool import TransferToAgentTool
+
+    tool = TransferToAgentTool(
+        agent_names=["local", "remote"],
+        target_origin_by_name={
+            "local": "TRANSFER_AGENT",
+            "remote": "TRANSFER_A2A",
+        },
+    )
+    result = bigquery_agent_analytics_plugin._resolve_transfer_origin(
+        tool, "TRANSFER_AGENT", {"agent_name": "local"}
+    )
+    assert result == "TRANSFER_AGENT"
+
+  def test_transfer_tool_no_tool_args_returns_transfer_agent(self):
+    from google.adk.tools.transfer_to_agent_tool import TransferToAgentTool
+
+    tool = TransferToAgentTool(
+        agent_names=["remote"],
+        target_origin_by_name={"remote": "TRANSFER_A2A"},
+    )
+    result = bigquery_agent_analytics_plugin._resolve_transfer_origin(
+        tool, "TRANSFER_AGENT", None
+    )
+    assert result == "TRANSFER_AGENT"
+
+  def test_transfer_tool_missing_agent_name_returns_transfer_agent(self):
+    from google.adk.tools.transfer_to_agent_tool import TransferToAgentTool
+
+    tool = TransferToAgentTool(
+        agent_names=["remote"],
+        target_origin_by_name={"remote": "TRANSFER_A2A"},
+    )
+    result = bigquery_agent_analytics_plugin._resolve_transfer_origin(
+        tool, "TRANSFER_AGENT", {"other": "val"}
+    )
+    assert result == "TRANSFER_AGENT"
+
+  def test_transfer_tool_unknown_name_returns_transfer_agent(self):
+    from google.adk.tools.transfer_to_agent_tool import TransferToAgentTool
+
+    tool = TransferToAgentTool(
+        agent_names=["remote"],
+        target_origin_by_name={"remote": "TRANSFER_A2A"},
+    )
+    result = bigquery_agent_analytics_plugin._resolve_transfer_origin(
+        tool, "TRANSFER_AGENT", {"agent_name": "unknown"}
+    )
+    assert result == "TRANSFER_AGENT"
+
 
 class TestHITLTracing:
   """Tests for HITL-specific event emission via on_event_callback.

tests/unittests/tools/test_transfer_to_agent_tool.py

@@ -122,6 +122,25 @@ def test_transfer_to_agent_tool_maintains_inheritance():
   assert hasattr(tool, 'process_llm_request')
 
 
+def test_target_origin_by_name_default_empty():
+  """Test that _target_origin_by_name defaults to empty dict."""
+  tool = TransferToAgentTool(agent_names=['agent_a'])
+  assert tool._target_origin_by_name == {}
+
+
+def test_target_origin_by_name_stored():
+  """Test that _target_origin_by_name stores the provided mapping."""
+  origin_map = {
+      'local_agent': 'TRANSFER_AGENT',
+      'remote_agent': 'TRANSFER_A2A',
+  }
+  tool = TransferToAgentTool(
+      agent_names=['local_agent', 'remote_agent'],
+      target_origin_by_name=origin_map,
+  )
+  assert tool._target_origin_by_name == origin_map
+
+
 def test_transfer_to_agent_tool_handles_parameters_json_schema():
   """Test that TransferToAgentTool handles parameters_json_schema format."""
   agent_names = ['agent_x', 'agent_y', 'agent_z']