Audit Approval External

1"""
2Audit Approval External
3=============================
4
5Audit approval with external execution: @approval(type="audit") + @tool(external_execution=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(external_execution=True)
21def run_security_scan(target: str) -> str:
22    """Run a security scan against a target system.
23
24    Args:
25        target (str): The target to scan.
26
27    Returns:
28        str: Scan results.
29    """
30    return f"Scan complete for {target}: no vulnerabilities found"
31
32
33# ---------------------------------------------------------------------------
34# Create Agent
35# ---------------------------------------------------------------------------
36db = SqliteDb(
37    db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
38)
39agent = Agent(
40    model=OpenAIResponses(id="gpt-5-mini"),
41    tools=[run_security_scan],
42    markdown=True,
43    db=db,
44)
45
46# ---------------------------------------------------------------------------
47# Run Agent
48# ---------------------------------------------------------------------------
49if __name__ == "__main__":
50    # Clean up from previous runs
51    if os.path.exists(DB_FILE):
52        os.remove(DB_FILE)
53    os.makedirs("tmp", exist_ok=True)
54
55    # Re-create after cleanup
56    db = SqliteDb(
57        db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
58    )
59    agent = Agent(
60        model=OpenAIResponses(id="gpt-5-mini"),
61        tools=[run_security_scan],
62        markdown=True,
63        db=db,
64    )
65
66    # Step 1: Run - agent will pause because the tool requires external execution
67    print("--- Step 1: Running agent (expects pause for external execution) ---")
68    run_response = agent.run("Run a security scan on the production server.")
69    print(f"Run status: {run_response.status}")
70    assert run_response.is_paused, f"Expected paused, got {run_response.status}"
71    print("Agent paused as expected.")
72
73    # Step 2: Verify no approval record yet (logged approvals are created after resolution)
74    print("\n--- Step 2: Verifying no approval records yet ---")
75    approvals_list, total = db.get_approvals()
76    print(f"Total approvals before resolution: {total}")
77    assert total == 0, f"Expected 0 approvals before resolution, got {total}"
78    print("No approval records yet (as expected for audit approval).")
79
80    # Step 3: Provide external execution result and continue
81    print("\n--- Step 3: Setting external result and continuing ---")
82    for requirement in run_response.active_requirements:
83        if requirement.needs_external_execution:
84            print(f"  Setting result for tool: {requirement.tool_execution.tool_name}")
85            requirement.set_external_execution_result(
86                "Scan complete: no vulnerabilities found"
87            )
88
89    run_response = agent.continue_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    print(f"  Source:         {approval_record['source_type']}")
106    assert approval_record["status"] == "approved", (
107        f"Expected status 'approved', got {approval_record['status']}"
108    )
109    assert approval_record["approval_type"] == "audit", (
110        f"Expected type 'audit', got {approval_record['approval_type']}"
111    )
112
113    print("\n--- All checks passed! ---")
114    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_external.py