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
How It Works
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
is_paused is set to True
tools_requiring_user_input populated 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 List
from agno.agent import Agent
from agno.tools.function import UserInputField
from agno.models.openai import OpenAIChat
from agno.tools import tool
from agno.tools.toolkit import Toolkit
from agno.tools.user_control_flow import UserControlFlowTools
from agno.utils import pprint
# Example toolkit for handling emails
class 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=OpenAIChat(id="gpt-5-mini"),
tools=[EmailTools(), UserControlFlowTools()],
markdown=True,
)
run_response = agent.run("Send an email with the body 'How is it going in Tokyo?'")
# Use a while loop to continue running until the agent is satisfied with the user input
while run_response.is_paused:
for tool in run_response.tools_requiring_user_input:
input_schema: List[UserInputField] = tool.user_input_schema
for field in input_schema:
# Display field information to the user
print(f"\nField: {field.name} ({field.field_type.__name__}) -> {field.description}")
# Get user input (if the value is not set, it means the user needs to provide the value)
if field.value is None:
user_value = input(f"Please enter a value for {field.name}: ")
field.value = user_value
else:
print(f"Value provided by the agent: {field.value}")
run_response = agent.continue_run(run_id=run_response.run_id, updated_tools=run_response.tools)
# Exit the loop once the agent finishes execution
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.
The While Loop Pattern
Notice the while run_response.is_paused: loop? This is crucial for dynamic user input. The agent might need to request input multiple times:
run_response = agent.run("Send an email and schedule a meeting")
# First iteration: Agent needs email details
while run_response.is_paused:
for tool in run_response.tools_requiring_user_input:
for field in tool.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, updated_tools=run_response.tools)
# 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 case
custom_instructions = """
When you need user input:
1. Only request fields you absolutely need
2. Group related fields together
3. Provide clear, concise descriptions
4. Never request the same information twice
"""
agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
tools=[
EmailTools(),
UserControlFlowTools(
instructions=custom_instructions,
add_instructions=True
)
],
markdown=True,
)
You can also disable the tool entirely if needed:
UserControlFlowTools(enable_get_user_input=False)
Handling Pre-Filled Values
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.
Best Practices
- Always use a while loop: The agent may need multiple rounds of input
- Check field values: Don’t overwrite fields the agent has already filled
- Provide clear prompts: Use the
field.description to help users understand what’s needed
- Validate input: Add your own validation before setting
field.value
- Handle interruptions gracefully: Store
run_id to resume later if needed
Async Support
Dynamic user input works seamlessly with async agents. Use arun() and acontinue_run() for asynchronous flows:
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.user_control_flow import UserControlFlowTools
agent = Agent(
model=OpenAIChat(id="gpt-4o-mini"),
tools=[EmailTools(), UserControlFlowTools()],
markdown=True,
)
run_response = await agent.arun("Send an email")
while run_response.is_paused:
for tool in run_response.tools_requiring_user_input:
for field in tool.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, updated_tools=run_response.tools)
Streaming Support
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 tool in run_event.tools_requiring_user_input:
for field in tool.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,
updated_tools=run_event.tools,
stream=True
):
print(continued_event.content)
When to Use This Pattern
Use Dynamic User Input when:
- The agent needs to adapt its questions based on previous responses
- You want the agent to intelligently determine what information is missing
- The interaction flow changes based on context
Use User Input when:
- 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
Usage Examples
Developer Resources