Audit Approval Async

1"""
2Audit Approval Async
3=============================
4
5Async audit approval: @approval(type="audit") + @tool(requires_confirmation=True) with async.
6"""
7
8import asyncio
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 message.
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# ---------------------------------------------------------------------------
49# Run Agent
50# ---------------------------------------------------------------------------
51async def main():
52    # Clean up from previous runs
53    if os.path.exists(DB_FILE):
54        os.remove(DB_FILE)
55    os.makedirs("tmp", exist_ok=True)
56
57    # Re-create after cleanup
58    _db = SqliteDb(
59        db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
60    )
61    _agent = Agent(
62        model=OpenAIResponses(id="gpt-5-mini"),
63        tools=[delete_user_data],
64        markdown=True,
65        db=_db,
66    )
67
68    # Step 1: Async run - agent will pause because the tool requires confirmation
69    print("--- Step 1: Running agent async (expects pause) ---")
70    run_response = await _agent.arun("Delete all data for user U-99887.")
71    print(f"Run status: {run_response.status}")
72    assert run_response.is_paused, f"Expected paused, got {run_response.status}"
73    print("Agent paused as expected.")
74
75    # Step 2: Verify no approval record yet (logged approvals are created after resolution)
76    print("\n--- Step 2: Verifying no approval records yet ---")
77    approvals_list, total = _db.get_approvals()
78    print(f"Total approvals before resolution: {total}")
79    assert total == 0, f"Expected 0 approvals before resolution, got {total}"
80    print("No approval records yet (as expected for audit approval).")
81
82    # Step 3: Confirm and continue async
83    print("\n--- Step 3: Confirming and continuing async ---")
84    for requirement in run_response.active_requirements:
85        if requirement.needs_confirmation:
86            print(f"  Confirming tool: {requirement.tool_execution.tool_name}")
87            requirement.confirm()
88
89    run_response = await _agent.acontinue_run(
90        run_id=run_response.run_id,
91        requirements=run_response.requirements,
92    )
93    print(f"Run status after continue: {run_response.status}")
94    assert not run_response.is_paused, "Expected run to complete, but it's still paused"
95
96    # Step 4: Verify logged approval record was created in DB
97    print("\n--- Step 4: Verifying logged approval record in DB ---")
98    approvals_list, total = _db.get_approvals(approval_type="audit")
99    print(f"Logged approvals: {total}")
100    assert total >= 1, f"Expected at least 1 logged approval, got {total}"
101    approval_record = approvals_list[0]
102    print(f"  Approval ID:    {approval_record['id']}")
103    print(f"  Status:         {approval_record['status']}")
104    print(f"  Approval type:  {approval_record['approval_type']}")
105    assert approval_record["status"] == "approved", (
106        f"Expected status 'approved', got {approval_record['status']}"
107    )
108    assert approval_record["approval_type"] == "audit", (
109        f"Expected type 'audit', got {approval_record['approval_type']}"
110    )
111
112    print("\n--- All checks passed! ---")
113    print(f"\nAgent output (truncated): {str(run_response.content)[:200]}...")
114
115
116if __name__ == "__main__":
117    asyncio.run(main())

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