Approval Async

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