ag2-hitl

Human-in-the-loop

When to use

• The agent should ask for confirmation before doing something risky.
• The agent needs information from the user mid-conversation (a password, an API key, missing context).
• A specific tool call should require human approval before it runs (irreversible / expensive / sensitive).
• Quality assurance — show a draft, get human edits/approval before finalising.

Two distinct mechanisms — pick by intent:

Need	Use
Tool asks an open question and waits for a typed answer	context.input() from inside the tool + hitl_hook on the agent
Approve / deny a specific tool call before its body runs	approval_required() tool middleware

Pattern 1 — context.input() for open questions

A tool requests input via Context.input(message, timeout=...). The agent must have a hitl_hook that knows how to collect that input.

from ag2 import Agent, Context, tool
from ag2.events import HumanInputRequest, HumanMessage

@tool
async def execute_query(context: Context) -> str:
    answer = await context.input(
        "Are you sure you want to run this query? (yes/no)",
        timeout=60.0,
    )
    if answer.strip().lower() != "yes":
        return "Query cancelled."
    return "Query executed successfully."

def hitl_hook(event: HumanInputRequest) -> HumanMessage:
    print(f"Agent asks: {event.content}")
    return HumanMessage(content=input("Your answer: "))

agent = Agent("dba", tools=[execute_query], hitl_hook=hitl_hook)

The hook receives a HumanInputRequest (the prompt is in event.content) and returns either a HumanMessage or a plain str (the framework wraps a str via HumanMessage.ensure_message). Both def and async def hooks are supported.

You can also register the hook after construction:

agent = Agent("dba", tools=[execute_query])

@agent.hitl_hook
async def async_hitl_hook(event: HumanInputRequest) -> HumanMessage:
    answer = await collect_from_ui(event.content)
    return HumanMessage(content=answer)