External Tool Execution

1"""
2External Tool Execution
3=============================
4
5Demonstrates resolving external tool execution requirements in team flows.
6"""
7
8from kern.agent import Agent
9from kern.db.sqlite import SqliteDb
10from kern.models.openai import OpenAIResponses
11from kern.team import Team
12from kern.tools import tool
13from kern.utils import pprint
14from rich.console import Console
15from rich.prompt import Prompt
16
17# ---------------------------------------------------------------------------
18# Setup
19# ---------------------------------------------------------------------------
20console = Console()
21
22db = SqliteDb(session_table="team_ext_exec_sessions", db_file="tmp/team_hitl.db")
23
24
25@tool(external_execution=True)
26def send_email(to: str, subject: str, body: str) -> str:
27    """Send an email to someone. Executed externally."""
28    return ""
29
30
31# ---------------------------------------------------------------------------
32# Create Members
33# ---------------------------------------------------------------------------
34email_agent = Agent(
35    name="EmailAgent",
36    model=OpenAIResponses(id="gpt-5.2"),
37    tools=[send_email],
38    db=db,
39    telemetry=False,
40)
41
42# ---------------------------------------------------------------------------
43# Create Team
44# ---------------------------------------------------------------------------
45team = Team(
46    name="CommunicationTeam",
47    model=OpenAIResponses(id="gpt-5.2"),
48    members=[email_agent],
49    db=db,
50    telemetry=False,
51    add_history_to_context=True,
52)
53
54# ---------------------------------------------------------------------------
55# Run Team
56# ---------------------------------------------------------------------------
57if __name__ == "__main__":
58    session_id = "team_email_session"
59    run_response = team.run(
60        "Send an email to john@example.com with subject 'Meeting Tomorrow' and body 'Let's meet at 3pm.'",
61        session_id=session_id,
62    )
63
64    if run_response.is_paused:
65        console.print("[bold yellow]Team is paused - external execution needed[/]")
66
67        for requirement in run_response.active_requirements:
68            if requirement.needs_external_execution:
69                tool_args = requirement.tool_execution.tool_args
70                console.print(
71                    f"Member [bold cyan]{requirement.member_agent_name}[/] needs external execution of "
72                    f"[bold blue]{requirement.tool_execution.tool_name}[/]"
73                )
74                console.print(f"  To: {tool_args.get('to')}")
75                console.print(f"  Subject: {tool_args.get('subject')}")
76                console.print(f"  Body: {tool_args.get('body')}")
77
78                result = Prompt.ask(
79                    "Enter the result of the email send",
80                    default="Email sent successfully",
81                )
82                requirement.set_external_execution_result(result)
83
84        run_response = team.continue_run(run_response)
85
86    pprint.pprint_run_response(run_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.py