Agentic User Input with Control Flow

1from typing import Any, Dict, List
2
3from kern.agent import Agent
4from kern.db.sqlite import SqliteDb
5from kern.models.openai import OpenAIResponses
6from kern.tools import Toolkit
7from kern.tools.function import UserInputField
8from kern.tools.user_control_flow import UserControlFlowTools
9from kern.utils import pprint
10
11
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) -> list[dict[str, str]]:
29        """Get all emails between the given dates.
30
31        Args:
32            date_from (str): The start date (in YYYY-MM-DD format).
33            date_to (str): The end date (in YYYY-MM-DD format).
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    db=SqliteDb(db_file="tmp/agentic_user_input.db"),
56)
57
58run_response = agent.run("Send an email with the body 'What is the weather in Tokyo?'")
59
60# We use a while loop to continue the running until the agent is satisfied with the user input
61while run_response.is_paused:
62    for requirement in run_response.active_requirements:
63        if requirement.needs_user_input:
64            input_schema: List[UserInputField] = requirement.user_input_schema  # type: ignore
65
66            for field in input_schema:
67                # Get user input for each field in the schema
68                field_type = field.field_type  # type: ignore
69                field_description = field.description  # type: ignore
70
71                # Display field information to the user
72                print(f"\nField: {field.name}")  # type: ignore
73                print(f"Description: {field_description}")
74                print(f"Type: {field_type}")
75
76                # Get user input
77                if field.value is None:  # type: ignore
78                    user_value = input(f"Please enter a value for {field.name}: ")  # type: ignore
79                else:
80                    print(f"Value: {field.value}")  # type: ignore
81                    user_value = field.value  # type: ignore
82
83                # Update the field value
84                field.value = user_value  # type: ignore
85
86    run_response = agent.continue_run(
87        run_id=run_response.run_id,
88        requirements=run_response.requirements,
89    )
90    if not run_response.is_paused:
91        pprint.pprint_run_response(run_response)
92        break
93
94
95run_response = agent.run("Get me all my emails")
96
97while run_response.is_paused:
98    for requirement in run_response.active_requirements:
99        if requirement.needs_user_input:
100            input_schema: Dict[str, Any] = requirement.user_input_schema  # type: ignore
101
102            for field in input_schema:
103                # Get user input for each field in the schema
104                field_type = field.field_type  # type: ignore
105                field_description = field.description  # type: ignore
106
107                # Display field information to the user
108                print(f"\nField: {field.name}")  # type: ignore
109                print(f"Description: {field_description}")
110                print(f"Type: {field_type}")
111
112                # Get user input
113                if field.value is None:  # type: ignore
114                    user_value = input(f"Please enter a value for {field.name}: ")  # type: ignore
115                else:
116                    print(f"Value: {field.value}")  # type: ignore
117                    user_value = field.value  # type: ignore
118
119                # Update the field value
120                field.value = user_value  # type: ignore
121
122    run_response = agent.continue_run(
123        run_id=run_response.run_id,
124        requirements=run_response.requirements,
125    )
126
127    if not run_response.is_paused:
128        pprint.pprint_run_response(run_response)
129        break

1uv pip install -U kern-ai openai

1python agentic_user_input.py