Approval User Input

1"""
2Approval User Input
3=============================
4
5Approval + user input HITL: @approval + @tool(requires_user_input=True).
6"""
7
8import os
9import time
10
11from kern.agent import Agent
12from kern.approval import approval
13from kern.db.sqlite import SqliteDb
14from kern.models.openai import OpenAIResponses
15from kern.tools import tool
16
17DB_FILE = "tmp/approvals_test.db"
18
19
20@approval
21@tool(requires_user_input=True, user_input_fields=["recipient"])
22def send_money(amount: float, recipient: str, note: str) -> str:
23    """Send money to a recipient.
24
25    Args:
26        amount (float): The amount of money to send.
27        recipient (str): The recipient to send money to (provided by user).
28        note (str): A note to include with the transfer.
29
30    Returns:
31        str: Confirmation of the transfer.
32    """
33    return f"Sent ${amount} to {recipient}: {note}"
34
35
36# ---------------------------------------------------------------------------
37# Create Agent
38# ---------------------------------------------------------------------------
39db = SqliteDb(
40    db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
41)
42agent = Agent(
43    model=OpenAIResponses(id="gpt-5-mini"),
44    tools=[send_money],
45    markdown=True,
46    db=db,
47)
48
49# ---------------------------------------------------------------------------
50# Run Agent
51# ---------------------------------------------------------------------------
52if __name__ == "__main__":
53    # Clean up from previous runs
54    if os.path.exists(DB_FILE):
55        os.remove(DB_FILE)
56    os.makedirs("tmp", exist_ok=True)
57
58    # Re-create after cleanup
59    db = SqliteDb(
60        db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
61    )
62    agent = Agent(
63        model=OpenAIResponses(id="gpt-5-mini"),
64        tools=[send_money],
65        markdown=True,
66        db=db,
67    )
68
69    # Step 1: Run - agent will pause
70    print("--- Step 1: Running agent (expects pause) ---")
71    run_response = agent.run("Send $50 to someone with the note 'lunch money'.")
72    print(f"Run status: {run_response.status}")
73    assert run_response.is_paused, f"Expected paused, got {run_response.status}"
74    print("Agent paused as expected.")
75
76    # Step 2: Check that an approval record was created in the DB
77    print("\n--- Step 2: Checking approval record in DB ---")
78    approvals_list, total = db.get_approvals(status="pending", approval_type="required")
79    print(f"Pending approvals: {total}")
80    assert total >= 1, f"Expected at least 1 pending approval, got {total}"
81    approval_record = approvals_list[0]
82    print(f"  Approval ID: {approval_record['id']}")
83    print(f"  Run ID:      {approval_record['run_id']}")
84    print(f"  Status:      {approval_record['status']}")
85    print(f"  Source:      {approval_record['source_type']}")
86    print(f"  Context:     {approval_record.get('context')}")
87
88    # Step 3: Provide user input for recipient and confirm
89    print("\n--- Step 3: Providing user input and confirming ---")
90    for requirement in run_response.active_requirements:
91        if requirement.needs_user_input:
92            print(
93                f"  Providing user input for tool: {requirement.tool_execution.tool_name}"
94            )
95            requirement.provide_user_input({"recipient": "Alice"})
96        if requirement.needs_confirmation:
97            print(f"  Confirming tool: {requirement.tool_execution.tool_name}")
98            requirement.confirm()
99
100    run_response = agent.continue_run(
101        run_id=run_response.run_id,
102        requirements=run_response.requirements,
103    )
104    print(f"Run status after continue: {run_response.status}")
105    assert not run_response.is_paused, "Expected run to complete, but it's still paused"
106
107    # Step 4: Resolve the approval record in the DB
108    print("\n--- Step 4: Resolving approval in DB ---")
109    resolved = db.update_approval(
110        approval_record["id"],
111        expected_status="pending",
112        status="approved",
113        resolved_by="test_user",
114        resolved_at=int(time.time()),
115    )
116    assert resolved is not None, "Approval resolution failed (possible race condition)"
117    print(f"  Resolved status: {resolved['status']}")
118    print(f"  Resolved by:     {resolved['resolved_by']}")
119
120    # Step 5: Verify no more pending approvals
121    print("\n--- Step 5: Verifying no pending approvals ---")
122    count = db.get_pending_approval_count()
123    print(f"Remaining pending approvals: {count}")
124    assert count == 0, f"Expected 0 pending approvals, got {count}"
125
126    print("\n--- All checks passed! ---")
127    print(f"\nAgent output (truncated): {str(run_response.content)[:200]}...")

1# Clone and setup repo
2git clone https://github.com/kern-ai/kern.git
3cd kern/cookbook/02_agents/11_approvals
4
5# Create and activate virtual environment
6./scripts/demo_setup.sh
7source .venvs/demo/bin/activate
8
9python approval_user_input.py