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

def __init__(self, app: "AgentApp", *args, **kwargs):
    """Connects the DefaultAgent to the TextualApp."""
    self.app = app
    super().__init__(*args, config_class=TextualAgentConfig, **kwargs)
    self._current_action_from_human = False

app instance-attribute

app = app

add_message

add_message(role: str, content: str)

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

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)

query

query() -> dict

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

def query(self) -> dict:
    if self.config.mode == "human":
        human_input = self.app.input_container.request_input("Enter your command:")
        self._current_action_from_human = True
        msg = {"content": f"\n```bash\n{human_input}\n```"}
        self.add_message("assistant", msg["content"])
        return msg
    self._current_action_from_human = False
    return super().query()

run

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

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

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.action_quit)
        print(traceback.format_exc())
        return "ERROR", result
    else:
        self.app.call_from_thread(self.app.on_agent_finished, exit_status, result)
    self.app.call_from_thread(self.app.action_quit)
    return exit_status, result

execute_action

execute_action(action: dict) -> dict

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

def execute_action(self, action: dict) -> dict:
    if self.config.mode == "human" and not self._current_action_from_human:  # threading, grrrrr
        raise NonTerminatingException("Command not executed because user switched to manual mode.")
    if (
        self.config.mode == "confirm"
        and action["action"].strip()
        and not any(re.match(r, action["action"]) for r in self.config.whitelist_actions)
    ):
        result = self.app.input_container.request_input("Press ENTER to confirm or provide rejection reason")
        if result:  # Non-empty string means rejection
            raise NonTerminatingException(f"Command not executed: {result}")
    return super().execute_action(action)

has_finished

has_finished(output: dict[str, str])

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

def has_finished(self, output: dict[str, str]):
    try:
        return super().has_finished(output)
    except Submitted as e:
        if self.config.confirm_exit:
            if new_task := self.app.input_container.request_input(
                "[bold green]Agent wants to finish.[/bold green] "
                "[green]Type a comment to give it a new task or press enter to quit.\n"
            ).strip():
                raise NonTerminatingException(f"The user added a new task: {new_task}")
        raise e

minisweagent.agents.interactive_textual.AgentApp

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

Bases: App

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

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.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 | None = None
    self.result: str | None = None

    self._vscroll = VerticalScroll()

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"),
    Binding("u", "human", "Switch to Human 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

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 | 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

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)
    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

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()

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)

minisweagent.agents.interactive_textual.SmartInputContainer

SmartInputContainer(app: AgentApp)

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: "AgentApp"):
    """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