Human in the Loop#

Agents often need guidance or approval from human users to proceed safely and effectively. The Human-in-the-Loop (HITL) feature allows an agent to temporarily pause its execution and wait for human input before continuing.

What is Human in the Loop? Why do we need it?#

Human-in-the-Loop (HITL) is a pattern where a human user is integrated into the decision-making process of an autonomous system. In the context of the agent framework, it provides a structured way to ask for human confirmation or additional information during the execution of a tool.

We need HITL for several reasons:

• Safety: To prevent the agent from performing irreversible or harmful actions (e.g., executing arbitrary code, dropping a database table, or making financial transactions).
• Quality Assurance: To allow humans to review and approve generated artifacts (e.g., code, emails, or reports) before they are finalized.
• Handling Ambiguity: To provide the agent with additional context or clarification when it encounters a situation it cannot resolve autonomously.

Typical Use Cases#

• Execution Approval: Asking the user "Are you sure you want to execute this shell command?" before proceeding.
• Requesting Missing Information: Prompting the user for a password, an API key, or a specific piece of data required by a tool.
• Content Review: Displaying an AI-generated draft to the user and asking for approval or edit suggestions.

HITL Usage from Context#

You can request human input directly from within a tool using the Context object. The ctx.input() method pauses the tool's execution until the user provides a response.

Here is an example of a tool that asks for human confirmation:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from ag2 import Context, Agent, tool

@tool
async def execute_query(context: Context) -> str:
    # Pause and ask for human input
    user_response = await context.input(
        "Are you sure you want to run this query? (yes/no)",
        timeout=60.0
    )

    if user_response.strip().lower() != "yes":
        return "Query cancelled."

    return "Query executed successfully."

agent = Agent(
    name="my_agent",
    tools=[execute_query]
)

HITL Registration#

To handle the input requests made by context.input(), you must register a HITL hook on your Agent.

A HITL hook is a callback function that consumes a HumanInputRequest event and returns a HumanMessage. Like tools, HITL hooks support all context features, including dependency injection and variables.

By Argument#

You can register a HITL hook by passing it as the hitl_hook argument when initializing the Agent.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

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

def my_hitl_hook(event: HumanInputRequest) -> HumanMessage:
    # event.content contains the prompt passed to context.input()
    print(f"Agent asks: {event.content}")

    # Collect input from the user (e.g., via standard input)
    user_input = input("Your answer: ")

    return HumanMessage(content=user_input)

agent = Agent(
    name="my_agent",
    hitl_hook=my_hitl_hook,
)

By Decorator#

Alternatively, you can register or override a HITL hook using the @my_agent.hitl_hook decorator after the agent has been created.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

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

agent = Agent(name="my_agent", tools=[execute_query])

@agent.hitl_hook
async def async_hitl_hook(event: HumanInputRequest) -> HumanMessage:
    # An asynchronous hook example
    print(f"Prompt: {event.content}")

    # A hypothetical async UI function
    user_input = await get_input_from_ui()

    return HumanMessage(content=user_input)

When Nobody Can Answer#

A tool that raises is recorded as a failed tool call and the turn carries on: the model reads the error and decides what to do next. A human-input failure is not treated that way. Every way context.input() can end without an answer ends the turn instead:

All three are HumanInputError, which is the type tool execution refuses to record as a tool result. The same holds one level down: a sub-agent whose question nobody could answer fails the delegating turn rather than reporting a failed sub-task.

The difference matters most under approval_required(). An approval that could never be requested is not the same as a tool that broke — reported as a tool failure, it invites the model to look for another route to the same effect, which is the opposite of what the middleware is for. The gated tool does not run in any of the three cases, including the timeout.

Two ways to carry on regardless:

Catch it at the call site, when the tool has something sensible to do without an answer:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from ag2 import Context
from ag2.exceptions import HumanInputError

async def execute_query(query: str, context: Context) -> str:
    try:
        await context.input(f"Approve this query?\n{query}", timeout=60)
    except HumanInputError:
        return "Skipped: nobody was available to approve the query."

    return "Query executed successfully."

Catch HumanInputError to cover all three; catch one subclass to handle only that case.

Answer inside the hook, when the application wants a policy rather than a failure. For approval_required(), a reply other than y/yes/1/always is a denial:

1
2
3
4
5

async def ask_our_user(event: HumanInputRequest) -> str:
    try:
        return await my_approval_queue.ask(event.content)
    except QueueUnavailable:
        return "n"

A hook that raises is wrapped rather than re-typed, so the original stays reachable for an application that wants to tell its own failures apart:

1
2
3
4
5
6
7

from ag2.exceptions import HumanInputFailedError

try:
    await agent.ask("...")
except HumanInputFailedError as exc:
    if isinstance(exc.cause, QueueUnavailable):
        ...

The conversation stays usable#

The failure escapes between a tool call and its result, which would otherwise leave the history holding an assistant tool call that nothing answers — a shape providers reject. The turn closes that call off with a stand-in result before the exception reaches you, so a stream you reuse (a stream= passed to a second ask(), an ACP or MCP session, a persistent_stream()) still works on its next turn:

1
2
3
4
5
6
7
8
9

from ag2.exceptions import HumanInputError
from ag2.stream import MemoryStream

stream = MemoryStream()
try:
    await agent.ask("Delete the stale accounts.", stream=stream)
except HumanInputError:
    pass  # nobody could approve it
reply = await agent.ask("What did you manage to do?", stream=stream)

The model reads the stand-in on the next turn, so it is told the call was abandoned rather than left to guess from a gap in the transcript.

The rest of the batch#

A model can ask for several tools at once, and the others in that batch are stopped as the turn ends — none of them gets to write a result into a conversation the caller has already been told about.