Ask User
Let the model ask the user multiple-choice questions mid-run and wait for the answers.
While Pydantic AI Harness is on 0.x releases, the API may change between minor releases; when it does, deprecation warnings and release-note migration guidance tell you (or your agent) exactly how to upgrade. See the version policy.
An agent given an ambiguous task either guesses or buries a question in its output and stops. Guessing wastes a run; a question in prose is only found by a person reading the whole answer. The model needs a way to ask a small, structured question and get the answer back as data.
AskUser exposes one tool, ask_user_question. The model passes one to ten questions, each
with a short header, the question text, two to six options (a label and an optional
description), and multi_select when several answers are allowed. The capability validates
the call, hands it to your answerer, and returns the picked labels keyed by header. If the
user declines, the model is told so and the run continues.
The capability owns the schema and validation. It never prints, reads stdin, or imports a
terminal library: the answerer is whatever puts the questions in front of a person, and you
supply it. There is no default, because a capability that reads stdin is unusable from a server.
from pydantic_ai import Agent
from pydantic_ai_harness import AskUser
from pydantic_ai_harness.ask_user import AskUserAnswer, AskUserRequest, AskUserResponse
async def pick_first(request: AskUserRequest) -> AskUserResponse:
answers = [AskUserAnswer(header=q.header, selected=(q.options[0].label,)) for q in request.questions]
return AskUserResponse(answers=tuple(answers))
agent = Agent('anthropic:claude-fable-5-1', capabilities=[AskUser(answerer=pick_first)])
pick_first stands in for a real UI. CLAI
ships a terminal menu built on the same protocol; a web form would be another.
An Answerer is one async callable: it takes an AskUserRequest and returns an
AskUserResponse. A plain async def qualifies; a class with async def __call__ does too.
AskUserRequest.questionsholds the validatedQuestionobjects;AskUserRequest.iddistinguishes concurrent or repeated calls so a UI can match its reply to the request.- A completed
AskUserResponsecarries oneAskUserAnswerper question, keyed byheader, with the picked option labels inselected: exactly one unless the question ismulti_select, at least one either way, and no label twice. - An answerer may instead return
AskUserAnswer(header=q.header, custom_answer='My own answer'). Leaveselectedempty. Custom text must be nonblank and contain no control characters other than newlines and no format characters such as zero-width spaces, joiners and bidirectional overrides, so runcheck_responseon text a user typed. Its length is not limited, so cap free text that arrives from a client before returning it. The model receives it as a one-item list under the question’s header. Whether to offer text entry is up to the answerer; the model cannot request it. - When the user declines, return
AskUserResponse(cancelled=True)with no answers. The tool result tells the model the user declined; nothing is raised into the run. - A response that does not fit the request (an unknown header, a label the question did not
offer, several labels on a single-select question, a missing answer) is a bug in the answerer
and raises
ValueError, which fails the run.check_responseis exported so an answerer can validate before returning. - By default, two
ask_user_questioncalls in one model response reach the answerer concurrently, so serialize in the answerer if your UI shows one request at a time. Other tool calls from that response may run while the user decides: an answer shapes the model’s next response, not calls it already made. To run a response’s calls one at a time, wrap the run inwith Agent.parallel_tool_call_execution_mode('sequential'):. - The model writes the questions, and a prompt injection can make it ask for a password or a token. Present the questions as coming from the model, not as a system or sign-in prompt.
The run waits inside the tool call for the answerer to return. Set timeout, in seconds, to
bound that wait. When it runs out, the answerer is cancelled, AskUserAnsweredEvent fires with
timed_out=True and a cancelled response, and the model is told the user did not answer in time
(the TIMED_OUT result) and continues.
The timeout cancels the answerer, so the answerer must be cancellable. A thread cannot be
cancelled: run blocking work with anyio.to_thread.run_sync(..., abandon_on_cancel=True), or the
run waits for the thread and then uses its late answer. An abandoned thread keeps running, so a
terminal answerer should read input with an async prompt library rather than input() in a
thread, which would take the next line the user types.
AskUser(answerer=pick_first, timeout=300)
A web server or a durable worker often cannot hold a run open while a person decides. Pass
answerer=None and every ask_user_question call is deferred instead:
the run ends with DeferredToolRequests output, so your output_type must include it, and you
answer in a later run, from the same process or another one.
AskUserRequest.from_tool_call(call)rebuilds the validated request from a pending call. Itsidis the call’stool_call_id, the same requestAskUserRequestedEventcarried before the run paused.ask_user_result(request, response)checks the response like an answerer’s (raisingValueErrorif it does not fit) and renders the tool result the model would have got inline. Pass it inDeferredToolResults.callsunder the call’s ID.- No
AskUserAnsweredEventfires for a deferred call: the answer comes from you, in the next run.
from pydantic_ai import Agent, AgentRunResult, DeferredToolRequests, DeferredToolResults, ModelMessage
from pydantic_ai_harness import AskUser
from pydantic_ai_harness.ask_user import TOOL_NAME, AskUserRequest, AskUserResponse, ask_user_result
agent = Agent(
'anthropic:claude-fable-5-1',
output_type=[str, DeferredToolRequests],
capabilities=[AskUser(answerer=None)],
)
def pending_questions(requests: DeferredToolRequests) -> dict[str, AskUserRequest]:
"""What the paused run is waiting on, keyed by tool call ID: show these to the user."""
return {
call.tool_call_id: AskUserRequest.from_tool_call(call)
for call in requests.calls
if call.tool_name == TOOL_NAME
}
async def resume(
messages: list[ModelMessage],
questions: dict[str, AskUserRequest],
responses: dict[str, AskUserResponse],
) -> AgentRunResult[str | DeferredToolRequests]:
"""Answer the paused run's questions and carry on."""
calls = {call_id: ask_user_result(request, responses[call_id]) for call_id, request in questions.items()}
return await agent.run(message_history=messages, deferred_tool_results=DeferredToolResults(calls=calls))
Keep the paused run’s all_messages() and its pending questions wherever you keep sessions,
then call resume when the answers arrive. timeout needs an answerer to bound, so combining it
with answerer=None raises UserError.
Two CapabilityEvents let anything else in the run observe the exchange:
| Event | When | Fields |
|---|---|---|
AskUserRequestedEvent | before the answerer is called, or before a deferred call pauses the run | request |
AskUserAnsweredEvent | after it returns, times out, or raises, before the response is checked or the model sees the result | request_id, response, timed_out, failed |
Both dispatch immediately, so a listener that shows a “waiting for you” state sees the wait
start and end in step with the run. When the answerer raises, the answered event carries
failed=True and a cancelled response, then the error propagates from the tool call: a
ModelRetry sends the model a retry prompt, and any other error fails the run. A run
cancelled while the answerer waits emits no answered event, so also clear waiting state when the
run ends. Subscribe with @on_event on a capability or through the run’s event stream:
from pydantic_ai import RunContext
from pydantic_ai.capabilities import AbstractCapability, on_event
from pydantic_ai_harness.ask_user import AskUserAnsweredEvent, AskUserRequestedEvent
class WaitIndicator(AbstractCapability[None]):
@on_event(AskUserRequestedEvent)
async def waiting(self, ctx: RunContext[None], event: AskUserRequestedEvent) -> None:
print(f'waiting on {len(event.request.questions)} question(s)')
@on_event(AskUserAnsweredEvent)
async def done(self, ctx: RunContext[None], event: AskUserAnsweredEvent) -> None:
if event.failed:
print('failed')
elif event.timed_out:
print('timed out')
else:
print('declined' if event.response.cancelled else 'answered')
The tool schema mirrors Code Puppy’s
ask_user_question, so prompts written for it carry over.
| Field | Limit |
|---|---|
questions | 1 to 10 per call; headers unique, ignoring case and compatibility forms such as full-width letters |
header | 1 to 25 characters, one line |
question | 1 to 500 characters, may span lines |
options | 2 to 6 per question; labels unique, ignoring case and compatibility forms such as full-width letters |
label | 1 to 50 characters, one line |
description | optional, at most 200 characters, may span lines |
These strings are drawn on terminals, where an escape sequence in a prompt-injected call is an attack, so control characters and format characters such as zero-width spaces, joiners and bidirectional overrides are rejected. Spaces of any width are allowed, and uniqueness treats most of them as an ordinary space. A call outside these limits, or with a field the schema does not have, goes back to the model as a validation retry and never reaches the answerer. Once validated the questions are frozen: what the answerer sees is what the model asked.
The result is a JSON object mapping each header to the list of picked labels (or one custom answer), or the sentence
The user declined to answer. Continue without the answer, or ask differently if it is essential.
When timeout runs out, it is the TIMED_OUT sentence instead.
The capability adds one instruction: ask when the task is ambiguous and the answer is not in the workspace, offer concrete options, batch related questions, and make a stated choice if the user declines.
AskUser declares the default id 'ask_user'. The id keys the capability’s instructions, so they
can be addressed, and lets defer_loading=True work without one. Two AskUsers on one agent that
share it are one answerer stated twice: they become one when they agree, and the agent raises a
UserError when it’s built if they differ. To keep two, give each its own id and wrap them
in PrefixTools, since both register ask_user_question.
To serve several UIs, pass one answerer that dispatches to them.
Earlier releases declared no default id, so two always failed on the tool name when the run
started. Two identical ones now become one, and two that differ fail when the agent is built.
AskUser emits no spans. Core’s tool-call span already covers the wait, and the two events
above carry what was asked and answered. That span records the tool’s arguments and result,
which here are the questions and the user’s answers, custom text included. Set
include_content=False to leave them out.
Agent.from_spec cannot construct AskUser: the answerer is a live object a spec has no way
to name.
check_response, ask_user_result, TOOL_NAME, DECLINED, TIMED_OUT, and MAX_QUESTIONS are also
exported from pydantic_ai_harness.ask_user, along with AskUserToolset, the toolset AskUser
registers, and ASK_USER_EVENTS, the namespace of both events.
Bases: AbstractCapability[AgentDepsT]
Let the model ask the user multiple-choice questions mid-run.
The capability owns the question schema and validation; the answerer owns the person.
Whatever it is, a terminal menu, a web form, or a scripted function in a test, the run waits
for it inside the tool call and the model gets the picked labels back. Pass answerer=None to
defer each call instead, so the run pauses with DeferredToolRequests until the host answers.
from pydantic_ai import Agent
from pydantic_ai_harness import AskUser
from pydantic_ai_harness.ask_user import AskUserAnswer, AskUserRequest, AskUserResponse
async def pick_first(request: AskUserRequest) -> AskUserResponse:
answers = [AskUserAnswer(header=q.header, selected=(q.options[0].label,)) for q in request.questions]
return AskUserResponse(answers=tuple(answers))
agent = Agent('anthropic:claude-fable-5-1', capabilities=[AskUser(answerer=pick_first)])
Presents each AskUserRequest to the user and returns their AskUserResponse.
None defers every call: the run ends with DeferredToolRequests output, and the host answers
in a later run with ask_user_result in DeferredToolResults.calls.
Type: Answerer | None
Seconds to wait for the answerer before cancelling it and telling the model the user did not answer.
Must be positive; None waits indefinitely. Needs an answerer: a deferred call has nothing to wait on.
Type: float | None Default: None
Stable capability ID, so its instructions are addressable and defer_loading=True needs none.
An agent has one ask_user_question tool, so two AskUser sharing this id are one answerer stated
twice when they agree, and an error when they differ. To keep two, give each its own id and wrap them
in PrefixTools, since their tool names are the same.
Type: str | None Default: 'ask_user'
@classmethod
def combine(
cls,
capabilities: Sequence[AbstractCapability[AgentDepsT]],
) -> AbstractCapability[AgentDepsT]
Two under one id are one answerer stated twice; two that disagree raise rather than merge.
AbstractCapability[AgentDepsT]
def get_instructions() -> AgentInstructions[AgentDepsT] | None
Static, cache-stable guidance on when to ask.
AgentInstructions[AgentDepsT] | None
def get_toolset() -> AgentToolset[AgentDepsT] | None
The ask_user_question tool bound to this capability’s answerer.
AgentToolset[AgentDepsT] | None
Bases: Protocol
Whatever puts the questions in front of a person: a terminal menu, a web form, a scripted test.
A plain async def with this signature satisfies it. Return a cancelled response when the user
declines; raise only for failures that should fail the run.
@async
def __call__(request: AskUserRequest, /) -> AskUserResponse
Put request to the user and return what they picked.
AskUserResponse
One ask_user_question call, handed to the Answerer and carried by AskUserRequestedEvent.
Distinguishes concurrent or repeated calls; a UI matches its reply to it.
A deferred request uses the tool call ID, so from_tool_call rebuilds the same request.
Type: str Default: field(default_factory=(lambda: uuid4().hex))
@classmethod
def from_tool_call(cls, call: ToolCallPart) -> AskUserRequest
Rebuild the request behind a deferred ask_user_question call, to put it to the user.
Raises ValueError when call is for another tool.
AskUserRequest
What the Answerer returns: one answer per question, or cancelled when the user declined.
Picked labels, or a custom answer instead of those labels, for one question.
Bases: BaseModel
One multiple-choice question.
Bases: BaseModel
One choice the user can pick.
Bases: CapabilityEvent
The model asked the user something; the answerer is about to be called, or the call is deferred.
Bases: CapabilityEvent
The answerer returned, timed out, or raised; if it returned, response.cancelled says whether the user declined.
Emitted before the response is checked against the request, so a listener waiting since the request event is released even when the answerer misbehaved and the run is about to fail. Not emitted for a deferred call, whose answer arrives in a later run from the host, or when the run itself is cancelled while the answerer waits: clear waiting state when the run ends.
The answerer ran past AskUser.timeout and was cancelled; response is then a cancelled one.
Type: bool Default: False
The answerer raised; response is then a cancelled one, and the error propagates from the tool call.
Type: bool Default: False