fix(a2a): warn when output_key cannot reach remote session

src/google/adk/agents/sequential_agent.py

@@ -18,10 +18,12 @@
 
 import inspect
 import logging
+import sys
 from typing import AsyncGenerator
 from typing import ClassVar
 from typing import Type
 
+from pydantic import model_validator
 from typing_extensions import deprecated
 from typing_extensions import override
 
@@ -43,6 +45,13 @@
 logger = logging.getLogger('google_adk.' + __name__)
 
 
+def _is_remote_a2a_agent(agent: BaseAgent) -> bool:
+  """Checks the agent type without importing the optional A2A dependency."""
+  remote_module = sys.modules.get('google.adk.agents.remote_a2a_agent')
+  remote_type = getattr(remote_module, 'RemoteA2aAgent', None)
+  return remote_type is not None and isinstance(agent, remote_type)
+
+
 def _tool_name(tool: ToolUnion) -> str | None:
   if isinstance(tool, BaseTool):
     return tool.name
@@ -95,6 +104,28 @@ class SequentialAgent(BaseAgent):
   version, along with the AgentConfig YAML loader.
   """
 
+  @model_validator(mode='after')
+  def _warn_on_output_key_handoff_to_remote(self) -> SequentialAgent:
+    """Warns when caller session state is expected to cross A2A."""
+    for current_agent, next_agent in zip(self.sub_agents, self.sub_agents[1:]):
+      if (
+          isinstance(current_agent, LlmAgent)
+          and current_agent.output_key
+          and _is_remote_a2a_agent(next_agent)
+      ):
+        logger.warning(
+            "SequentialAgent '%s' runs LlmAgent '%s' with output_key '%s'"
+            " immediately before RemoteA2aAgent '%s'. The output is saved in"
+            " the caller's session and is not available in the remote"
+            ' session; include values needed by the remote agent in event'
+            ' content instead.',
+            self.name,
+            current_agent.name,
+            current_agent.output_key,
+            next_agent.name,
+        )
+    return self
+
   @override
   async def _run_async_impl(
       self, ctx: InvocationContext

tests/unittests/agents/test_sequential_agent.py

@@ -18,6 +18,8 @@
 
 from google.adk.agents.base_agent import BaseAgent
 from google.adk.agents.invocation_context import InvocationContext
+from google.adk.agents.llm_agent import LlmAgent
+from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
 from google.adk.agents.sequential_agent import SequentialAgent
 from google.adk.agents.sequential_agent import SequentialAgentState
 from google.adk.apps import ResumabilityConfig
@@ -206,3 +208,61 @@ async def test_run_live(request: pytest.FixtureRequest):
 def test_deprecation_mentions_sub_agent_limitation():
   with pytest.warns(DeprecationWarning, match='sub-agent'):
     SequentialAgent(name='deprecated_sequential', sub_agents=[])
+
+
+def test_output_key_handoff_to_remote_warns(caplog: pytest.LogCaptureFixture):
+  """An output_key hand-off to a remote agent warns about session scope."""
+  writer = LlmAgent(name='writer', output_key='result')
+  remote = RemoteA2aAgent(
+      name='remote', agent_card='https://example.com/agent-card.json'
+  )
+
+  with caplog.at_level('WARNING'):
+    SequentialAgent(name='pipeline', sub_agents=[writer, remote])
+
+  assert "output_key 'result'" in caplog.text
+  assert "RemoteA2aAgent 'remote'" in caplog.text
+  assert 'not available in the remote session' in caplog.text
+
+
+def test_output_key_after_remote_does_not_warn(
+    caplog: pytest.LogCaptureFixture,
+):
+  """An output_key retained after a remote step does not cross A2A."""
+  remote = RemoteA2aAgent(
+      name='remote', agent_card='https://example.com/agent-card.json'
+  )
+  writer = LlmAgent(name='writer', output_key='result')
+
+  with caplog.at_level('WARNING'):
+    SequentialAgent(name='pipeline', sub_agents=[remote, writer])
+
+  assert 'not available in the remote session' not in caplog.text
+
+
+def test_remote_handoff_without_output_key_does_not_warn(
+    caplog: pytest.LogCaptureFixture,
+):
+  """A content-only hand-off to a remote agent does not warn."""
+  writer = LlmAgent(name='writer')
+  remote = RemoteA2aAgent(
+      name='remote', agent_card='https://example.com/agent-card.json'
+  )
+
+  with caplog.at_level('WARNING'):
+    SequentialAgent(name='pipeline', sub_agents=[writer, remote])
+
+  assert 'not available in the remote session' not in caplog.text
+
+
+def test_local_output_key_handoff_does_not_warn(
+    caplog: pytest.LogCaptureFixture,
+):
+  """An output_key hand-off between local agents remains in one session."""
+  writer = LlmAgent(name='writer', output_key='result')
+  reader = LlmAgent(name='reader')
+
+  with caplog.at_level('WARNING'):
+    SequentialAgent(name='pipeline', sub_agents=[writer, reader])
+
+  assert 'not available in the remote session' not in caplog.text