Skip to content

TextualAgent

See also

  • This agent subclass builds on top of the default agent, make sure to read that first.
  • This class powers the mini command line tool, see usage for more details.

minisweagent.agents.interactive_textual.TextualAgent 

TextualAgent(app: AgentApp, *args, **kwargs)

Bases: DefaultAgent

Connects the DefaultAgent to the TextualApp.

Source code in src/minisweagent/agents/interactive_textual.py 
36
37
38
39
def __init__(self, app: "AgentApp", *args, **kwargs):
    """Connects the DefaultAgent to the TextualApp."""
    self.app = app
    super().__init__(*args, config_class=TextualAgentConfig, **kwargs)

app instance-attribute 

app = app

add_message 

add_message(role: str, content: str)
Source code in src/minisweagent/agents/interactive_textual.py 
41
42
43
44
def add_message(self, role: str, content: str):
    super().add_message(role, content)
    if self.app.agent_state != "UNINITIALIZED":
        self.app.call_from_thread(self.app.on_message_added)

run 

run(task: str) -> tuple[str, str]
Source code in src/minisweagent/agents/interactive_textual.py 
46
47
48
49
50
51
52
53
54
55
def run(self, task: str) -> tuple[str, str]:
    try:
        exit_status, result = super().run(task)
    except Exception as e:
        result = str(e)
        self.app.call_from_thread(self.app.on_agent_finished, "ERROR", result)
        return "ERROR", result
    else:
        self.app.call_from_thread(self.app.on_agent_finished, exit_status, result)
    return exit_status, result

execute_action 

execute_action(action: dict) -> dict
Source code in src/minisweagent/agents/interactive_textual.py 
57
58
59
60
61
62
63
def execute_action(self, action: dict) -> dict:
    if self.config.mode == "confirm" and not any(
        re.match(r, action["action"]) for r in self.config.whitelist_actions
    ):
        if result := self.app.confirmation_container.request_confirmation(action["action"]):
            raise NonTerminatingException(f"Command not executed: {result}")
    return super().execute_action(action)

minisweagent.agents.interactive_textual.AgentApp 

AgentApp(model, env, task: str, **kwargs)

Bases: App

Source code in src/minisweagent/agents/interactive_textual.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def __init__(self, model, env, task: str, **kwargs):
    css_path = os.environ.get("MSWEA_MINI_STYLE_PATH", str(Path(__file__).parent.parent / "config" / "mini.tcss"))
    self.__class__.CSS = Path(css_path).read_text()
    super().__init__()
    self.agent_state = "UNINITIALIZED"
    self.agent_task = task
    self.agent = TextualAgent(self, model=model, env=env, **kwargs)
    self._i_step = 0
    self.n_steps = 1
    self.confirmation_container = ConfirmationPromptContainer(self)
    self.log_handler = AddLogEmitCallback(lambda record: self.call_from_thread(self.on_log_message_emitted, record))
    logging.getLogger().addHandler(self.log_handler)
    self._spinner = Spinner("dots")
    self.exit_status: str | None = None
    self.result: str | None = None

BINDINGS class-attribute instance-attribute 

BINDINGS = [
    Binding("right,l", "next_step", "Step++"),
    Binding("left,h", "previous_step", "Step--"),
    Binding("0", "first_step", "Step=0"),
    Binding("$", "last_step", "Step=-1"),
    Binding("j,down", "scroll_down", "Scroll down"),
    Binding("k,up", "scroll_up", "Scroll up"),
    Binding("q", "quit", "Quit"),
    Binding("y", "yolo", "Switch to YOLO Mode"),
    Binding("c", "confirm", "Switch to Confirm Mode"),
]

agent_state instance-attribute 

agent_state = 'UNINITIALIZED'

agent_task instance-attribute 

agent_task = task

agent instance-attribute 

agent = TextualAgent(self, model=model, env=env, **kwargs)

n_steps instance-attribute 

n_steps = 1

confirmation_container instance-attribute 

confirmation_container = ConfirmationPromptContainer(self)

log_handler instance-attribute 

log_handler = AddLogEmitCallback(
    lambda record: call_from_thread(
        on_log_message_emitted, record
    )
)

exit_status instance-attribute 

exit_status: str | None = None

result instance-attribute 

result: str | None = None

i_step property writable 

i_step: int

Current step index.

compose 

compose() -> ComposeResult
Source code in src/minisweagent/agents/interactive_textual.py 
207
208
209
210
211
212
213
def compose(self) -> ComposeResult:
    yield Header()
    with Container(id="main"):
        with VerticalScroll():
            yield Vertical(id="content")
        yield self.confirmation_container
    yield Footer()

on_mount 

on_mount() -> None
Source code in src/minisweagent/agents/interactive_textual.py 
215
216
217
218
219
def on_mount(self) -> None:
    self.agent_state = "RUNNING"
    self.update_content()
    self.set_interval(1 / 8, self._update_headers)
    threading.Thread(target=lambda: self.agent.run(self.agent_task), daemon=True).start()

on_message_added 

on_message_added() -> None
Source code in src/minisweagent/agents/interactive_textual.py 
223
224
225
226
227
228
229
def on_message_added(self) -> None:
    vs = self.query_one(VerticalScroll)
    auto_follow = self.i_step == self.n_steps - 1 and vs.scroll_target_y <= 1
    self.n_steps = len(_messages_to_steps(self.agent.messages))
    self.update_content()
    if auto_follow:
        self.action_last_step()

on_log_message_emitted 

on_log_message_emitted(record: LogRecord) -> None

Handle log messages of warning level or higher by showing them as notifications.

Source code in src/minisweagent/agents/interactive_textual.py 
231
232
233
234
def on_log_message_emitted(self, record: logging.LogRecord) -> None:
    """Handle log messages of warning level or higher by showing them as notifications."""
    if record.levelno >= logging.WARNING:
        self.notify(f"[{record.levelname}] {record.getMessage()}", severity="warning")

on_unmount 

on_unmount() -> None

Clean up the log handler when the app shuts down.

Source code in src/minisweagent/agents/interactive_textual.py 
236
237
238
239
def on_unmount(self) -> None:
    """Clean up the log handler when the app shuts down."""
    if hasattr(self, "log_handler"):
        logging.getLogger().removeHandler(self.log_handler)

on_agent_finished 

on_agent_finished(exit_status: str, result: str)
Source code in src/minisweagent/agents/interactive_textual.py 
241
242
243
244
245
246
def on_agent_finished(self, exit_status: str, result: str):
    self.agent_state = "STOPPED"
    self.notify(f"Agent finished with status: {exit_status}")
    self.exit_status = exit_status
    self.result = result
    self.update_content()

update_content 

update_content() -> None
Source code in src/minisweagent/agents/interactive_textual.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def update_content(self) -> None:
    container = self.query_one("#content", Vertical)
    container.remove_children()
    items = _messages_to_steps(self.agent.messages)

    if not items:
        container.mount(Static("Waiting for agent to start..."))
        return

    for message in items[self.i_step]:
        if isinstance(message["content"], list):
            content_str = "\n".join([item["text"] for item in message["content"]])
        else:
            content_str = str(message["content"])
        message_container = Vertical(classes="message-container")
        container.mount(message_container)
        role = message["role"].replace("assistant", "mini-swe-agent")
        message_container.mount(Static(role.upper(), classes="message-header"))
        message_container.mount(Static(Text(content_str, no_wrap=False), classes="message-content"))

    if self.confirmation_container._pending_action is not None:
        self.agent_state = "AWAITING_CONFIRMATION"
    self.confirmation_container.display = (
        self.confirmation_container._pending_action is not None and self.i_step == len(items) - 1
    )
    if self.confirmation_container.display:
        self.confirmation_container.focus()

    self._update_headers()
    self.refresh()

action_yolo 

action_yolo()
Source code in src/minisweagent/agents/interactive_textual.py 
295
296
297
298
def action_yolo(self):
    self.agent.config.mode = "yolo"
    self.confirmation_container._complete_confirmation(None)
    self.notify("YOLO mode enabled - actions will execute immediately")

action_confirm 

action_confirm()
Source code in src/minisweagent/agents/interactive_textual.py 
300
301
302
def action_confirm(self):
    self.agent.config.mode = "confirm"
    self.notify("Confirm mode enabled - actions will require confirmation")

action_next_step 

action_next_step() -> None
Source code in src/minisweagent/agents/interactive_textual.py 
304
305
def action_next_step(self) -> None:
    self.i_step += 1

action_previous_step 

action_previous_step() -> None
Source code in src/minisweagent/agents/interactive_textual.py 
307
308
def action_previous_step(self) -> None:
    self.i_step -= 1

action_first_step 

action_first_step() -> None
Source code in src/minisweagent/agents/interactive_textual.py 
310
311
def action_first_step(self) -> None:
    self.i_step = 0

action_last_step 

action_last_step() -> None
Source code in src/minisweagent/agents/interactive_textual.py 
313
314
def action_last_step(self) -> None:
    self.i_step = self.n_steps - 1

action_scroll_down 

action_scroll_down() -> None
Source code in src/minisweagent/agents/interactive_textual.py 
316
317
318
def action_scroll_down(self) -> None:
    vs = self.query_one(VerticalScroll)
    vs.scroll_to(y=vs.scroll_target_y + 15)

action_scroll_up 

action_scroll_up() -> None
Source code in src/minisweagent/agents/interactive_textual.py 
320
321
322
def action_scroll_up(self) -> None:
    vs = self.query_one(VerticalScroll)
    vs.scroll_to(y=vs.scroll_target_y - 15)

minisweagent.agents.interactive_textual.ConfirmationPromptContainer 

ConfirmationPromptContainer(app: AgentApp)

Bases: Container

This class is responsible for handling the action execution confirmation.

Source code in src/minisweagent/agents/interactive_textual.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, app: "AgentApp"):
    """This class is responsible for handling the action execution confirmation."""
    super().__init__(id="confirmation-container")
    self._app = app
    self.rejecting = False
    self.can_focus = True
    self.display = False

    self._pending_action: str | None = None
    self._confirmation_event = threading.Event()
    self._confirmation_result: str | None = None

rejecting instance-attribute 

rejecting = False

can_focus instance-attribute 

can_focus = True

display instance-attribute 

display = False

compose 

compose() -> ComposeResult
Source code in src/minisweagent/agents/interactive_textual.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
def compose(self) -> ComposeResult:
    yield Static(
        "Press [bold]ENTER[/bold] to confirm action or [bold]BACKSPACE[/bold] to reject (or [bold]y[/bold] to toggle YOLO mode)",
        classes="confirmation-prompt",
    )
    yield TextArea(id="rejection-input")
    rejection_help = Static(
        "Press [bold]Ctrl+D[/bold] to submit rejection message",
        id="rejection-help",
        classes="rejection-help",
    )
    rejection_help.display = False
    yield rejection_help

request_confirmation 

request_confirmation(action: str) -> str | None

Request confirmation for an action. Returns rejection message or None.

Source code in src/minisweagent/agents/interactive_textual.py 
117
118
119
120
121
122
123
124
def request_confirmation(self, action: str) -> str | None:
    """Request confirmation for an action. Returns rejection message or None."""
    self._confirmation_event.clear()
    self._confirmation_result = None
    self._pending_action = action
    self._app.call_from_thread(self._app.update_content)
    self._confirmation_event.wait()
    return self._confirmation_result

on_key 

on_key(event: Key) -> None
Source code in src/minisweagent/agents/interactive_textual.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def on_key(self, event: Key) -> None:
    if self.rejecting and event.key == "ctrl+d":
        event.prevent_default()
        rejection_input = self.query_one("#rejection-input", TextArea)
        self._complete_confirmation(rejection_input.text)
        return
    if not self.rejecting:
        if event.key == "enter":
            event.prevent_default()
            self._complete_confirmation(None)
        elif event.key == "backspace":
            event.prevent_default()
            self.rejecting = True
            rejection_input = self.query_one("#rejection-input", TextArea)
            rejection_input.display = True
            rejection_input.focus()
            rejection_help = self.query_one("#rejection-help", Static)
            rejection_help.display = True

minisweagent.agents.interactive_textual.AddLogEmitCallback 

AddLogEmitCallback(callback)

Bases: Handler

Custom log handler that forwards messages via callback.

Source code in src/minisweagent/agents/interactive_textual.py 
67
68
69
70
def __init__(self, callback):
    """Custom log handler that forwards messages via callback."""
    super().__init__()
    self.callback = callback

callback instance-attribute 

callback = callback

emit 

emit(record: LogRecord)
Source code in src/minisweagent/agents/interactive_textual.py 
72
73
def emit(self, record: logging.LogRecord):
    self.callback(record)  # type: ignore[attr-defined]
bug_report Something broken/unclear?

Open an issue on GitHub!
help Open-ended discussions

Join our Slack!
Our projects
SWE-agent SWE-agent SWE-rex SWE-ReX SWE-smith SWE-smith SWE-bench SWE-bench sb-cli sb-cli