User Input Required Async
Collect specific user input fields with the requires_user_input parameter in an async environment.
Create a Python file
import asyncio
from typing import List
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.tools import tool
from agno.tools.function import UserInputField
from agno.utils import pprint
# Explicitly list fields to withhold from the model and collect from the user.
@tool(requires_user_input=True, user_input_fields=["to_address"])
def send_email(subject: str, body: str, to_address: str) -> str:
"""
Send an email.
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}"
agent = Agent(
model=OpenAIResponses(id="gpt-5.2"),
tools=[send_email],
markdown=True,
db=SqliteDb(db_file="tmp/example.db"),
)
async def main():
run_response = await agent.arun("Send an email with the subject 'Hello' and the body 'Hello, world!'")
if run_response.is_paused:
for requirement in run_response.active_requirements:
if requirement.needs_user_input:
input_schema: List[UserInputField] = requirement.user_input_schema
for field in input_schema:
field_type = field.field_type
field_description = field.description
print(f'\nField: {field.name}')
print(f'Description: {field_description}')
print(f'Type: {field_type}')
if field.value is None:
user_value = input(f'Please enter a value for {field.name}: ')
else:
print(f'Value: {field.value}')
user_value = field.value
field.value = user_value
run_response = await agent.acontinue_run(run_response=run_response)
pprint.pprint_run_response(run_response)
asyncio.run(main())
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activateInstall dependencies
uv pip install -U agno openai sqlalchemyExport your OpenAI API key
export OPENAI_API_KEY="your_openai_api_key_here"Run Agent
python user_input_required_async.py