Class A2aAgentExecutor

src/google/adk/a2a/executor/a2a_agent_executor.py:58–378

An AgentExecutor that runs an ADK Agent against an A2A request and publishes updates to an event queue. Args: runner: The runner to use for the agent. config: The config to use for the executor. use_legacy: If true, force the legacy implementation. force_new_version: If true, f

@a2a_experimental
class A2aAgentExecutor(AgentExecutor):
  """An AgentExecutor that runs an ADK Agent against an A2A request and

  publishes updates to an event queue.

  Args:
    runner: The runner to use for the agent.
    config: The config to use for the executor.
    use_legacy: If true, force the legacy implementation.
    force_new_version: If true, force the new implementation regardless of the
      extension.
  """

  def __init__(
      self,
      *,
      runner: Runner | Callable[..., Runner | Awaitable[Runner]],
      config: Optional[A2aAgentExecutorConfig] = None,
      use_legacy: bool = False,
      force_new_version: bool = False,
  ):
    super().__init__()
    self._runner = runner
    self._config = config or A2aAgentExecutorConfig()
    self._use_legacy = use_legacy
    self._force_new_version = force_new_version
    self._executor_impl = None

  async def _resolve_runner(self) -> Runner:
    """Resolve the runner, handling cases where it's a callable that returns a Runner."""
    # If already resolved and cached, return it
    if isinstance(self._runner, Runner):
      return self._runner
    if callable(self._runner):
      # Call the function to get the runner
      result = self._runner()

      # Handle async callables
      if inspect.iscoroutine(result):
        resolved_runner = await result
      else:
        resolved_runner = result

      # Cache the resolved runner for future calls
      self._runner = resolved_runner
      return resolved_runner

    raise TypeError(
        'Runner must be a Runner instance or a callable that returns a'
        f' Runner, got {type(self._runner)}'
    )

  @override
  async def cancel(self, context: RequestContext, event_queue: EventQueue):
    """Cancel the execution."""
    if self._executor_impl:
      await self._executor_impl.cancel(context, event_queue)
      return