Human-in-the-Loop Tools¶

RoomKit's human-in-the-loop primitive lets AI tool calls pause execution, request input from a human user, and resume with the answer. This enables tools like AskUserQuestion, confirmation dialogs, and any workflow where the AI needs to collect information from a person before continuing.

Why Human-in-the-Loop?¶

AI agents often need to ask clarifying questions, request approval, or collect data from users mid-conversation. Without a dedicated primitive, developers resort to workarounds:

Wrapping tool handlers — only intercepts tools in the admin chain, not the AI's own tools
BEFORE_TOOL_USE hooks — can allow, deny, or rewrite arguments, but can't provide a result or pause the loop
External state management — manual asyncio.Event tracking scattered across REST endpoints

RoomKit's HumanInputHandler solves this as a first-class feature with two layers:

HumanInputHandler — core async primitive for create / wait / resolve / reject
HumanInputToolHandler — the tools that ask a person, served by an AIChannel, a RealtimeVoiceChannel or a conference given it as human_input_handler=

Quick Start¶

Native AIChannel (built-in tools)¶

For AI agents that define their own human-input tools:

from roomkit import RoomKit, HumanInputToolHandler, HookTrigger, HookExecution
from roomkit.channels.ai import AIChannel
from roomkit.providers.ai.anthropic import AnthropicAIProvider

kit = RoomKit()

# Define which tools need human input
human = HumanInputToolHandler(
    tool_names={"AskUserQuestion", "ConfirmAction"},
    timeout=300,  # 5 minutes
    # A tool the turn does not offer is dropped before any handler sees it.
    # Declare the definition here, or on the channel via `tools=`.
    tool_definitions=[
        AITool(
            name="AskUserQuestion",
            description="Ask the user a question and wait for the answer.",
            parameters={
                "type": "object",
                "properties": {"question": {"type": "string"}},
                "required": ["question"],
            },
        ),
    ],
)

ai = AIChannel(
    "agent",
    provider=AnthropicAIProvider(model="claude-opus-5", api_key="..."),
    system_prompt="You are a helpful assistant. Use AskUserQuestion to ask the user.",
    tool_handler=my_other_tools,        # Your regular tools
    human_input_handler=human,           # Human-input tools
)

kit.register_channel(ai)

# React when the AI asks for human input
@kit.hook(HookTrigger.ON_USER_INPUT_REQUIRED, execution=HookExecution.SYNC)
async def notify_user(event, ctx):
    # Broadcast to frontend via WebSocket, REST, etc.
    # `actor_id` is who to ask: the participant whose turn raised the request.
    # `None` when the turn had no author — see the note below.
    await ws_manager.notify(event.actor_id or BROADCAST, {
        "type": "question_pending",
        "pending_id": event.pending_id,
        "tool_name": event.tool_name,
        "arguments": event.arguments,
    })

# When the user answers (from your REST endpoint, WebSocket handler, etc.)
human.handler.resolve(pending_id, '{"answer": "blue"}')

Voice channels¶

A RealtimeVoiceChannel and a conference's realtime configuration take the same option, with the same rules:

from roomkit import ConferenceChannel, ConferenceRealtimeConfig, RealtimeVoiceChannel

voice = RealtimeVoiceChannel(
    "voice",
    provider=provider,
    transport=transport,
    human_input_handler=human,   # declared in every session, served first
)

conference = ConferenceChannel(
    "conf",
    backend=backend,
    realtime=ConferenceRealtimeConfig(provider=provider, human_input_handler=human),
)

On every door of the channel (the provider's function call, a call recovered from speech, a reasoning backend's call, a conference's call) the tool is served by the channel before tool_handler, under the handler's own timeout rather than the channel's default call bound (10 s on a voice channel), and each request fires ON_USER_INPUT_REQUIRED with channel_type naming the door (realtime_voice, conference). The requests still open are settled when the channel closes, and on a conference when its realtime provider is unplugged.

A HumanInputToolHandler passed as tool_handler= is served as any host handler, without those rules; a RealtimeVoiceChannel logs a warning pointing to human_input_handler=.

External Provider (Claude Code)¶

For providers that execute tools internally (like Claude Code sandboxes), use HumanInputHandler directly inside your ExternalToolHandler. The channel decides who serves each call: one the provider already ran is reported — the provider says so on the call (AIToolCall.served = ServedCall(result=..., is_error=...)), never in its arguments, so a _result key the model writes is just an argument and its call goes through the gate; a pending call to a tool the channel does not serve goes to the ExternalToolHandler; every other call is the channel's own, gate first, so a BEFORE_TOOL_USE block refuses it even on a channel without a tool_handler.

from roomkit import HumanInputRejectedError
from roomkit.tools.human_input import HumanInputHandler

handler = HumanInputHandler()

# Inside your ExternalToolHandler.process_tool_call():
async def process_tool_call(self, tool_name, tool_input, *, room_id=None, **kw):
    if tool_name == "AskUserQuestion":
        pending = await handler.create(
            tool_name, tool_input,
            room_id=room_id or "",
            tool_call_id=kw.get("tool_call_id", ""),
            channel_id=self._channel_id,
        )
        # ON_USER_INPUT_REQUIRED hook fires via _on_input_required callback
        try:
            result = await handler.wait(pending.pending_id, timeout=300)
        except HumanInputRejectedError as rejected:
            # A refusal: the model reads why (RFC §9.3).
            return ToolDecision(approved=False, reason=f"Rejected: {rejected}")
        return ToolDecision(approved=False, reason=result)
    return ToolDecision(approved=True)

# When user answers:
handler.resolve(pending_id, answer_json)

On an AI channel, a turn cancelled while the handler still waits cancels process_tool_call and calls the handler's on_tool_cancelled(tool_name, tool_input, tool_call_id=..., room_id=...); a call the turn cut before the handler was asked comes there too. A steering stop or a transport that stops reading lets a pending decision finish, and the call is reported with its outcome. The default reports the call to ON_TOOL_CALL's observers as cancelled, once; override it to withdraw the question the human was shown, then call await super().on_tool_cancelled(...) to keep the report. It runs in the turn's teardown: keep it short.

A call the handler refuses (process_tool_call returns approved=False) comes back to it through on_tool_refused(tool_name, tool_input, reason, tool_call_id=..., room_id=...), not on_tool_result; the default reports it to ON_TOOL_CALL's observers only, marked refused. A decision that carries a detail (what failed, for a refusal that came from a failure, as PolicyExternalToolHandler sets it for a BEFORE_TOOL_USE hook that failed closed) reaches on_tool_refused(..., detail=...) and the report's error_detail; it is passed only to an override that takes it, so write overrides with **kwargs. A call the channel refuses itself (its arguments were cut) and a process_tool_call that raises are reported by the channel, the latter with what failed (error_detail).

How It Works¶

Data Flow¶

AI calls AskUserQuestion(questions=[...])
  → HumanInputToolHandler intercepts (tool name matches)
  → HumanInputHandler.create() registers pending request — answerable from here
  → HumanInputHandler.wait() blocks (asyncio.Event)
    ⇄ alongside: ON_USER_INPUT_REQUIRED hooks run (sync, can deny)
        → App broadcasts "question_pending" to frontend via WebSocket
  → User sees question in UI, selects answer
  → App calls handler.resolve(pending_id, answer_json)
  → wait() unblocks, returns answer as tool result
  → AI continues with user's response

The notification runs alongside the wait, not in front of it. A user who answers while a slow broadcast is still fanning out — or while a hook is burning its 30-second budget — is answering a request that is already listening, and the tool resumes on the answer. A denial from an ON_USER_INPUT_REQUIRED hook still rejects the request, wherever wait() has got to.

Hook Execution Order¶

Three hooks interact during a human-input tool call:

Order	Hook	Type	Purpose
1	`BEFORE_TOOL_USE`	Sync	Gate or rewrite the validated arguments before execution
2	`ON_USER_INPUT_REQUIRED`	Sync	Notify: broadcast question to user; can deny
3	`ON_TOOL_CALL`	Sync	Observe: tool completed with result

BEFORE_TOOL_USE fires before the handler — it can deny the tool (e.g., rate limiting) or return replacement arguments under metadata["arguments"]. RoomKit validates the replacement against the declared tool schema before execution. ON_USER_INPUT_REQUIRED is scheduled by create() and keeps sync semantics — hooks run in priority order and a HookResult.block() rejects the request — but it does not hold up the wait. ON_TOOL_CALL fires after wait() returns — standard observability.

Who Owns the Cleanup¶

wait() owns it. It retires the request when the answer, the rejection, or the timeout arrives — callers should not track requests on the side and drop them, because a request removed under wait()'s feet is an answer turned into an error.

For a runtime that owns its own tool loop and carries the answer back another way — nobody calls wait() — say so at the call site with create_detached(), and free the request with release():

pending = await handler.create_detached("AskUserQuestion", args, room_id=room_id)
# ... answer travels back through the runtime's own path ...
handler.release(pending.pending_id)

release() rejects a request that is still unanswered, so a stray waiter unblocks rather than hanging.

Late and Repeated Reads¶

An outcome that has been recorded stays readable. A request settling — answered, rejected, timed out — goes into a bounded retention (the last 128 outcomes by default, HumanInputHandler(retention=…)), and wait() replays it once the request has left the active set: the same answer, the same rejection, the same timeout. Only a genuinely unknown id — never seen, or evicted from the retention — raises ValueError.

The retention fills when the answer arrives, not when someone reads it, so a host that keeps its own bookkeeping and drops the request on resolve() cannot turn the answer it just recorded into a tool failure for the model that asked the question.

Parallel Tool Execution¶

When the AI calls multiple tools in one round, asyncio.gather runs them concurrently. If one tool is a human-input tool, it blocks while the others complete normally. The tool loop waits for all results before the next AI generation round.

API Reference¶

HumanInputHandler¶

The core primitive that manages pending requests:

Method	Signature	Description
`create`	`async (tool_name, arguments, *, room_id, tool_call_id, channel_id) → PendingInput`	Register a pending request and schedule the `_on_input_required` callback. `wait()` owns the cleanup
`create_detached`	`async (…same…) → PendingInput`	Same, for a request no one will `wait()` on. The caller owns the cleanup and MUST `release()` it
`wait`	`async (pending_id, *, timeout=300) → str`	Block until resolved/rejected/timeout; replays an outcome already consumed
`resolve`	`(pending_id, result) → bool`	Unblock with answer. Returns `True` if found
`reject`	`(pending_id, reason="") → bool`	Unblock with error. Returns `True` if found
`release`	`(pending_id) → bool`	Drop a request the caller owns; rejects it if still unanswered. Returns `True` if one was dropped
`close`	`async (*, channel_id=None)`	Reject active work and prevent later creates globally, or only for one channel
`pending`	`property → dict[str, PendingInput]`	Snapshot of active pending requests

HumanInputHandler(retention=128) sets how many consumed outcomes stay replayable; retention=0 switches the retention off. One handler may be shared by several channels: framework callbacks are routed by channel_id, and closing one channel leaves the other owners available. A create for the closed scope raises RuntimeError rather than arming work that can outlive its channel.

HumanInputToolHandler¶

The tools that ask a person, for AIChannel, RealtimeVoiceChannel and ConferenceRealtimeConfig:

human = HumanInputToolHandler(
    tool_names={"AskUserQuestion"},  # Tools that need human input
    timeout=300,                      # Seconds before timeout error
    handler=None,                     # Optional: share a HumanInputHandler
    tool_definitions=[...],           # Definitions that put those tools in the turn
)

Pass as human_input_handler=human: the channel declares tool_definitions and serves tool_names itself, before its tool_handler; replacing channel.tool_handler later leaves them served
Access the inner handler via human.handler for resolve() / reject()
await human.ask(name, arguments, channel_type=...) asks on a door of that channel type; called as a ToolHandler, it asks on an AI channel's door
Falls through for non-matching tools (works with compose_tool_handlers)

tool_names gates dispatch; it does not offer the tool

A tool absent from the turn's resolved toolset is dropped by the loop before any handler runs — the agent gets an error and carries on, and no human is ever asked. Either pass tool_definitions here or declare the same tools on the channel (tools= / binding metadata).

AIChannel logs a warning (once per channel per name) when it intercepts a tool it never offers, so the misconfiguration shows up in the log rather than as an agent that quietly never asks anything.

PendingInput¶

Mutable dataclass representing a pending request:

Field	Type	Description
`pending_id`	`str`	Unique ID for resolving
`tool_name`	`str`	Name of the tool
`arguments`	`dict`	Tool arguments
`room_id`	`str`	Room ID
`tool_call_id`	`str`	Provider-assigned tool call ID
`channel_id`	`str`	Originating channel
`status`	`PendingInputStatus`	`PENDING` / `RESOLVED` / `REJECTED` / `TIMED_OUT`
`result`	`str \\| None`	Answer (set on resolve)
`detached`	`bool`	No one will `wait()` on it — its creator frees it with `release()`
`created_at`	`datetime`	Timestamp
`actor_id`	`str \\| None`	Participant whose turn raised the request

PendingInputEvent¶

Frozen dataclass passed to ON_USER_INPUT_REQUIRED hooks:

@kit.hook(HookTrigger.ON_USER_INPUT_REQUIRED, execution=HookExecution.SYNC)
async def on_input(event, ctx):
    print(event.pending_id)    # ID for resolving
    print(event.tool_name)     # "AskUserQuestion"
    print(event.arguments)     # {"questions": [...]}
    print(event.room_id)       # Room context
    print(event.actor_id)      # "alice" — whose turn raised it, or None

Who the Request Is For¶

actor_id carries the participant whose turn raised the request, taken from the tool loop (current_tool_actor_id()). One AIChannel object serves every room and speaker it is attached to, so without it a notification layer can only broadcast — and in a room where two people talk to one agent, whoever answers first answers for someone else.

It is None when the turn had no author (a system injection, a webhook, a scheduled run) or when a caller driving its own tool loop did not supply one. And like the accessor it comes from, it names the turn without authenticating it: resolve it against the room's roster before treating it as a principal — see What a handler knows about the call.

Timeout Handling¶

If the user doesn't respond within the timeout:

wait() raises TimeoutError
HumanInputToolHandler catches it and fails the call (ToolFailedError): the tool ran and nobody answered, so the call is failed (is_error, refused=False), the ON_TOOL_CALL observers see it, and the AI reads:
```
{"error": "Human input timed out after 300s for tool 'AskUserQuestion'"}
```
A request that was rejected (by the human, or an ON_USER_INPUT_REQUIRED hook) makes wait() raise HumanInputRejectedError, a RuntimeError; the tool refuses the call (ToolRefusedError) and the AI reads why
A request the handler gave up on (closed, or released, before an answer) makes wait() raise a plain RuntimeError, as calling a closed handler does: like any other error, it takes the generic failure path, and the AI reads that the call failed, never the error's message
A request whose waiting call is cut (the turn cancelled, the session ended) is withdrawn: a late answer is not taken (resolve() returns False)
The AI sees the error and can retry, skip, or inform the user

Full Example¶

See examples/ai_human_input.py for a complete working example with:

MockAIProvider simulating an AskUserQuestion tool call
HumanInputToolHandler pausing the tool loop
Async resolution simulating a user answering
ON_USER_INPUT_REQUIRED hook for notifications