Executor HITL

1from kern.agent import Agent
2from kern.db.sqlite import SqliteDb
3from kern.models.openai import OpenAIResponses
4from kern.tools import tool
5from kern.workflow.step import Step
6from kern.workflow.workflow import Workflow
7
8
9@tool(requires_confirmation=True)
10def get_the_weather(city: str) -> str:
11    return f"It is 70 degrees and cloudy in {city}"
12
13
14weather_agent = Agent(
15    name="WeatherAgent",
16    model=OpenAIResponses(id="gpt-5.4"),
17    tools=[get_the_weather],
18    db=SqliteDb(db_file="workflow.db"),
19)
20
21workflow = Workflow(
22    name="WeatherWorkflow",
23    db=SqliteDb(db_file="workflow.db"),
24    steps=[Step(name="get_weather", agent=weather_agent)],
25)
26
27response = workflow.run("What is the weather in Tokyo?")
28
29if response.is_paused:
30    for step_req in response.step_requirements or []:
31        if step_req.requires_executor_input:
32            for executor_req in step_req.executor_requirements or []:
33                executor_req.confirm()  # or executor_req.reject(note="...")
34
35    response = workflow.continue_run(response)
36
37print(response.content)

1from kern.tools import tool
2
3
4@tool(requires_confirmation=True)
5def delete_user(user_id: str) -> str:
6    ...
7
8
9@tool(requires_user_input=True, user_input_fields=["recipient"])
10def send_money(amount: float, recipient: str, note: str) -> str:
11    ...
12
13
14@tool(external_execution=True)
15def run_query(sql: str) -> str:
16    ...

1step_req.executor_name                          # "WeatherAgent"
2step_req.executor_type                          # "agent" | "team"
3executor_req = step_req.executor_requirements[0]
4executor_req.tool_execution.tool_name           # "get_the_weather"
5executor_req.tool_execution.tool_args           # {"city": "Tokyo"}
6executor_req.needs_confirmation                 # True for requires_confirmation
7executor_req.needs_user_input                   # True for requires_user_input
8executor_req.needs_external_execution           # True for external_execution

1for step_req in response.step_requirements or []:
2    if step_req.requires_executor_input:
3        for executor_req in step_req.executor_requirements or []:
4            if executor_req.needs_confirmation:
5                if user_approves():
6                    executor_req.confirm()
7                else:
8                    executor_req.reject(note="User declined")
9
10response = workflow.continue_run(response)

1for step_req in response.step_requirements or []:
2    if step_req.requires_executor_input:
3        for executor_req in step_req.executor_requirements or []:
4            if executor_req.needs_user_input:
5                values = {
6                    field.name: prompt(field.name)
7                    for field in executor_req.user_input_schema or []
8                }
9                executor_req.provide_user_input(values)
10
11response = workflow.continue_run(response)

1for step_req in response.step_requirements or []:
2    if step_req.requires_executor_input:
3        for executor_req in step_req.executor_requirements or []:
4            if executor_req.needs_external_execution:
5                tool_name = executor_req.tool_execution.tool_name
6                tool_args = executor_req.tool_execution.tool_args
7
8                # Replace this with your own dispatch logic, for example:
9                #   result = my_backend.run(tool_name, **tool_args)
10                #   result = subprocess.run(...).stdout
11                #   result = await microservice.call(tool_name, tool_args)
12                result = my_dispatcher(tool_name, tool_args)
13
14                executor_req.tool_execution.result = result
15
16response = workflow.continue_run(response)

1from kern.run.workflow import (
2    StepExecutorPausedEvent,
3    StepExecutorContinuedEvent,
4    WorkflowCompletedEvent,
5)
6
7for event in workflow.run("What is the weather in Tokyo?", stream=True):
8    if isinstance(event, StepExecutorPausedEvent):
9        print(f"Paused: {event.executor_name} ({event.executor_type})")
10    elif isinstance(event, StepExecutorContinuedEvent):
11        print(f"Resumed: {event.executor_name}")
12    elif isinstance(event, WorkflowCompletedEvent):
13        print("Done")
14
15# The full WorkflowRunOutput is persisted to the session, not yielded in the stream.
16session = workflow.get_session()
17run_output = session.runs[-1] if session and session.runs else None
18
19if run_output and run_output.is_paused:
20    for step_req in run_output.step_requirements or []:
21        for executor_req in step_req.executor_requirements or []:
22            executor_req.confirm()
23
24    for event in workflow.continue_run(run_output, stream=True):
25        if hasattr(event, "content") and event.content:
26            print(event.content, end="", flush=True)

1from kern.workflow.condition import Condition
2from kern.workflow.step import Step
3
4workflow = Workflow(
5    name="ConditionExecutorHITL",
6    db=db,
7    steps=[
8        Step(name="gather_data", executor=gather_data),
9        Condition(
10            name="analysis_decision",
11            evaluator=lambda step_input: True,
12            steps=[Step(name="detailed_analysis", agent=analysis_agent)],
13            else_steps=[Step(name="quick_summary", executor=quick_summary)],
14        ),
15        Step(name="report", executor=generate_report),
16    ],
17)