pexpect-cli-automation

name

pexpect-cli-automation

description

Use when the agent needs to interact with interactive CLI programs — programs that prompt for input, ask yes/no questions, require passwords, or run REPL sessions. Provides patterns for spawn, expect, sendline, branch handling, timeout/EOF management, cleanup, error diagnosis, log sanitization, retry, graceful exit, and agent-to-agent delegation with call-chain safety. Triggers on tasks like "automate this CLI", "script this interactive program", "handle the password prompt", "drive this REPL", "delegate to a sub-agent", or when the agent encounters a program that pauses for user input.

Pexpect CLI Automation

When to use pexpect

Priority order:

1. Existing API / SDK — always best
2. Non-interactive command — use subprocess
3. Interactive terminal required — use pexpect

Pexpect is the right tool when a program outputs a prompt and blocks waiting for input:

Password:
→ program stops, waits
→ you type, press Enter
→ program continues

Line-buffered vs raw-mode TUI

Raw mode does NOT make pexpect impossible — it raises the difficulty. The boundary is:

Ordinary line-buffered CLI
→ pexpect.sendline()

Raw-mode TUI (Ink, ncurses, etc.)
→ pexpect.send() + sendcontrol("m")   ← \r, not \n
→ May need ANSI parsing, bracketed paste, or tmux
→ Harder, but the PTY still works

Formal programming protocol (API, -p mode, SDK)
→ Always prefer this when available

Claude Code: three approaches, all valid

Approach 1 (production): -p mode via subprocess

claude -p --output-format json "your prompt"
claude -p --input-format stream-json --output-format stream-json --verbose

Structured JSON, context preserved across turns, lowest cost. See examples/claude-code.py.

Approach 2 (pexpect): drive the interactive TUI

child = pexpect.spawn(
    "claude",
    ["--ax-screen-reader", "--model", "sonnet", "Compute 23+19."],
    encoding="utf-8",
    echo=False,          # prevent PTY echo — Ink renders its own
    timeout=180,
)

# Wait for the answer
child.expect(r"(?<!\d)42(?!\d)")

# Submit a follow-up — use send() + sendcontrol("m"), NOT sendline()
child.send("Add one to the previous answer.")
child.sendcontrol("m")  # Ctrl+M = \r (real Enter in raw mode)

child.expect(r"(?<!\d)43(?!\d)")

Why sendline() fails on raw-mode TUIs: in raw mode, the kernel no longer converts \n to \r. sendline() appends os.linesep which is \n on Unix — the TUI sees a newline character (renders it) but never receives a carriage return (doesn't submit). A real Enter key sends \r (Ctrl+M). Use:

• child.send(text) + child.sendcontrol("m") for normal input
• child.send("\x1b[200~" + text + "\x1b[201~") + child.sendcontrol("m") for long/multi-line bracketed paste

Approach 3 (sdk): Claude Agent SDK

from claude_agent_sdk import query, ClaudeAgentOptions
async for msg in query(prompt="...", options=ClaudeAgentOptions(...)):
    ...

Use the SDK when integrating at the application level rather than the shell level.

Standard code structure

Every pexpect script must follow this skeleton:

import pexpect
import os
import signal
import sys


def run_interactive_cli() -> str:
    child = pexpect.spawn(
        "some-command",
        ["--option"],
        encoding="utf-8",
        timeout=30,
    )

    try:
        index = child.expect([
            "Ready>",
            pexpect.EOF,
            pexpect.TIMEOUT,
        ])

        if index == 0:
            child.sendline("do something")
            child.expect("Ready>")
            return child.before

        if index == 1:
            raise CLIEOFError(
                f"Program exited unexpectedly: {child.before!r}",
                child=child,
            )

        raise CLITimeoutError(
            f"Program did not become ready: {child.before!r}",
            child=child,
        )

    finally:
        shutdown(child, quit_cmd="exit")

Mandatory reliability rules

• Always set encoding explicitly (usually "utf-8")
• Always set timeout explicitly (seconds, not milliseconds)
• Always handle pexpect.EOF in every expect() call
• Always handle pexpect.TIMEOUT in every expect() call
• Prefer expect_exact() for literal strings (faster, no regex escaping)
• When using regex, be specific — never use bare $ or # as a prompt matcher
• Use typed exceptions (CLIError, CLIEOFError, CLITimeoutError) instead of bare RuntimeError
• Use three-stage shutdown: quit command → EOF → SIGTERM → SIGKILL
• Pass command and arguments as separate parameters: spawn("cmd", ["arg1", "arg2"])
• spawn() does NOT interpret |, >, * — use spawn("/bin/bash", ["-c", "cmd | grep foo"]) if you need shell features
• Do NOT use expect_exact() with pexpect.EOF/pexpect.TIMEOUT — use expect() for lists that include these sentinels
• Raw-mode TUIs: use send() + sendcontrol("m") instead of sendline(). In raw mode the kernel no longer converts \n → \r; sendline() sends \n (renders but doesn't submit), while a real Enter key sends \r (Ctrl+M)
• Raw-mode TUIs: set echo=False in spawn() to prevent PTY echo — the TUI renders its own input. If text still appears on screen after echo=False, it's the application rendering it, confirming input is reaching the app

Error diagnosis (v0.2)

Use typed exceptions that capture child process state:

class CLIError(Exception):
    def __init__(self, message, child=None):
        super().__init__(message)
        self.before = child.before if child else None
        self.exitstatus = child.exitstatus if child else None
        self.signalstatus = child.signalstatus if child else None

class CLITimeoutError(CLIError): ...
class CLIEOFError(CLIError): ...
class CLIMatchError(CLIError): ...

Always include child.before in error messages and check child.exitstatus/child.signalstatus for the cause.

Log sanitization (v0.2)

Use SanitizingLogger to filter secrets before they reach log output:

sanitizer = SanitizingLogger(sys.stdout)
sanitizer.register_secret(os.environ["PASSWORD"])
sanitizer.register_pattern(r"Bearer [A-Za-z0-9\-_.~+/]+=*", "[REDACTED_TOKEN]")

child.logfile_read = sanitizer
child.logfile_send = None  # NEVER log secrets we send

Never log passwords, API keys, or verification codes sent to the CLI.

Retry (v0.2)

Retry on transient failures (timeout, connection refused). Do NOT retry on auth failure, permission denied, or command not found:

for attempt in range(1, max_retries + 1):
    try:
        return do_interaction()
    except pexpect.TIMEOUT:
        if attempt < max_retries:
            time.sleep(backoff ** attempt)
        else:
            raise

Full-session retry spawns a new child each attempt. Single-step retry reuses the same session.

Graceful exit (v0.2)

Never jump straight to close(force=True). Use three-stage shutdown:

1. sendline("quit") → wait for EOF
2. sendcontrol("d")  → wait for EOF
3. kill(SIGTERM)     → wait for EOF
4. close(force=True) → last resort

Use the cli_session context manager to guarantee cleanup:

with cli_session("some-daemon", ["--interactive"], quit_cmd="quit") as child:
    child.expect_exact("Ready>")
    child.sendline("process data.txt")
    child.expect_exact("Done")
    result = child.before
# No manual close() needed

Agent call chain protocol (v0.3)

When spawning one agent from another via pexpect, pass call chain context through environment variables to prevent infinite recursion:

ctx = AgentCallContext(agent_id="controller-01")  # depth=0, max_depth=1

child = spawn_agent(
    sys.executable,
    ["worker_agent.py"],
    ctx,           # passes AGENT_DEPTH=1, AGENT_ROLE=worker via env
    task_id,       # unique task ID for dedup
)

Depth limits

Depth 0 (controller): can spawn workers      → depth < max_depth
Depth 1 (worker):     CANNOT spawn, CAN execute → depth == max_depth
Depth 2+:             impossible by protocol → depth > max_depth, blocked

Loop prevention checklist

Before spawning any sub-agent:

• depth < max_depth — haven't reached the limit
• role != "worker" — workers can't spawn
• task_id is not in the completed TaskRegistry
• The worker's prompt includes a boundary declaration ("you cannot delegate")
• The child runs with a timeout
• Save a Transcript of every session for debugging

Worker boundary declaration

Every worker agent's prompt must include:

You are a worker agent. Your ONLY job is to execute the task given to you
and return the result. You CANNOT spawn sub-agents or delegate work.

State thinking

Simple interactions can be linear:

READY → SEND → COMPLETED

When there are 3+ branches, explicitly track state:

STARTING
├── FIRST_RUN_CONFIRMATION
├── LOGIN_REQUIRED
├── READY
├── FAILED
└── TIMEOUT

Name states clearly in code (variables, comments, or enum) so the decision flow is visible.

Reference docs

• references/core-api.md — complete pexpect API surface
• references/patterns.md — common interaction patterns
• references/error-diagnosis.md — structured exceptions and diagnostic snapshots (v0.2)
• references/log-sanitization.md — SanitizingLogger and secret filtering (v0.2)
• references/retry-patterns.md — retry strategies and backoff (v0.2)
• references/graceful-exit.md — three-stage shutdown and context manager (v0.2)
• references/agent-protocol.md — agent call chain protocol, depth limits, dedup (v0.3)
• references/agent-transcript.md — structured session transcript format (v0.3)
• references/debugging.md — troubleshooting techniques
• references/security.md — security considerations

Examples

• examples/simple-prompt.py — basic spawn, expect, sendline
• examples/multi-branch.py — login with multiple outcomes
• examples/repl-session.py — Python REPL interaction
• examples/error-diagnosis.py — structured exceptions and diagnostic snapshots (v0.2)
• examples/reliability-patterns.py — sanitization + retry + graceful exit combined (v0.2)
• examples/agent-cli.py — agent delegation with call chain safety, transcript (v0.3)
• examples/claude-code.py — correct way to drive Claude Code: -p --output-format json via subprocess, NOT pexpect

Testing

• tests/fake-cli.py — deterministic fake CLI (12 modes: simple, login, timeout, crash, confirm, password, repl, agent, agent-protocol, flaky, hang-on-exit, slow-start)
• tests/test-examples.py — 21 tests covering v0.1 + v0.2 + v0.3