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(model, env, **kwargs)

Bases: App

Source code in src/minisweagent/agents/interactive_textual.py

def __init__(self, model, env, **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 = _TextualAgent(self, model=model, env=env, **kwargs)
    self._i_step = 0
    self.n_steps = 1
    self.input_container = SmartInputContainer(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 = "ExitStatusUnset"
    self.result: str = ""

    self._vscroll = VerticalScroll()

BINDINGS class-attribute instance-attribute

BINDINGS = [
    Binding(
        "right,l",
        "next_step",
        "Step++",
        tooltip="Show next step of the agent",
    ),
    Binding(
        "left,h",
        "previous_step",
        "Step--",
        tooltip="Show previous step of the agent",
    ),
    Binding(
        "0",
        "first_step",
        "Step=0",
        tooltip="Show first step of the agent",
        show=False,
    ),
    Binding(
        "$",
        "last_step",
        "Step=-1",
        tooltip="Show last step of the agent",
        show=False,
    ),
    Binding(
        "j,down", "scroll_down", "Scroll down", show=False
    ),
    Binding("k,up", "scroll_up", "Scroll up", show=False),
    Binding(
        "q,ctrl+q", "quit", "Quit", tooltip="Quit the agent"
    ),
    Binding(
        "y,ctrl+y",
        "yolo",
        "YOLO mode",
        tooltip="Switch to YOLO Mode (LM actions will execute immediately)",
    ),
    Binding(
        "c",
        "confirm",
        "CONFIRM mode",
        tooltip="Switch to Confirm Mode (LM proposes commands and you confirm/reject them)",
    ),
    Binding(
        "u,ctrl+u",
        "human",
        "HUMAN mode",
        tooltip="Switch to Human Mode (you can now type commands directly)",
    ),
    Binding(
        "f1,question_mark",
        "toggle_help_panel",
        "Help",
        tooltip="Show help",
    ),
]

agent_state instance-attribute

agent_state = 'UNINITIALIZED'

agent instance-attribute

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

n_steps instance-attribute

n_steps = 1

input_container instance-attribute

input_container = SmartInputContainer(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 = 'ExitStatusUnset'

result instance-attribute

result: str = ''

config property

config

i_step property writable

i_step: int

Current step index.

messages property

messages: list[dict]

model property

model

env property

env

run

run(task: str, **kwargs) -> tuple[str, str]

Source code in src/minisweagent/agents/interactive_textual.py

def run(self, task: str, **kwargs) -> tuple[str, str]:
    threading.Thread(target=lambda: self.agent.run(task, **kwargs), daemon=True).start()
    super().run()
    return self.exit_status, self.result

compose

compose() -> ComposeResult

Source code in src/minisweagent/agents/interactive_textual.py

def compose(self) -> ComposeResult:
    yield Header()
    with Container(id="main"):
        with self._vscroll:
            with Vertical(id="content"):
                pass
            yield self.input_container
    yield Footer()

on_mount

on_mount() -> None

Source code in src/minisweagent/agents/interactive_textual.py

def on_mount(self) -> None:
    self.agent_state = "RUNNING"
    self.update_content()
    self.set_interval(1 / 8, self._update_headers)

on_message_added

on_message_added() -> None

Source code in src/minisweagent/agents/interactive_textual.py

def on_message_added(self) -> None:
    auto_follow = self.i_step == self.n_steps - 1 and self._vscroll.scroll_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

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

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

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

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.input_container.pending_prompt is not None:
        self.agent_state = "AWAITING_INPUT"
    self.input_container.display = self.input_container.pending_prompt is not None and self.i_step == len(items) - 1
    if self.input_container.display:
        self.input_container.on_focus()

    self._update_headers()
    self.refresh()

get_system_commands

get_system_commands(
    screen: Screen,
) -> Iterable[SystemCommand]

Source code in src/minisweagent/agents/interactive_textual.py

def get_system_commands(self, screen: Screen) -> Iterable[SystemCommand]:
    # Add to palette
    yield from super().get_system_commands(screen)
    for binding in self.BINDINGS:
        description = f"{binding.description} (shortcut {' OR '.join(binding.key.split(','))})"  # type: ignore[attr-defined]
        yield SystemCommand(description, binding.tooltip, binding.action)  # type: ignore[attr-defined]

action_yolo

action_yolo()

Source code in src/minisweagent/agents/interactive_textual.py

def action_yolo(self):
    self.agent.config.mode = "yolo"
    if self.input_container.pending_prompt is not None:
        self.input_container._complete_input("")  # accept
    self.notify("YOLO mode enabled - LM actions will execute immediately")

action_human

action_human()

Source code in src/minisweagent/agents/interactive_textual.py

def action_human(self):
    if self.agent.config.mode == "confirm" and self.input_container.pending_prompt is not None:
        self.input_container._complete_input("User switched to manual mode, this command will be ignored")
    self.agent.config.mode = "human"
    self.notify("Human mode enabled - you can now type commands directly")

action_confirm

action_confirm()

Source code in src/minisweagent/agents/interactive_textual.py

def action_confirm(self):
    if self.agent.config.mode == "human" and self.input_container.pending_prompt is not None:
        self.input_container._complete_input("")  # just submit blank action
    self.agent.config.mode = "confirm"
    self.notify("Confirm mode enabled - LM proposes commands and you confirm/reject them")

action_next_step

action_next_step() -> None

Source code in src/minisweagent/agents/interactive_textual.py

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

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

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

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

def action_scroll_down(self) -> None:
    self._vscroll.scroll_to(y=self._vscroll.scroll_target_y + 15)

action_scroll_up

action_scroll_up() -> None

Source code in src/minisweagent/agents/interactive_textual.py

def action_scroll_up(self) -> None:
    self._vscroll.scroll_to(y=self._vscroll.scroll_target_y - 15)

action_toggle_help_panel

action_toggle_help_panel() -> None

Source code in src/minisweagent/agents/interactive_textual.py

def action_toggle_help_panel(self) -> None:
    if self.query("HelpPanel"):
        self.action_hide_help_panel()
    else:
        self.action_show_help_panel()

minisweagent.agents.interactive_textual.TextualAgentConfig dataclass

TextualAgentConfig(
    system_template: str = "You are a helpful assistant that can do anything.",
    instance_template: str = "Your task: {{task}}. Please reply with a single shell command in triple backticks. To finish, the first line of the output of the shell command must be 'COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT'.",
    timeout_template: str = "The last command <command>{{action['action']}}</command> timed out and has been killed.\nThe output of the command was:\n <output>\n{{output}}\n</output>\nPlease try another command and make sure to avoid those requiring interactive input.",
    format_error_template: str = "Please always provide EXACTLY ONE action in triple backticks.",
    action_observation_template: str = "Observation: {{output}}",
    step_limit: int = 0,
    cost_limit: float = 3.0,
    mode: Literal["confirm", "yolo"] = "confirm",
    whitelist_actions: list[str] = list(),
    confirm_exit: bool = True,
)

Bases: AgentConfig

mode class-attribute instance-attribute

mode: Literal['confirm', 'yolo'] = 'confirm'

Mode for action execution: 'confirm' requires user confirmation, 'yolo' executes immediately.

whitelist_actions class-attribute instance-attribute

whitelist_actions: list[str] = field(default_factory=list)

Never confirm actions that match these regular expressions.

confirm_exit class-attribute instance-attribute

confirm_exit: bool = True

If the agent wants to finish, do we ask for confirmation from user?

minisweagent.agents.interactive_textual.SmartInputContainer

SmartInputContainer(app: TextualAgent)

Bases: Container

Smart input container supporting single-line and multi-line input modes.

Source code in src/minisweagent/agents/interactive_textual.py

def __init__(self, app: "TextualAgent"):
    """Smart input container supporting single-line and multi-line input modes."""
    super().__init__(classes="smart-input-container")
    self._app = app
    self._multiline_mode = False
    self.can_focus = True
    self.display = False

    self.pending_prompt: str | None = None
    self._input_event = threading.Event()
    self._input_result: str | None = None

    self._header_display = Static(id="input-header-display", classes="message-header input-request-header")
    self._hint_text = Static(classes="hint-text")
    self._single_input = Input(placeholder="Type your input...")
    self._multi_input = TextArea(show_line_numbers=False, classes="multi-input")
    self._input_elements_container = Vertical(
        self._header_display,
        self._hint_text,
        self._single_input,
        self._multi_input,
        classes="message-container",
    )

can_focus instance-attribute

can_focus = True

display instance-attribute

display = False

pending_prompt instance-attribute

pending_prompt: str | None = None

compose

compose() -> ComposeResult

Source code in src/minisweagent/agents/interactive_textual.py

def compose(self) -> ComposeResult:
    yield self._input_elements_container

on_mount

on_mount() -> None

Initialize the widget state.

Source code in src/minisweagent/agents/interactive_textual.py

def on_mount(self) -> None:
    """Initialize the widget state."""
    self._multi_input.display = False
    self._update_mode_display()

on_focus

on_focus() -> None

Called when the container gains focus.

Source code in src/minisweagent/agents/interactive_textual.py

def on_focus(self) -> None:
    """Called when the container gains focus."""
    if self._multiline_mode:
        self._multi_input.focus()
    else:
        self._single_input.focus()

request_input

request_input(prompt: str) -> str

Request input from user. Returns input text (empty string if confirmed without reason).

Source code in src/minisweagent/agents/interactive_textual.py

def request_input(self, prompt: str) -> str:
    """Request input from user. Returns input text (empty string if confirmed without reason)."""
    self._input_event.clear()
    self._input_result = None
    self.pending_prompt = prompt
    self._header_display.update(prompt)
    self._update_mode_display()
    self._app.call_from_thread(self._app.update_content)
    self._input_event.wait()
    return self._input_result or ""

action_toggle_mode

action_toggle_mode() -> None

Switch from single-line to multi-line mode (one-way only).

Source code in src/minisweagent/agents/interactive_textual.py

def action_toggle_mode(self) -> None:
    """Switch from single-line to multi-line mode (one-way only)."""
    if self.pending_prompt is None or self._multiline_mode:
        return

    self._multiline_mode = True
    self._update_mode_display()
    self.on_focus()

on_input_submitted

on_input_submitted(event: Submitted) -> None

Handle single-line input submission.

Source code in src/minisweagent/agents/interactive_textual.py

def on_input_submitted(self, event: Input.Submitted) -> None:
    """Handle single-line input submission."""
    if not self._multiline_mode:
        text = event.input.value.strip()
        self._complete_input(text)

on_key

on_key(event: Key) -> None

Handle key events.

Source code in src/minisweagent/agents/interactive_textual.py

def on_key(self, event: Key) -> None:
    """Handle key events."""
    if event.key == "ctrl+t" and not self._multiline_mode:
        event.prevent_default()
        self.action_toggle_mode()
        return

    if self._multiline_mode and event.key == "ctrl+d":
        event.prevent_default()
        self._complete_input(self._multi_input.text.strip())
        return

    if event.key == "escape":
        event.prevent_default()
        self.can_focus = False
        self._app.set_focus(None)
        return

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

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

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-ReX SWE-smith SWE-bench sb-cli