feat(agents): bridge node_input to user_content for ManagedAgent nodes

Override `ManagedAgent._run_impl` so that, when the agent runs as a node (e.g.
as a single-turn tool), the parent's tool-call argument arriving as `node_input`
is surfaced as the agent's `user_content` and forwarded to the interactions API.
When `node_input` is `None` (classic agent-tree run), behavior matches
`BaseAgent._run_impl`. Uses the shared `node_input_to_content` helper.

Co-authored-by: Haran Rajkumar <haranrk@google.com>
PiperOrigin-RevId: 945949834

Author: haranrk

src/google/adk/agents/_managed_agent.py

@@ -31,6 +31,7 @@
 from pydantic import ConfigDict
 from pydantic import Field
 from pydantic import PrivateAttr
+from typing_extensions import override
 
 from ..events.event import Event
 from ..flows.llm_flows.interactions_processor import _find_previous_interaction_state
@@ -47,9 +48,11 @@
 from ..tools.tool_context import ToolContext
 from ..utils._google_client_headers import get_tracking_http_options
 from ..utils._google_client_headers import merge_tracking_headers
+from ..utils.content_utils import to_user_content
 from ..utils.context_utils import Aclosing
 from ..utils.env_utils import is_enterprise_mode_enabled
 from .base_agent import BaseAgent
+from .context import Context
 from .invocation_context import InvocationContext
 from .readonly_context import ReadonlyContext
 from .run_config import StreamingMode
@@ -311,6 +314,30 @@ def _error_event(
         turn_complete=True,
     )
 
+  @override
+  async def _run_impl(
+      self, *, ctx: Context, node_input: Any
+  ) -> AsyncGenerator[Event, None]:
+    """Runs the ManagedAgent as a node, threading node_input into user_content.
+
+    When invoked as a single-turn tool (``mode='single_turn'``), the parent's
+    tool-call argument arrives as ``node_input``; surface it as the agent's
+    ``user_content`` so ``_run_async_impl`` sends it to the interactions API.
+    When ``node_input`` is ``None`` (classic agent-tree run), behavior is
+    identical to ``BaseAgent._run_impl``.
+    """
+    parent_context = ctx.get_invocation_context()
+    if node_input is not None:
+      parent_context = parent_context.model_copy(
+          update={'user_content': to_user_content(node_input)}
+      )
+    async for event in self.run_async(parent_context=parent_context):
+      if event.author:
+        ctx.event_author = event.author
+      if not event.node_info.path and event.author == self.name:
+        event.node_info.path = ctx.node_path
+      yield event
+
   async def _run_async_impl(
       self, ctx: InvocationContext
   ) -> AsyncGenerator[Event, None]:

tests/unittests/agents/test_managed_agent.py

@@ -775,12 +775,51 @@ def test_mode_single_turn_is_accepted():
 
 def test_mode_chat_is_rejected():
   from pydantic import ValidationError
-  import pytest
 
   with pytest.raises(ValidationError):
     ManagedAgent(name='m', agent_id='a', mode='chat')
 
 
+async def test_run_impl_bridges_node_input_to_user_content():
+  from google.adk.agents.context import Context
+  from google.adk.agents.invocation_context import InvocationContext
+  from google.adk.sessions.in_memory_session_service import InMemorySessionService
+  from google.adk.sessions.session import Session
+  from google.adk.utils.content_utils import to_user_content
+
+  captured = {}
+
+  # ManagedAgent is a Pydantic model, so `run_async` cannot be reassigned on an
+  # instance; subclassing to override it is the pydantic-safe equivalent of
+  # stubbing it. The override captures the `parent_context.user_content` that
+  # `_run_impl` threads in from `node_input`.
+  class _CapturingManagedAgent(ManagedAgent):
+
+    async def run_async(self, parent_context):
+      captured['user_content'] = parent_context.user_content
+      return
+      yield  # make this an async generator
+
+  agent = _CapturingManagedAgent(name='m', agent_id='a', mode='single_turn')
+
+  # Build a minimal node Context (mirrors
+  # tests/unittests/workflow/test_agent_node.py).
+  session = Session(app_name='test', user_id='user', id='session')
+  ic = InvocationContext(
+      invocation_id='inv',
+      session=session,
+      session_service=InMemorySessionService(),
+  )
+  ctx = Context(ic, node_path='wf')
+
+  events = [
+      e async for e in agent._run_impl(ctx=ctx, node_input='compute primes')
+  ]
+
+  assert events == []
+  assert captured['user_content'] == to_user_content('compute primes')
+
+
 async def _drain(agen):
   async for _ in agen:
     pass