Approval External Execution

1"""
2Approval External Execution
3=============================
4
5Approval + external execution HITL: @approval + @tool(external_execution=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(external_execution=True)
22def deploy_to_production(service_name: str, version: str) -> str:
23    """Deploy a service to production.
24
25    Args:
26        service_name (str): The name of the service to deploy.
27        version (str): The version to deploy.
28
29    Returns:
30        str: Confirmation of the deployment.
31    """
32    return f"Deployed {service_name} v{version}"
33
34
35# ---------------------------------------------------------------------------
36# Create Agent
37# ---------------------------------------------------------------------------
38db = SqliteDb(
39    db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
40)
41agent = Agent(
42    model=OpenAIResponses(id="gpt-5-mini"),
43    tools=[deploy_to_production],
44    markdown=True,
45    db=db,
46)
47
48# ---------------------------------------------------------------------------
49# Run Agent
50# ---------------------------------------------------------------------------
51if __name__ == "__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=[deploy_to_production],
64        markdown=True,
65        db=db,
66    )
67
68    # Step 1: Run - agent will pause
69    print("--- Step 1: Running agent (expects pause) ---")
70    run_response = agent.run("Deploy the auth-service version 2.1.0 to production.")
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: Check that an approval record was created in the DB
76    print("\n--- Step 2: Checking approval record in DB ---")
77    approvals_list, total = db.get_approvals(status="pending", approval_type="required")
78    print(f"Pending approvals: {total}")
79    assert total >= 1, f"Expected at least 1 pending approval, got {total}"
80    approval_record = approvals_list[0]
81    print(f"  Approval ID: {approval_record['id']}")
82    print(f"  Run ID:      {approval_record['run_id']}")
83    print(f"  Status:      {approval_record['status']}")
84    print(f"  Source:      {approval_record['source_type']}")
85    print(f"  Context:     {approval_record.get('context')}")
86
87    # Step 3: Set external execution result and continue
88    print("\n--- Step 3: Setting external result and continuing ---")
89    for requirement in run_response.active_requirements:
90        if requirement.needs_external_execution:
91            print(f"  Setting result for tool: {requirement.tool_execution.tool_name}")
92            requirement.set_external_execution_result("Deployed auth-service v2.1.0")
93
94    run_response = agent.continue_run(
95        run_id=run_response.run_id,
96        requirements=run_response.requirements,
97    )
98    print(f"Run status after continue: {run_response.status}")
99    assert not run_response.is_paused, "Expected run to complete, but it's still paused"
100
101    # Step 4: Resolve the approval record in the DB
102    print("\n--- Step 4: Resolving approval in DB ---")
103    resolved = db.update_approval(
104        approval_record["id"],
105        expected_status="pending",
106        status="approved",
107        resolved_by="test_user",
108        resolved_at=int(time.time()),
109    )
110    assert resolved is not None, "Approval resolution failed (possible race condition)"
111    print(f"  Resolved status: {resolved['status']}")
112    print(f"  Resolved by:     {resolved['resolved_by']}")
113
114    # Step 5: Verify no more pending approvals
115    print("\n--- Step 5: Verifying no pending approvals ---")
116    count = db.get_pending_approval_count()
117    print(f"Remaining pending approvals: {count}")
118    assert count == 0, f"Expected 0 pending approvals, got {count}"
119
120    print("\n--- All checks passed! ---")
121    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_external_execution.py