Human Approval Steps

import time

# Human-in-the-loop workflow node
def request_approval_node(state: dict) -> str:
    """
    Pause the workflow and request human approval.
    In production, this sends a notification and polls for a response.
    """
    action_summary = state["data"].get("proposed_action", "Unknown action")
    amount = state["data"].get("amount", 0)

    # Send approval request (in production: email, Slack, dashboard notification)
    approval_request = {
        "workflow_id": state.get("workflow_id", "WF-001"),
        "action": action_summary,
        "amount": amount,
        "proposed_by": "AI Agent",
        "requires_response_by": "2024-01-15T18:00:00Z",
    }

    print(f"  [APPROVAL REQUEST SENT]")
    print(f"  Action: {approval_request['action']}")
    print(f"  Amount: $" + str(approval_request['amount']))
    state["data"]["approval_status"] = "pending"
    state["status"] = "waiting"

    # Poll for approval (in production: this is async, not a blocking loop)
    # Here we simulate an immediate approval for demonstration
    simulated_response = {"approved": True, "reviewer": "alice@example.com"}
    time.sleep(0.1)  # Simulated wait

    if simulated_response["approved"]:
        state["data"]["approved_by"] = simulated_response["reviewer"]
        state["data"]["approval_status"] = "approved"
        state["status"] = "running"
        print(f"  [APPROVED] by {simulated_response['reviewer']}")
        return "approved"
    else:
        state["data"]["approval_status"] = "rejected"
        state["status"] = "running"
        print(f"  [REJECTED] by {simulated_response['reviewer']}")
        return "rejected"

state = {
    "workflow_id": "WF-001",
    "status": "running",
    "data": {"proposed_action": "Issue refund", "amount": 149.99}
}
result = request_approval_node(state)
print(f"Approval outcome: {result}")