Approval Basic

1"""
2Approval Basic
3=============================
4
5Approval-backed HITL: @approval + @tool(requires_confirmation=True) with persistent DB record.
6"""
7
8import json
9import os
10import time
11
12import httpx
13from kern.agent import Agent
14from kern.approval import approval
15from kern.db.sqlite import SqliteDb
16from kern.models.openai import OpenAIResponses
17from kern.tools import tool
18
19DB_FILE = "tmp/approvals_test.db"
20
21
22@approval
23@tool(requires_confirmation=True)
24def get_top_hackernews_stories(num_stories: int) -> str:
25    """Fetch top stories from Hacker News.
26
27    Args:
28        num_stories (int): Number of stories to retrieve.
29
30    Returns:
31        str: JSON string of story details.
32    """
33    response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
34    story_ids = response.json()
35    stories = []
36    for story_id in story_ids[:num_stories]:
37        story = httpx.get(
38            f"https://hacker-news.firebaseio.com/v0/item/{story_id}.json"
39        ).json()
40        story.pop("text", None)
41        stories.append(story)
42    return json.dumps(stories)
43
44
45# ---------------------------------------------------------------------------
46# Create Agent
47# ---------------------------------------------------------------------------
48db = SqliteDb(
49    db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
50)
51agent = Agent(
52    model=OpenAIResponses(id="gpt-5-mini"),
53    tools=[get_top_hackernews_stories],
54    markdown=True,
55    db=db,
56)
57
58# ---------------------------------------------------------------------------
59# Run Agent
60# ---------------------------------------------------------------------------
61if __name__ == "__main__":
62    # Clean up from previous runs
63    if os.path.exists(DB_FILE):
64        os.remove(DB_FILE)
65    os.makedirs("tmp", exist_ok=True)
66
67    # Re-create after cleanup
68    db = SqliteDb(
69        db_file=DB_FILE, session_table="agent_sessions", approvals_table="approvals"
70    )
71    agent = Agent(
72        model=OpenAIResponses(id="gpt-5-mini"),
73        tools=[get_top_hackernews_stories],
74        markdown=True,
75        db=db,
76    )
77
78    # Step 1: Run - agent will pause because the tool requires approval
79    print("--- Step 1: Running agent (expects pause) ---")
80    run_response = agent.run("Fetch the top 2 hackernews stories.")
81    print(f"Run status: {run_response.status}")
82    assert run_response.is_paused, f"Expected paused, got {run_response.status}"
83    print("Agent paused as expected.")
84
85    # Step 2: Check that an approval record was created in the DB
86    print("\n--- Step 2: Checking approval record in DB ---")
87    approvals_list, total = db.get_approvals(status="pending")
88    print(f"Pending approvals: {total}")
89    assert total >= 1, f"Expected at least 1 pending approval, got {total}"
90    approval_record = approvals_list[0]
91    print(f"  Approval ID: {approval_record['id']}")
92    print(f"  Run ID:      {approval_record['run_id']}")
93    print(f"  Status:      {approval_record['status']}")
94    print(f"  Source:      {approval_record['source_type']}")
95    print(f"  Context:     {approval_record.get('context')}")
96
97    # Step 3: Confirm the requirement and continue the run
98    print("\n--- Step 3: Confirming and continuing ---")
99    for requirement in run_response.active_requirements:
100        if requirement.needs_confirmation:
101            print(f"  Confirming tool: {requirement.tool_execution.tool_name}")
102            requirement.confirm()
103
104    run_response = agent.continue_run(
105        run_id=run_response.run_id,
106        requirements=run_response.requirements,
107    )
108    print(f"Run status after continue: {run_response.status}")
109    assert not run_response.is_paused, "Expected run to complete, but it's still paused"
110
111    # Step 4: Resolve the approval record in the DB
112    print("\n--- Step 4: Resolving approval in DB ---")
113    resolved = db.update_approval(
114        approval_record["id"],
115        expected_status="pending",
116        status="approved",
117        resolved_by="test_user",
118        resolved_at=int(time.time()),
119    )
120    assert resolved is not None, "Approval resolution failed (possible race condition)"
121    print(f"  Resolved status: {resolved['status']}")
122    print(f"  Resolved by:     {resolved['resolved_by']}")
123
124    # Step 5: Verify no more pending approvals
125    print("\n--- Step 5: Verifying no pending approvals ---")
126    count = db.get_pending_approval_count()
127    print(f"Remaining pending approvals: {count}")
128    assert count == 0, f"Expected 0 pending approvals, got {count}"
129
130    print("\n--- All checks passed! ---")
131    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_basic.py