Function _prompt_for_function_call

src/google/adk/cli/cli.py:139–184

Prompts the user for a HITL function call and returns the response.

(
    fc_id: str, fc_name: str, args: dict[str, Any]
)

def _prompt_for_function_call(
    fc_id: str, fc_name: str, args: dict[str, Any]
) -> types.Content:
  """Prompts the user for a HITL function call and returns the response."""
  if fc_name == _REQUEST_INPUT:
    message = args.get('message') or 'Input requested'
    schema = args.get('response_schema')
    click.echo(f'[HITL input] {message}')
    if schema:
      click.echo(f'  Schema: {json.dumps(schema)}')
  elif fc_name == _REQUEST_CONFIRMATION:
    tool_confirmation = args.get('toolConfirmation', {})
    hint = tool_confirmation.get('hint', '')
    original_fc = args.get('originalFunctionCall', {})
    original_name = original_fc.get('name', 'unknown')
    click.echo(f'[HITL confirm] {hint or f"Confirm {original_name}?"}')
    click.echo('  Type "yes" to confirm, anything else to reject.')
  else:
    click.echo(f'[HITL] Waiting for input for {fc_name}({args})')

  user_input = input('[user]: ')

  # Build the FunctionResponse.
  if fc_name == _REQUEST_CONFIRMATION:
    confirmed = _is_positive_response(user_input)
    response = {'confirmed': confirmed}
  else:
    # Try to parse as JSON, fall back to wrapping as {"result": value}.
    try:
      parsed = json.loads(user_input)
      response = parsed if isinstance(parsed, dict) else {'result': parsed}
    except (json.JSONDecodeError, ValueError):
      response = {'result': user_input}

  return types.Content(
      role='user',
      parts=[
          types.Part(
              function_response=types.FunctionResponse(
                  id=fc_id,
                  name=fc_name,
                  response=response,
              )
          )
      ],
  )


async def run_interactively(