Audit Approval Confirmation

1"""
2Audit Approval Confirmation
3=============================
4
5Audit approval with confirmation: @approval(type="audit") + @tool(requires_confirmation=True).
6Demonstrates both approval and rejection paths.
7"""
8
9import os
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(type="audit")
21@tool(requires_confirmation=True)
22def delete_user_data(user_id: str) -> str:
23    """Permanently delete all data for a user.
24
25    Args:
26        user_id (str): The user ID whose data should be deleted.
27
28    Returns:
29        str: Confirmation of the deletion.
30    """
31    return f"Deleted data for user {user_id}"
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=[delete_user_data],
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=[delete_user_data],
63        markdown=True,
64        db=db,
65    )
66
67    # ========== Part 1: Approval path ==========
68
69    # Step 1: Run - agent will pause because the tool requires confirmation
70    print("--- Step 1: Running agent for approval path (expects pause) ---")
71    run_response = agent.run("Delete all data for user U-100.")
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: Confirm and continue
77    print("\n--- Step 2: Confirming and continuing ---")
78    for requirement in run_response.active_requirements:
79        if requirement.needs_confirmation:
80            print(f"  Confirming tool: {requirement.tool_execution.tool_name}")
81            requirement.confirm()
82
83    run_response = agent.continue_run(
84        run_id=run_response.run_id,
85        requirements=run_response.requirements,
86    )
87    print(f"Run status after continue: {run_response.status}")
88    assert not run_response.is_paused, "Expected run to complete, but it's still paused"
89
90    # Step 3: Verify logged approval record was created
91    print("\n--- Step 3: Verifying logged approval record (approved) ---")
92    approvals_list, total = db.get_approvals(approval_type="audit")
93    print(f"Logged approvals: {total}")
94    assert total >= 1, f"Expected at least 1 logged approval, got {total}"
95    approval_record = approvals_list[0]
96    print(f"  Approval ID: {approval_record['id']}")
97    print(f"  Status:      {approval_record['status']}")
98    print(f"  Type:        {approval_record['approval_type']}")
99    assert approval_record["status"] == "approved", (
100        f"Expected 'approved', got {approval_record['status']}"
101    )
102    assert approval_record["approval_type"] == "audit", (
103        f"Expected 'audit', got {approval_record['approval_type']}"
104    )
105    print("Logged approval record verified (approved).")
106
107    # ========== Part 2: Rejection path ==========
108
109    # Step 4: Run again - agent will pause for a new confirmation
110    print("\n--- Step 4: Running agent for rejection path (expects pause) ---")
111    run_response = agent.run("Delete all data for user U-200.")
112    print(f"Run status: {run_response.status}")
113    assert run_response.is_paused, f"Expected paused, got {run_response.status}"
114    print("Agent paused as expected.")
115
116    # Step 5: Reject and continue
117    print("\n--- Step 5: Rejecting and continuing ---")
118    for requirement in run_response.active_requirements:
119        if requirement.needs_confirmation:
120            print(f"  Rejecting tool: {requirement.tool_execution.tool_name}")
121            requirement.reject("Rejected by admin: not authorized")
122
123    run_response = agent.continue_run(
124        run_id=run_response.run_id,
125        requirements=run_response.requirements,
126    )
127    print(f"Run status after continue: {run_response.status}")
128    assert not run_response.is_paused, "Expected run to complete, but it's still paused"
129
130    # Step 6: Verify logged approval record for rejection
131    print("\n--- Step 6: Verifying logged approval record (rejected) ---")
132    approvals_list, total = db.get_approvals(approval_type="audit")
133    print(f"Total logged approvals: {total}")
134    assert total >= 2, f"Expected at least 2 logged approvals, got {total}"
135    # Find the rejected one (most recent)
136    rejected = [a for a in approvals_list if a["status"] == "rejected"]
137    assert len(rejected) >= 1, (
138        f"Expected at least 1 rejected approval, got {len(rejected)}"
139    )
140    rej = rejected[0]
141    print(f"  Approval ID: {rej['id']}")
142    print(f"  Status:      {rej['status']}")
143    print(f"  Type:        {rej['approval_type']}")
144    assert rej["approval_type"] == "audit", (
145        f"Expected 'audit', got {rej['approval_type']}"
146    )
147    print("Logged approval record verified (rejected).")
148
149    # Final check: total logged approvals
150    print("\n--- Final: Checking total logged approvals ---")
151    all_logged, all_total = db.get_approvals(approval_type="audit")
152    print(f"Total logged approvals: {all_total}")
153    assert all_total == 2, f"Expected 2 total logged approvals, got {all_total}"
154
155    print("\n--- All checks passed! ---")

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_confirmation.py