Audit Approval User Input

1"""
2Audit Approval User Input
3=============================
4
5Audit approval with user input: @approval(type="audit") + @tool(requires_user_input=True).
6"""
7
8import os
9
10from kern.agent import Agent
11from kern.approval import approval
12from kern.db.sqlite import SqliteDb
13from kern.models.openai import OpenAIResponses
14from kern.tools import tool
15
16DB_FILE = "tmp/approvals_test.db"
17
18
19@approval(type="audit")
20@tool(requires_user_input=True, user_input_fields=["account"])
21def transfer_funds(amount: float, account: str) -> str:
22    """Transfer funds to an account.
23
24    Args:
25        amount (float): The amount to transfer.
26        account (str): The destination account (provided by user).
27
28    Returns:
29        str: Confirmation of the transfer.
30    """
31    return f"Transferred ${amount} to {account}"
32
33
34# ---------------------------------------------------------------------------
35# Create Agent
36# ---------------------------------------------------------------------------
37db = SqliteDb(
38    db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
39)
40agent = Agent(
41    model=OpenAIResponses(id="gpt-5-mini"),
42    tools=[transfer_funds],
43    markdown=True,
44    db=db,
45)
46
47# ---------------------------------------------------------------------------
48# Run Agent
49# ---------------------------------------------------------------------------
50if __name__ == "__main__":
51    # Clean up from previous runs
52    if os.path.exists(DB_FILE):
53        os.remove(DB_FILE)
54    os.makedirs("tmp", exist_ok=True)
55
56    # Re-create after cleanup
57    db = SqliteDb(
58        db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
59    )
60    agent = Agent(
61        model=OpenAIResponses(id="gpt-5-mini"),
62        tools=[transfer_funds],
63        markdown=True,
64        db=db,
65    )
66
67    # Step 1: Run - agent will pause because the tool requires user input
68    print("--- Step 1: Running agent (expects pause) ---")
69    run_response = agent.run("Transfer $250 to my savings account.")
70    print(f"Run status: {run_response.status}")
71    assert run_response.is_paused, f"Expected paused, got {run_response.status}"
72    print("Agent paused as expected.")
73
74    # Step 2: Verify no logged approvals exist yet (audit approval creates records AFTER resolution)
75    print("\n--- Step 2: Verifying no logged approvals yet ---")
76    approvals_list, total = db.get_approvals(approval_type="audit")
77    print(f"Logged approvals before resolution: {total}")
78    assert total == 0, f"Expected 0 logged approvals before resolution, got {total}"
79    print("No logged approvals yet (as expected).")
80
81    # Step 3: Provide user input and continue
82    print("\n--- Step 3: Providing user input and continuing ---")
83    for requirement in run_response.active_requirements:
84        if requirement.needs_user_input:
85            print(
86                f"  Providing user input for tool: {requirement.tool_execution.tool_name}"
87            )
88            requirement.provide_user_input({"account": "SAVINGS-9876"})
89
90    run_response = agent.continue_run(
91        run_id=run_response.run_id,
92        requirements=run_response.requirements,
93    )
94    print(f"Run status after continue: {run_response.status}")
95    assert not run_response.is_paused, "Expected run to complete, but it's still paused"
96
97    # Step 4: Verify logged approval record was created after resolution
98    print("\n--- Step 4: Verifying logged approval record ---")
99    approvals_list, total = db.get_approvals(approval_type="audit")
100    print(f"Logged approvals after resolution: {total}")
101    assert total >= 1, f"Expected at least 1 logged approval, got {total}"
102    approval_record = approvals_list[0]
103    print(f"  Approval ID: {approval_record['id']}")
104    print(f"  Status:      {approval_record['status']}")
105    print(f"  Type:        {approval_record['approval_type']}")
106    assert approval_record["status"] == "approved", (
107        f"Expected 'approved', got {approval_record['status']}"
108    )
109    assert approval_record["approval_type"] == "audit", (
110        f"Expected 'audit', got {approval_record['approval_type']}"
111    )
112    print("Logged approval record verified.")
113
114    # Step 5: Verify total state
115    print("\n--- Step 5: Verifying final state ---")
116    pending_count = db.get_pending_approval_count()
117    print(f"Pending approvals: {pending_count}")
118    assert pending_count == 0, f"Expected 0 pending approvals, got {pending_count}"
119    all_approvals, all_total = db.get_approvals(approval_type="audit")
120    print(f"Total logged approvals: {all_total}")
121    assert all_total == 1, f"Expected 1 logged approval, got {all_total}"
122
123    print("\n--- All checks passed! ---")
124    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 audit_approval_user_input.py