Method _run_node_async

src/google/adk/runners.py:437–585

Run a BaseNode through NodeRunner. Events flow through ic._event_queue via NodeRunner.

(
      self,
      *,
      user_id: str,
      session_id: str,
      invocation_id: Optional[str] = None,
      new_message: Optional[types.Content] = None,
      state_delta: Optional[dict[str, Any]] = None,
      run_config: Optional[RunConfig] = None,
      yield_user_message: bool = False,
      node: Optional['BaseNode'] = None,
  )

    )

  async def _run_node_async(
      self,
      *,
      user_id: str,
      session_id: str,
      invocation_id: Optional[str] = None,
      new_message: Optional[types.Content] = None,
      state_delta: Optional[dict[str, Any]] = None,
      run_config: Optional[RunConfig] = None,
      yield_user_message: bool = False,
      node: Optional['BaseNode'] = None,
  ) -> AsyncGenerator[Event, None]:
    """Run a BaseNode through NodeRunner.

    Events flow through ic._event_queue via NodeRunner.
    """
    from .workflow._node_runner import NodeRunner

    with tracer.start_as_current_span('invocation'):
      # 1. Setup
      session = await self._get_or_create_session(
          user_id=user_id, session_id=session_id
      )

      # Validate and resolve resume inputs
      resume_inputs = self._extract_resume_inputs(new_message)
      self._validate_new_message(new_message, resume_inputs)

      if not invocation_id and new_message:
        invocation_id = self._resolve_invocation_id_from_fr(
            session, new_message
        )

      ic = self._new_invocation_context(
          session,
          new_message=new_message,
          run_config=run_config or RunConfig(),
          invocation_id=invocation_id,
      )
      ic._event_queue = asyncio.Queue()

      # 2. Append user message to session and resolve node_input
      node_input = None
      if resume_inputs or invocation_id:
        # Resume: recover the original user content. new_message here is a
        # function response (or None), so it can't populate user_content.
        node_input = self._find_original_user_content(
            ic.session, ic.invocation_id
        )
        if node_input:
          ic.user_content = node_input
      if not node_input:
        # Fresh: use user message as node_input
        node_input = new_message

      # Run callbacks on user message
      if new_message:
        modified_user_message = (