Streaming: Confirm Team Tool

1"""Team HITL Streaming: Tool on the team itself requiring confirmation.
2
3This example demonstrates HITL for tools provided directly to the Team
4(not to member agents) in streaming mode. When the team leader decides
5to use a tool that requires confirmation, the entire team run pauses
6until the human confirms.
7
8Note: For team-level tools (not member agent tools), you can use either
9isinstance(event, TeamRunPausedEvent) or event.is_paused since there's
10no member agent pause to confuse it with.
11"""
12
13from kern.agent import Agent
14from kern.db.sqlite import SqliteDb
15from kern.models.openai import OpenAIResponses
16from kern.run.team import RunPausedEvent as TeamRunPausedEvent
17from kern.team.team import Team
18from kern.tools import tool
19from kern.utils import pprint
20
21# ---------------------------------------------------------------------------
22# Setup
23# ---------------------------------------------------------------------------
24db = SqliteDb(db_file="tmp/team_hitl_stream.db")
25
26
27# ---------------------------------------------------------------------------
28# Tools
29# ---------------------------------------------------------------------------
30@tool(requires_confirmation=True)
31def approve_deployment(environment: str, service: str) -> str:
32    """Approve and execute a deployment to an environment.
33
34    Args:
35        environment (str): Target environment (staging, production)
36        service (str): Service to deploy
37    """
38    return f"Deployment of {service} to {environment} approved and executed"
39
40
41# ---------------------------------------------------------------------------
42# Create Members
43# ---------------------------------------------------------------------------
44research_agent = Agent(
45    name="Research Agent",
46    role="Researches deployment readiness",
47    model=OpenAIResponses(id="gpt-5.2-mini"),
48    db=db,
49)
50
51
52# ---------------------------------------------------------------------------
53# Create Team
54# ---------------------------------------------------------------------------
55team = Team(
56    name="Release Team",
57    members=[research_agent],
58    model=OpenAIResponses(id="gpt-5.2-mini"),
59    tools=[approve_deployment],
60    db=db,
61)
62
63
64# ---------------------------------------------------------------------------
65# Run Team
66# ---------------------------------------------------------------------------
67if __name__ == "__main__":
68    for run_event in team.run(
69        "Check if the auth service is ready and deploy it to staging", stream=True
70    ):
71        # Use isinstance to check for team's pause event
72        if isinstance(run_event, TeamRunPausedEvent):
73            print("Team paused - requires confirmation for team-level tool")
74            for req in run_event.active_requirements:
75                if req.needs_confirmation:
76                    print(f"  Tool: {req.tool_execution.tool_name}")
77                    print(f"  Args: {req.tool_execution.tool_args}")
78                    req.confirm()
79
80            response = team.continue_run(
81                run_id=run_event.run_id,
82                session_id=run_event.session_id,
83                requirements=run_event.requirements,
84                stream=True,
85            )
86            pprint.pprint_run_response(response)

1# Clone and setup repo
2git clone https://github.com/kern-ai/kern.git
3cd kern/cookbook/03_teams/human_in_the_loop
4
5# Create and activate virtual environment
6./scripts/demo_setup.sh
7source .venvs/demo/bin/activate
8
9python team_tool_confirmation_stream.py