Let agents request user input dynamically as needed during execution.
Dynamic user input lets your agent decide when it needs information from the user and proactively request it during execution. Unlike the User Input pattern where you predefine which tools need user input, this pattern gives the agent autonomy to pause and ask for information whenever it realizes it doesn’t have what it needs.This pattern is ideal when:
The interaction flow is unpredictable: The agent might need different information based on context
You want a conversational experience: Let the agent guide the user through a form-like interaction
The agent should be intelligent about what it needs: Rather than blindly requesting predefined fields, the agent determines what’s missing
The UserControlFlowTools toolkit provides your agent with a special get_user_input tool. When the agent realizes it’s missing information:
Agent calls get_user_input with a list of fields it needs filled
Execution pauses and requirements are added to the returned RunOutput
user_input_schema populated in the requirement, with the input schema the agent created
You collect the user’s input and set field values in user_input_schema
Call continue_run() to resume with the filled values
Repeat if needed: Agent may request more information based on previous responses
The key difference from other HITL patterns: the agent decides what fields to request and when to request them.
from typing import Listfrom agno.agent import Agentfrom agno.tools.function import UserInputFieldfrom agno.models.openai import OpenAIResponsesfrom agno.tools import toolfrom agno.tools.toolkit import Toolkitfrom agno.tools.user_control_flow import UserControlFlowToolsfrom agno.utils import pprint# Example toolkit for handling emailsclass EmailTools(Toolkit): def __init__(self, *args, **kwargs): super().__init__( name="EmailTools", tools=[self.send_email, self.get_emails], *args, **kwargs ) def send_email(self, subject: str, body: str, to_address: str) -> str: """Send an email to the given address with the given subject and body. Args: subject (str): The subject of the email. body (str): The body of the email. to_address (str): The address to send the email to. """ return f"Sent email to {to_address} with subject {subject} and body {body}" def get_emails(self, date_from: str, date_to: str) -> str: """Get all emails between the given dates. Args: date_from (str): The start date. date_to (str): The end date. """ return [ { "subject": "Hello", "body": "Hello, world!", "to_address": "test@test.com", "date": date_from, }, { "subject": "Random other email", "body": "This is a random other email", "to_address": "john@doe.com", "date": date_to, }, ]agent = Agent( model=OpenAIResponses(id="gpt-5.2"), tools=[EmailTools(), UserControlFlowTools()], markdown=True,)run_response = agent.run("Send an email with the body 'What is the weather in Tokyo?'")# We use a while loop to continue the running until the agent is satisfied with the user inputwhile run_response.is_paused: for requirement in run_response.active_requirements: if requirement.needs_user_input: input_schema: List[UserInputField] = requirement.user_input_schema # type: ignore for field in input_schema: # Get user input for each field in the schema field_type = field.field_type # type: ignore field_description = field.description # type: ignore # Display field information to the user print(f"\nField: {field.name}") # type: ignore print(f"Description: {field_description}") print(f"Type: {field_type}") # Get user input if field.value is None: # type: ignore user_value = input(f"Please enter a value for {field.name}: ") # type: ignore else: print(f"Value: {field.value}") # type: ignore user_value = field.value # type: ignore # Update the field value field.value = user_value # type: ignore run_response = agent.continue_run( run_id=run_response.run_id, requirements=run_response.requirements, ) if not run_response.is_paused: pprint.pprint_run_response(run_response) break
In this example, the agent identifies that it’s missing the email subject and recipient address, so it proactively calls get_user_input to collect that information. Pretty smart!
When your agent calls the get_user_input tool, it provides a list of fields using this format:
{ "field_name": "subject", # The field identifier "field_type": "str", # Python type (str, int, float, bool, list, dict, etc.) "field_description": "The subject of the email" # Helpful description for the user}
The agent constructs these fields intelligently based on what it needs. For example, if it’s trying to send an email but doesn’t have the recipient, it might request:
[ {"field_name": "to_address", "field_type": "str", "field_description": "The email address to send to"}, {"field_name": "subject", "field_type": "str", "field_description": "The subject line for the email"}]
These fields then appear in tool.user_input_schema as UserInputField objects that you can iterate through and fill. For a detailed breakdown of the UserInputField structure, see Understanding UserInputField.
Notice the while run_response.is_paused: loop? This is crucial for dynamic user input, because the agent might request input multiple times:
run_response = agent.run("Send an email and schedule a meeting")# First iteration: Agent needs email detailswhile run_response.is_paused: for requirement in run_response.requirements: if requirement.needs_user_input: for field in requirement.user_input_schema: if field.value is None: field.value = input(f"Enter {field.name}: ") run_response = agent.continue_run(run_id=run_response.run_id, requirements=run_response.requirements) # Agent might pause again if it needs meeting details!
The agent could:
First ask for email details
Send the email
Realize it needs meeting details
Pause again to request those fields
Complete the task
This multi-round capability makes the pattern extremely flexible.
Important: Always check field.value before prompting. If the agent has already filled a field based on context (like extracting it from the user’s message), field.value won’t be None and you shouldn’t overwrite it.
The UserControlFlowTools toolkit comes with default instructions that guide the agent, but you can customize them:
from agno.tools.user_control_flow import UserControlFlowTools# Custom instructions for your use casecustom_instructions = """When you need user input:1. Only request fields you absolutely need2. Group related fields together3. Provide clear, concise descriptions4. Never request the same information twice"""agent = Agent( model=OpenAIResponses(id="gpt-5.2"), tools=[ EmailTools(), UserControlFlowTools( instructions=custom_instructions, add_instructions=True ) ], markdown=True,)
The agent can pre-fill some fields based on the conversation context. This works the same way as in User Input—always check field.value before prompting:
for field in tool.user_input_schema: if field.value is None: user_value = input(f"Please enter {field.name}: ") field.value = user_value else: print(f"{field.name} (provided by agent): {field.value}")
For a more detailed explanation of how pre-filled values work, see the Handling Pre-Filled Values section in the User Input documentation.
Dynamic user input works seamlessly with async agents. Use arun() and acontinue_run() for asynchronous flows:
from agno.agent import Agentfrom agno.models.openai import OpenAIResponsesfrom agno.tools.user_control_flow import UserControlFlowToolsagent = Agent( model=OpenAIResponses(id="gpt-5.2"), tools=[EmailTools(), UserControlFlowTools()], markdown=True,)run_response = await agent.arun("Send an email")while run_response.is_paused: for requirement in run_response.active_requirements: if requirement.needs_user_input: for field in requirement.user_input_schema: if field.value is None: field.value = input(f"Please enter {field.name}: ") run_response = await agent.acontinue_run(run_id=run_response.run_id, requirements=run_response.requirements)
Dynamic user input also works with streaming. The agent will emit events until it needs user input, then pause:
run_response = agent.run("Send an email", stream=True)for run_event in run_response: if run_event.is_paused: for requirement in run_event.active_requirements: if requirement.needs_user_input: for field in requirement.user_input_schema: if field.value is None: field.value = input(f"Please enter {field.name}: ") # Continue streaming for continued_event in agent.continue_run( run_id=run_event.run_id, requirements=run_event.requirements, stream=True ): print(continued_event.content)
You know exactly which tool fields require user input upfront
The input requirements are always the same
You want more explicit control over what gets asked
Remember that tools marked with @tool(requires_user_input=True) are mutually exclusive with @tool(requires_confirmation=True) and @tool(external_execution=True).A tool can only use one of these patterns at a time.