Function run_interactively

src/google/adk/cli/cli.py:187–265

(
    root_agent_or_app: Union[LlmAgent, App],
    artifact_service: BaseArtifactService,
    session: Session,
    session_service: BaseSessionService,
    credential_service: BaseCredentialService,
    memory_service: Optional[BaseMemoryService] = None,
    timeout: Optional[str] = None,
    jsonl: bool = False,
)

async def run_interactively(
    root_agent_or_app: Union[LlmAgent, App],
    artifact_service: BaseArtifactService,
    session: Session,
    session_service: BaseSessionService,
    credential_service: BaseCredentialService,
    memory_service: Optional[BaseMemoryService] = None,
    timeout: Optional[str] = None,
    jsonl: bool = False,
) -> None:
  app = _to_app(root_agent_or_app, session.app_name)
  runner = Runner(
      app=app,
      artifact_service=artifact_service,
      session_service=session_service,
      memory_service=memory_service,
      credential_service=credential_service,
  )

  next_message = None
  resume_invocation_id = None
  while True:
    if next_message is None:
      query = input('[user]: ')
      if not query or not query.strip():
        continue
      if query == 'exit':
        break
      next_message = types.Content(role='user', parts=[types.Part(text=query)])

    collected_events = []
    invocation_id = None

    async def run_and_print():
      nonlocal invocation_id
      async with Aclosing(
          runner.run_async(
              user_id=session.user_id,
              session_id=session.id,
              new_message=next_message,
              invocation_id=resume_invocation_id,
          )
      ) as agen:
        async for event in agen:
          collected_events.append(event)
          if getattr(event, 'invocation_id', None):
            invocation_id = event.invocation_id
          _print_event(event, jsonl=jsonl, session_id=session.id)

    try:
      if timeout:
        seconds = _parse_timeout(timeout)
        await asyncio.wait_for(run_and_print(), timeout=seconds)
      else:
        await run_and_print()
    except asyncio.TimeoutError:
      click.secho(
          f'Error: Command timed out after {timeout}', fg='red', err=True