Class JoinNode

src/google/adk/workflow/_join_node.py:47–77

A node that waits for all specified predecessors to trigger it before outputting.

class JoinNode(BaseNode):
  """A node that waits for all specified predecessors to trigger it before
  outputting."""

  @property
  @override
  def _requires_all_predecessors(self) -> bool:
    return True

  @override
  def _validate_input_data(self, data: Any) -> Any:
    """Validates individual trigger inputs against input_schema."""
    if self.input_schema and isinstance(data, dict):
      return {
          k: self._validate_schema(v, self.input_schema)
          for k, v in data.items()
      }
    return super()._validate_input_data(data)

  @override
  async def _run_impl(
      self,
      *,
      ctx: Context,
      node_input: Any,
  ) -> AsyncGenerator[Any, None]:
    """JoinNode simply passes through the aggregated inputs provided by the orchestrator."""
    yield Event(
        output=node_input,
        branch=ctx._invocation_context.branch,
    )