Kern

Dynamic User Input

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

How It Works

The UserControlFlowTools toolkit provides your agent with a special get_user_input tool. When the agent realizes it's missing information:

  1. Agent calls get_user_input with a list of fields it needs filled
  2. Execution pauses and requirements are added to the returned RunOutput
  3. user_input_schema populated in the requirement, with the input schema the agent created
  4. You collect the user's input and set field values in user_input_schema
  5. Call continue_run() to resume with the filled values
  6. 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.

1from typing import List
2
3from kern.agent import Agent
4from kern.tools.function import UserInputField
5from kern.models.openai import OpenAIResponses
6from kern.tools import tool
7from kern.tools.toolkit import Toolkit
8from kern.tools.user_control_flow import UserControlFlowTools
9from kern.utils import pprint
10
11# Example toolkit for handling emails
12class EmailTools(Toolkit):
13    def __init__(self, *args, **kwargs):
14        super().__init__(
15            name="EmailTools", tools=[self.send_email, self.get_emails], *args, **kwargs
16        )
17
18    def send_email(self, subject: str, body: str, to_address: str) -> str:
19        """Send an email to the given address with the given subject and body.
20
21        Args:
22            subject (str): The subject of the email.
23            body (str): The body of the email.
24            to_address (str): The address to send the email to.
25        """
26        return f"Sent email to {to_address} with subject {subject} and body {body}"
27
28    def get_emails(self, date_from: str, date_to: str) -> str:
29        """Get all emails between the given dates.
30
31        Args:
32            date_from (str): The start date.
33            date_to (str): The end date.
34        """
35        return [
36            {
37                "subject": "Hello",
38                "body": "Hello, world!",
39                "to_address": "test@test.com",
40                "date": date_from,
41            },
42            {
43                "subject": "Random other email",
44                "body": "This is a random other email",
45                "to_address": "john@doe.com",
46                "date": date_to,
47            },
48        ]
49
50
51agent = Agent(
52    model=OpenAIResponses(id="gpt-5.2"),
53    tools=[EmailTools(), UserControlFlowTools()],
54    markdown=True,
55)
56
57run_response = agent.run("Send an email with the body 'What is the weather in Tokyo?'")
58
59# We use a while loop to continue the running until the agent is satisfied with the user input
60while run_response.is_paused:
61    for requirement in run_response.active_requirements:
62        if requirement.needs_user_input:
63            input_schema: List[UserInputField] = requirement.user_input_schema  # type: ignore
64
65            for field in input_schema:
66                # Get user input for each field in the schema
67                field_type = field.field_type  # type: ignore
68                field_description = field.description  # type: ignore
69
70                # Display field information to the user
71                print(f"\nField: {field.name}")  # type: ignore
72                print(f"Description: {field_description}")
73                print(f"Type: {field_type}")
74
75                # Get user input
76                if field.value is None:  # type: ignore
77                    user_value = input(f"Please enter a value for {field.name}: ")  # type: ignore
78                else:
79                    print(f"Value: {field.value}")  # type: ignore
80                    user_value = field.value  # type: ignore
81
82                # Update the field value
83                field.value = user_value  # type: ignore
84
85    run_response = agent.continue_run(
86        run_id=run_response.run_id,
87        requirements=run_response.requirements,
88    )
89    if not run_response.is_paused:
90        pprint.pprint_run_response(run_response)
91        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!

Understanding the get_user_input Tool

When your agent calls the get_user_input tool, it provides a list of fields using this format:

1{
2    "field_name": "subject",      # The field identifier
3    "field_type": "str",           # Python type (str, int, float, bool, list, dict, etc.)
4    "field_description": "The subject of the email"  # Helpful description for the user
5}

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:

1[
2    {"field_name": "to_address", "field_type": "str", "field_description": "The email address to send to"},
3    {"field_name": "subject", "field_type": "str", "field_description": "The subject line for the email"}
4]

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, because the agent might request input multiple times:

1run_response = agent.run("Send an email and schedule a meeting")
2
3# First iteration: Agent needs email details
4while run_response.is_paused:
5    for requirement in run_response.requirements:
6        if requirement.needs_user_input:
7            for field in requirement.user_input_schema:
8                if field.value is None:
9                    field.value = input(f"Enter {field.name}: ")
10
11    run_response = agent.continue_run(run_id=run_response.run_id, requirements=run_response.requirements)
12    # Agent might pause again if it needs meeting details!

The agent could:

  1. First ask for email details
  2. Send the email
  3. Realize it needs meeting details
  4. Pause again to request those fields
  5. Complete the task

This multi-round capability makes the pattern extremely flexible.

Note

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.

Customizing Toolkit Behavior

The UserControlFlowTools toolkit comes with default instructions that guide the agent, but you can customize them:

1from kern.tools.user_control_flow import UserControlFlowTools
2
3# Custom instructions for your use case
4custom_instructions = """
5When you need user input:
61. Only request fields you absolutely need
72. Group related fields together
83. Provide clear, concise descriptions
94. Never request the same information twice
10"""
11
12agent = Agent(
13    model=OpenAIResponses(id="gpt-5.2"),
14    tools=[
15        EmailTools(),
16        UserControlFlowTools(
17            instructions=custom_instructions,
18            add_instructions=True
19        )
20    ],
21    markdown=True,
22)

You can also disable the tool entirely if needed:

1UserControlFlowTools(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:

1for field in tool.user_input_schema:
2    if field.value is None:
3        user_value = input(f"Please enter {field.name}: ")
4        field.value = user_value
5    else:
6        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

  1. Always use a while loop: The agent may need multiple rounds of input
  2. Check field values: Don't overwrite fields the agent has already filled
  3. Provide clear prompts: Use the field.description to help users understand what's needed
  4. Validate input: Add your own validation before setting field.value
  5. 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:

1from kern.agent import Agent
2from kern.models.openai import OpenAIResponses
3from kern.tools.user_control_flow import UserControlFlowTools
4
5agent = Agent(
6    model=OpenAIResponses(id="gpt-5.2"),
7    tools=[EmailTools(), UserControlFlowTools()],
8    markdown=True,
9)
10
11run_response = await agent.arun("Send an email")
12
13while run_response.is_paused:
14    for requirement in run_response.active_requirements:
15        if requirement.needs_user_input:
16            for field in requirement.user_input_schema:
17                if field.value is None:
18                    field.value = input(f"Please enter {field.name}: ")
19
20    run_response = await agent.acontinue_run(run_id=run_response.run_id, requirements=run_response.requirements)

Streaming Support

Dynamic user input also works with streaming. The agent will emit events until it needs user input, then pause:

1run_response = agent.run("Send an email", stream=True)
2
3for run_event in run_response:
4    if run_event.is_paused:
5        for requirement in run_event.active_requirements:
6            if requirement.needs_user_input:
7                for field in requirement.user_input_schema:
8                    if field.value is None:
9                        field.value = input(f"Please enter {field.name}: ")
10
11        # Continue streaming
12        for continued_event in agent.continue_run(
13            run_id=run_event.run_id,
14            requirements=run_event.requirements,
15            stream=True
16        ):
17            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
Warning

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.

Usage Examples

comments

Agentic User Input

Let the agent dynamically request user input

Developer Resources