Streaming with External Execution of Tool

1"""Team HITL Streaming: Member agent tool with external execution.
2
3This example demonstrates how a team pauses when a member agent's tool
4requires external execution in streaming mode. The tool result is provided
5by the caller rather than being executed by the agent.
6
7Note: When streaming with member agents, use isinstance() with TeamRunPausedEvent
8to distinguish the team's pause from member agent pauses.
9"""
10
11import shlex
12import subprocess
13
14from kern.agent import Agent
15from kern.db.sqlite import SqliteDb
16from kern.models.openai import OpenAIResponses
17from kern.run.team import RunPausedEvent as TeamRunPausedEvent
18from kern.team.team import Team
19from kern.tools import tool
20from kern.utils import pprint
21
22# ---------------------------------------------------------------------------
23# Setup
24# ---------------------------------------------------------------------------
25db = SqliteDb(db_file="tmp/team_hitl_stream.db")
26
27
28# ---------------------------------------------------------------------------
29# Tools
30# ---------------------------------------------------------------------------
31@tool(external_execution=True)
32def run_shell_command(command: str) -> str:
33    """Execute a shell command on the server.
34
35    Args:
36        command (str): The shell command to execute
37    """
38    return subprocess.check_output(shlex.split(command)).decode("utf-8")
39
40
41# ---------------------------------------------------------------------------
42# Create Members
43# ---------------------------------------------------------------------------
44ops_agent = Agent(
45    name="Ops Agent",
46    role="Handles server operations",
47    model=OpenAIResponses(id="gpt-5.2-mini"),
48    tools=[run_shell_command],
49    db=db,
50)
51
52
53# ---------------------------------------------------------------------------
54# Create Team
55# ---------------------------------------------------------------------------
56team = Team(
57    name="SRE Team",
58    members=[ops_agent],
59    model=OpenAIResponses(id="gpt-5.2-mini"),
60    db=db,
61)
62
63
64# ---------------------------------------------------------------------------
65# Run Team
66# ---------------------------------------------------------------------------
67if __name__ == "__main__":
68    for run_event in team.run("List the files in the current directory", stream=True):
69        # Use isinstance to check for team's pause event (not the member agent's)
70        if isinstance(run_event, TeamRunPausedEvent):
71            print("Team paused - requires external execution")
72            for req in run_event.active_requirements:
73                if req.needs_external_execution:
74                    print(f"  Tool: {req.tool_execution.tool_name}")
75                    print(f"  Args: {req.tool_execution.tool_args}")
76
77                    # Execute the tool externally
78                    result = run_shell_command.entrypoint(
79                        **req.tool_execution.tool_args
80                    )
81                    req.set_external_execution_result(result)
82
83            response = team.continue_run(
84                run_id=run_event.run_id,
85                session_id=run_event.session_id,
86                requirements=run_event.requirements,
87                stream=True,
88            )
89            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 external_tool_execution_stream.py