Method run_live

src/google/adk/agents/base_agent.py:334–360

Entry method to run an agent via video/audio-based conversation. Args: parent_context: InvocationContext, the invocation context of the parent agent. Yields: Event: the events generated by the agent.

(
      self,
      parent_context: InvocationContext,
  )

  @final
  async def run_live(
      self,
      parent_context: InvocationContext,
  ) -> AsyncGenerator[Event, None]:
    """Entry method to run an agent via video/audio-based conversation.

    Args:
      parent_context: InvocationContext, the invocation context of the parent
        agent.

    Yields:
      Event: the events generated by the agent.
    """

    ctx = self._create_invocation_context(parent_context)
    async with _instrumentation.record_agent_invocation(ctx, self):
      if event := await self._handle_before_agent_callback(ctx):
        yield event
      if ctx.end_invocation:
        return

      async with Aclosing(self._run_live_impl(ctx)) as agen:
        async for event in agen:
          yield event

      if event := await self._handle_after_agent_callback(ctx):
        yield event

  async def _run_async_impl(
      self, ctx: InvocationContext