Skip to content

Commit 16ac3bd

Browse files
committed
mcp(feat[pane_tools]): Add wait_for_text tool for terminal automation
why: Terminal automation requires waiting for specific output to appear (e.g., build completion, prompt return). Agents currently must poll `capture_pane` repeatedly, consuming tokens and turns. A dedicated tool saves both by encapsulating the poll loop server-side. Uses libtmux's existing `retry_until` infrastructure (8s default timeout, 50ms poll interval) and the same regex pattern matching as `search_panes`. Refs: - #651 what: - Add WaitForTextResult Pydantic model to models.py - Add wait_for_text() tool using retry_until internally - Returns structured result with found/timed_out/matched_lines/elapsed - Tagged TAG_READONLY with ANNOTATIONS_RO (read-only, idempotent) - Add parametrized WaitForTextFixture tests (found + timeout cases) - Add test_wait_for_text_invalid_regex for bad pattern handling Closes #651
1 parent 35e43eb commit 16ac3bd

3 files changed

Lines changed: 197 additions & 2 deletions

File tree

src/libtmux/mcp/models.py

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,3 +120,16 @@ class EnvironmentSetResult(BaseModel):
120120
name: str = Field(description="Variable name")
121121
value: str = Field(description="Value that was set")
122122
status: str = Field(description="Operation status")
123+
124+
125+
class WaitForTextResult(BaseModel):
126+
"""Result of waiting for text to appear in a pane."""
127+
128+
found: bool = Field(description="Whether the pattern was found before timeout")
129+
matched_lines: list[str] = Field(
130+
default_factory=list,
131+
description="Lines matching the pattern (empty if not found)",
132+
)
133+
pane_id: str = Field(description="Pane ID that was polled")
134+
elapsed_seconds: float = Field(description="Time spent waiting in seconds")
135+
timed_out: bool = Field(description="Whether the timeout was reached")

src/libtmux/mcp/tools/pane_tools.py

Lines changed: 105 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
_serialize_pane,
2121
handle_tool_errors,
2222
)
23-
from libtmux.mcp.models import PaneContentMatch, PaneInfo
23+
from libtmux.mcp.models import PaneContentMatch, PaneInfo, WaitForTextResult
2424

2525
if t.TYPE_CHECKING:
2626
from fastmcp import FastMCP
@@ -497,6 +497,107 @@ def search_panes(
497497
return matches
498498

499499

500+
@handle_tool_errors
501+
def wait_for_text(
502+
pattern: str,
503+
pane_id: str | None = None,
504+
session_name: str | None = None,
505+
session_id: str | None = None,
506+
window_id: str | None = None,
507+
timeout: float = 8.0,
508+
interval: float = 0.05,
509+
match_case: bool = False,
510+
content_start: int | None = None,
511+
content_end: int | None = None,
512+
socket_name: str | None = None,
513+
) -> WaitForTextResult:
514+
"""Wait for text to appear in a tmux pane.
515+
516+
Polls the pane content at regular intervals until the pattern is found
517+
or the timeout is reached. Use this instead of polling capture_pane
518+
manually — it saves agent tokens and turns.
519+
520+
Parameters
521+
----------
522+
pattern : str
523+
Text or regex pattern to wait for.
524+
pane_id : str, optional
525+
Pane ID (e.g. '%1').
526+
session_name : str, optional
527+
Session name for pane resolution.
528+
session_id : str, optional
529+
Session ID (e.g. '$1') for pane resolution.
530+
window_id : str, optional
531+
Window ID for pane resolution.
532+
timeout : float
533+
Maximum seconds to wait. Default 8.0.
534+
interval : float
535+
Seconds between polls. Default 0.05 (50ms).
536+
match_case : bool
537+
Whether to match case. Default False (case-insensitive).
538+
content_start : int, optional
539+
Start line for capture. Negative values reach into scrollback.
540+
content_end : int, optional
541+
End line for capture.
542+
socket_name : str, optional
543+
tmux socket name.
544+
545+
Returns
546+
-------
547+
WaitForTextResult
548+
Result with found status, matched lines, and timing info.
549+
"""
550+
import time
551+
552+
from fastmcp.exceptions import ToolError
553+
554+
from libtmux.test.retry import retry_until
555+
556+
flags = 0 if match_case else re.IGNORECASE
557+
try:
558+
compiled = re.compile(pattern, flags)
559+
except re.error as e:
560+
msg = f"Invalid regex pattern: {e}"
561+
raise ToolError(msg) from e
562+
563+
server = _get_server(socket_name=socket_name)
564+
pane = _resolve_pane(
565+
server,
566+
pane_id=pane_id,
567+
session_name=session_name,
568+
session_id=session_id,
569+
window_id=window_id,
570+
)
571+
572+
assert pane.pane_id is not None
573+
matched_lines: list[str] = []
574+
start_time = time.monotonic()
575+
576+
def _check() -> bool:
577+
lines = pane.capture_pane(start=content_start, end=content_end)
578+
hits = [line for line in lines if compiled.search(line)]
579+
if hits:
580+
matched_lines.extend(hits)
581+
return True
582+
return False
583+
584+
found = retry_until(
585+
_check,
586+
seconds=timeout,
587+
interval=interval,
588+
raises=False,
589+
)
590+
591+
elapsed = time.monotonic() - start_time
592+
return WaitForTextResult(
593+
found=found,
594+
matched_lines=matched_lines,
595+
pane_id=pane.pane_id,
596+
elapsed_seconds=round(elapsed, 3),
597+
timed_out=not found,
598+
)
599+
600+
500601
def register(mcp: FastMCP) -> None:
501602
"""Register pane-level tools with the MCP instance."""
502603
mcp.tool(title="Send Keys", annotations=ANNOTATIONS_CREATE, tags={TAG_MUTATING})(
@@ -525,3 +626,6 @@ def register(mcp: FastMCP) -> None:
525626
mcp.tool(title="Search Panes", annotations=ANNOTATIONS_RO, tags={TAG_READONLY})(
526627
search_panes
527628
)
629+
mcp.tool(title="Wait For Text", annotations=ANNOTATIONS_RO, tags={TAG_READONLY})(
630+
wait_for_text
631+
)

tests/mcp/test_pane_tools.py

Lines changed: 79 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
import pytest
88
from fastmcp.exceptions import ToolError
99

10-
from libtmux.mcp.models import PaneContentMatch
10+
from libtmux.mcp.models import PaneContentMatch, WaitForTextResult
1111
from libtmux.mcp.tools.pane_tools import (
1212
capture_pane,
1313
clear_pane,
@@ -17,6 +17,7 @@
1717
search_panes,
1818
send_keys,
1919
set_pane_title,
20+
wait_for_text,
2021
)
2122
from libtmux.test.retry import retry_until
2223

@@ -413,3 +414,80 @@ def test_search_panes_is_caller(
413414
match = next((r for r in result if r.pane_id == mcp_pane.pane_id), None)
414415
assert match is not None
415416
assert match.is_caller is expected_is_caller
417+
418+
419+
# ---------------------------------------------------------------------------
420+
# wait_for_text tests
421+
# ---------------------------------------------------------------------------
422+
423+
424+
class WaitForTextFixture(t.NamedTuple):
425+
"""Test fixture for wait_for_text."""
426+
427+
test_id: str
428+
command: str | None
429+
pattern: str
430+
timeout: float
431+
expected_found: bool
432+
433+
434+
WAIT_FOR_TEXT_FIXTURES: list[WaitForTextFixture] = [
435+
WaitForTextFixture(
436+
test_id="text_found",
437+
command="echo WAIT_MARKER_abc123",
438+
pattern="WAIT_MARKER_abc123",
439+
timeout=2.0,
440+
expected_found=True,
441+
),
442+
WaitForTextFixture(
443+
test_id="timeout_not_found",
444+
command=None,
445+
pattern="NEVER_EXISTS_xyz999",
446+
timeout=0.3,
447+
expected_found=False,
448+
),
449+
]
450+
451+
452+
@pytest.mark.parametrize(
453+
WaitForTextFixture._fields,
454+
WAIT_FOR_TEXT_FIXTURES,
455+
ids=[f.test_id for f in WAIT_FOR_TEXT_FIXTURES],
456+
)
457+
def test_wait_for_text(
458+
mcp_server: Server,
459+
mcp_pane: Pane,
460+
test_id: str,
461+
command: str | None,
462+
pattern: str,
463+
timeout: float,
464+
expected_found: bool,
465+
) -> None:
466+
"""wait_for_text polls pane content for a pattern."""
467+
if command is not None:
468+
mcp_pane.send_keys(command, enter=True)
469+
470+
result = wait_for_text(
471+
pattern=pattern,
472+
pane_id=mcp_pane.pane_id,
473+
timeout=timeout,
474+
socket_name=mcp_server.socket_name,
475+
)
476+
assert isinstance(result, WaitForTextResult)
477+
assert result.found is expected_found
478+
assert result.timed_out is (not expected_found)
479+
assert result.pane_id == mcp_pane.pane_id
480+
assert result.elapsed_seconds >= 0
481+
482+
if expected_found:
483+
assert len(result.matched_lines) >= 1
484+
485+
486+
def test_wait_for_text_invalid_regex(mcp_server: Server, mcp_pane: Pane) -> None:
487+
"""wait_for_text raises ToolError on invalid regex."""
488+
with pytest.raises(ToolError, match="Invalid regex pattern"):
489+
wait_for_text(
490+
pattern="[invalid",
491+
pane_id=mcp_pane.pane_id,
492+
socket_name=mcp_server.socket_name,
493+
)

0 commit comments

Comments
 (0)