Function main

sdk/python/examples/hitl_confirmation_loop.py:69–111

()

def main() -> None:
    config_path = os.environ.get("A3S_CONFIG_FILE") or (sys.argv[1] if len(sys.argv) > 1 else None)
    if not config_path:
        raise RuntimeError("Set A3S_CONFIG_FILE or pass an agent.acl path as the first argument.")

    workspace = os.environ.get("A3S_CODE_WORKSPACE") or os.getcwd()
    prompt = os.environ.get("A3S_CODE_HITL_PROMPT") or (
        "Use the bash tool to run: echo A3S_CODE_HITL_OK. Then summarize the result."
    )

    opts = SessionOptions()
    opts.planning_mode = "disabled"
    opts.permission_policy = PermissionPolicy(ask=["bash*"], default_decision="allow")
    opts.confirmation_policy = ConfirmationPolicy(
        enabled=True,
        default_timeout_ms=120_000,
        timeout_action="reject",
        yolo_lanes=["query"],
    )

    agent = Agent.create(config_path)
    session = agent.session(workspace, opts)

    try:
        for event in session.stream(prompt):
            if event.event_type == "text_delta" and event.text:
                print(event.text, end="", flush=True)
            elif event.event_type == "tool_start":
                print(f"\n[tool:start] {event.tool_name or 'unknown'}")
            elif event.event_type == "tool_end":
                print(f"\n[tool:end] {event.tool_name or 'unknown'} exit={event.exit_code or 0}")
            elif event.event_type == "confirmation_required":
                confirm_from_cli(session, event)
            elif event.event_type == "confirmation_received":
                print(f"\n[hitl] confirmation_received {event.tool_id or ''}")
            elif event.event_type == "confirmation_timeout":
                print(f"\n[hitl] confirmation_timeout {event.tool_id or ''}")
            elif event.event_type == "error":
                raise RuntimeError(event.error or "stream error")

        print("\n[hitl] stream complete")
    finally:
        session.cancel_confirmations()


if __name__ == "__main__":