Executor HITL

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.tools import tool
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow


@tool(requires_confirmation=True)
def get_the_weather(city: str) -> str:
    return f"It is 70 degrees and cloudy in {city}"


weather_agent = Agent(
    name="WeatherAgent",
    model=OpenAIResponses(id="gpt-5.4"),
    tools=[get_the_weather],
    db=SqliteDb(db_file="workflow.db"),
)

workflow = Workflow(
    name="WeatherWorkflow",
    db=SqliteDb(db_file="workflow.db"),
    steps=[Step(name="get_weather", agent=weather_agent)],
)

response = workflow.run("What is the weather in Tokyo?")

if response.is_paused:
    for step_req in response.step_requirements or []:
        if step_req.requires_executor_input:
            for executor_req in step_req.executor_requirements or []:
                executor_req["confirmation"] = True  # False to reject
                executor_req["tool_execution"]["confirmed"] = True

    response = workflow.continue_run(response)

print(response.content)

from agno.tools import tool


@tool(requires_confirmation=True)
def delete_user(user_id: str) -> str:
    ...


@tool(requires_user_input=True, user_input_fields=["recipient"])
def send_money(amount: float, recipient: str, note: str) -> str:
    ...


@tool(external_execution=True)
def run_query(sql: str) -> str:
    ...

step_req.executor_name                                         # "WeatherAgent"
step_req.executor_type                                         # "agent" | "team"
executor_req = step_req.executor_requirements[0]               # a dict
executor_req["tool_execution"]["tool_name"]                    # "get_the_weather"
executor_req["tool_execution"]["tool_args"]                    # {"city": "Tokyo"}
executor_req["tool_execution"]["requires_confirmation"]        # True for requires_confirmation
executor_req["tool_execution"]["requires_user_input"]          # True for requires_user_input
executor_req["tool_execution"]["external_execution_required"]  # True for external_execution

for step_req in response.step_requirements or []:
    if step_req.requires_executor_input:
        for executor_req in step_req.executor_requirements or []:
            tool_exec = executor_req["tool_execution"]
            if tool_exec.get("requires_confirmation"):
                approved = user_approves()
                executor_req["confirmation"] = approved
                tool_exec["confirmed"] = approved
                if not approved:
                    tool_exec["confirmation_note"] = "User declined"

response = workflow.continue_run(response)

for step_req in response.step_requirements or []:
    if step_req.requires_executor_input:
        for executor_req in step_req.executor_requirements or []:
            tool_exec = executor_req["tool_execution"]
            if tool_exec.get("requires_user_input"):
                values = {
                    field["name"]: prompt(field["name"])
                    for field in executor_req.get("user_input_schema") or []
                }
                for schema in (
                    executor_req.get("user_input_schema"),
                    tool_exec.get("user_input_schema"),
                ):
                    for field in schema or []:
                        if field["name"] in values:
                            field["value"] = values[field["name"]]

response = workflow.continue_run(response)

for step_req in response.step_requirements or []:
    if step_req.requires_executor_input:
        for executor_req in step_req.executor_requirements or []:
            tool_exec = executor_req["tool_execution"]
            if tool_exec.get("external_execution_required"):
                tool_name = tool_exec["tool_name"]
                tool_args = tool_exec["tool_args"]

                # Replace this with your own dispatch logic, for example:
                #   result = my_backend.run(tool_name, **tool_args)
                #   result = subprocess.run(...).stdout
                #   result = await microservice.call(tool_name, tool_args)
                result = my_dispatcher(tool_name, tool_args)

                executor_req["external_execution_result"] = result
                tool_exec["result"] = result

response = workflow.continue_run(response)

from agno.run.workflow import (
    StepExecutorPausedEvent,
    StepExecutorContinuedEvent,
    WorkflowCompletedEvent,
)

for event in workflow.run("What is the weather in Tokyo?", stream=True):
    if isinstance(event, StepExecutorPausedEvent):
        print(f"Paused: {event.executor_name} ({event.executor_type})")
    elif isinstance(event, StepExecutorContinuedEvent):
        print(f"Resumed: {event.executor_name}")
    elif isinstance(event, WorkflowCompletedEvent):
        print("Done")

# The full WorkflowRunOutput is persisted to the session, not yielded in the stream.
session = workflow.get_session()
run_output = session.runs[-1] if session and session.runs else None

if run_output and run_output.is_paused:
    for step_req in run_output.step_requirements or []:
        for executor_req in step_req.executor_requirements or []:
            executor_req["confirmation"] = True
            executor_req["tool_execution"]["confirmed"] = True

    for event in workflow.continue_run(run_output, stream=True):
        if hasattr(event, "content") and event.content:
            print(event.content, end="", flush=True)

from agno.workflow.condition import Condition
from agno.workflow.step import Step

workflow = Workflow(
    name="ConditionExecutorHITL",
    db=db,
    steps=[
        Step(name="gather_data", executor=gather_data),
        Condition(
            name="analysis_decision",
            evaluator=lambda step_input: True,
            steps=[Step(name="detailed_analysis", agent=analysis_agent)],
            else_steps=[Step(name="quick_summary", executor=quick_summary)],
        ),
        Step(name="report", executor=generate_report),
    ],
)