pexpect automates interactive CLI programs by spawning child processes and sending/expecting text. pip install pexpect. Basic: import pexpect; child = pexpect.spawn("ssh user@host"); child.expect("password:"); child.sendline("secret"); child.expect(pexpect.EOF). One-shot: out = pexpect.run("ls -la"). pexpect.run("cmd", timeout=30). Spawn: child = pexpect.spawn("python3", encoding="utf-8"). Send: child.sendline("print('hi')"). Expect: child.expect(">>> ") waits for prompt. Match list: idx = child.expect(["Password:", "denied", pexpect.EOF]). Timeout: child.expect("prompt", timeout=30). pexpect.TIMEOUT. pexpect.EOF. Before: child.before — text before the match. After: child.after — matched text. Logging: child.logfile = sys.stdout. child.logfile_read = fh. FTP: ftp = pexpect.spawn("ftp host"); ftp.expect("Name:"); ftp.sendline(user). SSH key: child.expect("Are you sure"); child.sendline("yes"). pxssh: from pexpect import pxssh; s = pxssh.pxssh(); s.login(host, user, pw); s.sendline("ls"); s.prompt(); print(s.before). s.logout(). Interact: child.interact() — hand off terminal to user. Control: child.sendcontrol("c"). Kill: child.terminate(). child.isalive(). Winsize: child.setwinsize(50, 200). Delimiters: child.expect_exact(">>>") vs regex. child.expect(re.compile(r"\$\s*$")). Claude Code generates pexpect automation scripts for SSH, database CLIs, and interactive program testing.
CLAUDE.md for pexpect
## pexpect Stack
- Version: pexpect >= 4.9 | pip install pexpect
- Spawn: child = pexpect.spawn("cmd", encoding="utf-8", timeout=30)
- Expect: child.expect("pattern") | child.expect(["opt1","opt2",TIMEOUT,EOF])
- Send: child.sendline("text") | child.send("no-newline") | child.sendcontrol("c")
- Match: child.before (text before match) | child.after (matched text)
- pxssh: pxssh.pxssh(); s.login(host, user, pw); s.sendline("cmd"); s.prompt()
- Log: child.logfile = sys.stdout — mirror all I/O for debuggingpexpect Automation Pipeline
# app/automation.py — pexpect spawn, expect/send, SSH via pxssh, and test helpers
from __future__ import annotations
import re
import sys
import time
from pathlib import Path
from typing import Any
import pexpect
from pexpect import EOF, TIMEOUT
# ─────────────────────────────────────────────────────────────────────────────
# 1. Core spawn helpers
# ─────────────────────────────────────────────────────────────────────────────
def spawn(
command: str,
args: list[str] | None = None,
timeout: int = 30,
encoding: str = "utf-8",
echo: bool = False,
log_stdout: bool = False,
) -> pexpect.spawn:
"""
Spawn a child process.
log_stdout=True mirrors all output to sys.stdout for debugging.
"""
if args:
child = pexpect.spawn(command, args, timeout=timeout, encoding=encoding)
else:
child = pexpect.spawn(command, timeout=timeout, encoding=encoding)
child.echo = echo
if log_stdout:
child.logfile_read = sys.stdout
return child
def run_command(
command: str,
timeout: int = 30,
encoding: str = "utf-8",
env: dict | None = None,
) -> tuple[str, int]:
"""
Run a command and return (output, exit_code) using pexpect.run().
"""
events = {
r"(?i)password:": lambda m: "secret\n",
r"\(yes/no\)": lambda m: "yes\n",
}
output, status = pexpect.run(
command,
timeout=timeout,
encoding=encoding,
events=events,
env=env,
withexitstatus=True,
)
return output, status
# ─────────────────────────────────────────────────────────────────────────────
# 2. Conversation helpers
# ─────────────────────────────────────────────────────────────────────────────
class Conversation:
"""
Fluent wrapper around pexpect.spawn for multi-step interactions.
Usage:
with Conversation("python3") as c:
c.wait(">>> ").send("import os").wait(">>> ")
c.send("print(os.getcwd())").wait(">>> ")
result = c.child.before
"""
def __init__(
self,
command: str,
args: list[str] | None = None,
timeout: int = 30,
encoding: str = "utf-8",
):
self._command = command
self._args = args or []
self._timeout = timeout
self._encoding = encoding
self.child: pexpect.spawn | None = None
self._log: list[str] = []
def __enter__(self) -> "Conversation":
self.child = spawn(
self._command,
self._args,
timeout=self._timeout,
encoding=self._encoding,
)
return self
def __exit__(self, *_):
if self.child and self.child.isalive():
self.child.terminate(force=True)
def wait(self, pattern: str, timeout: int | None = None) -> "Conversation":
"""Wait for pattern; raises pexpect.TIMEOUT or pexpect.EOF on failure."""
t = timeout or self._timeout
self.child.expect(pattern, timeout=t)
return self
def wait_any(self, patterns: list[str], timeout: int | None = None) -> int:
"""Wait for any of the patterns. Returns matched index."""
t = timeout or self._timeout
return self.child.expect(patterns + [TIMEOUT, EOF], timeout=t)
def send(self, text: str, newline: bool = True) -> "Conversation":
"""Send text with optional newline."""
if newline:
self.child.sendline(text)
else:
self.child.send(text)
self._log.append(f"SEND: {text!r}")
return self
def ctrl(self, key: str) -> "Conversation":
"""Send a control character, e.g. ctrl("c") for Ctrl+C."""
self.child.sendcontrol(key)
return self
@property
def before(self) -> str:
"""Text captured before the most recent match."""
return self.child.before or ""
@property
def log(self) -> list[str]:
return list(self._log)
# ─────────────────────────────────────────────────────────────────────────────
# 3. SSH via pxssh
# ─────────────────────────────────────────────────────────────────────────────
def ssh_run(
host: str,
username: str,
password: str | None = None,
commands: list[str] | None = None,
timeout: int = 30,
port: int = 22,
) -> list[str]:
"""
Run commands over SSH using pexpect.pxssh.
Returns list of command outputs.
Requires: pip install pexpect
Usage:
outputs = ssh_run("host.example.com", "deploy", password="secret",
commands=["ls /srv", "systemctl status api"])
"""
from pexpect import pxssh
s = pxssh.pxssh(timeout=timeout)
s.login(host, username, password=password, port=port, auto_prompt_reset=False)
results = []
for cmd in (commands or []):
s.sendline(cmd)
s.prompt()
results.append(s.before.strip())
s.logout()
return results
def ssh_upload(
host: str,
username: str,
password: str,
local_path: str,
remote_path: str,
timeout: int = 60,
) -> bool:
"""
Upload a file via scp with pexpect handling the password prompt.
Returns True on success.
"""
cmd = f"scp {local_path} {username}@{host}:{remote_path}"
child = spawn(cmd, timeout=timeout)
idx = child.expect(["password:", "100%", TIMEOUT, EOF])
if idx == 0:
child.sendline(password)
idx2 = child.expect(["100%", TIMEOUT, EOF])
return idx2 == 0
return idx == 1
# ─────────────────────────────────────────────────────────────────────────────
# 4. Database CLI automation
# ─────────────────────────────────────────────────────────────────────────────
def mysql_query(
host: str,
user: str,
password: str,
database: str,
sql: str,
timeout: int = 30,
) -> str:
"""
Run a SQL query through the mysql CLI and return output.
"""
cmd = f"mysql -h {host} -u {user} -p{password} {database}"
child = spawn(cmd, timeout=timeout)
child.expect(r"mysql\s*>")
child.sendline(sql)
child.expect(r"mysql\s*>")
result = child.before.strip()
child.sendline("exit")
child.expect(EOF)
return result
def psql_query(
host: str,
user: str,
database: str,
sql: str,
port: int = 5432,
password: str | None = None,
timeout: int = 30,
) -> str:
"""Run a query through psql and return the output."""
cmd = f"psql -h {host} -p {port} -U {user} -d {database}"
child = spawn(cmd, timeout=timeout)
if password:
idx = child.expect(["Password", r"=# "])
if idx == 0:
child.sendline(password)
child.expect(r"=# ")
child.sendline(sql)
child.expect(r"=# ")
result = child.before.strip()
child.sendline(r"\q")
child.expect(EOF)
return result
# ─────────────────────────────────────────────────────────────────────────────
# 5. Python REPL automation
# ─────────────────────────────────────────────────────────────────────────────
class PythonREPL:
"""
Automate a Python REPL for testing interactive code.
Usage:
with PythonREPL() as repl:
repl.run("x = 42")
output = repl.run("print(x)")
assert "42" in output
"""
PROMPT = r">>> "
CONT = r"\.\.\. "
def __init__(self, timeout: int = 10):
self._timeout = timeout
self.child: pexpect.spawn | None = None
def __enter__(self) -> "PythonREPL":
self.child = spawn("python3", timeout=self._timeout)
self.child.expect(self.PROMPT)
return self
def __exit__(self, *_):
if self.child and self.child.isalive():
try:
self.child.sendline("exit()")
self.child.expect(EOF, timeout=3)
except Exception:
self.child.terminate(force=True)
def run(self, code: str, timeout: int | None = None) -> str:
"""
Send a line of code to the REPL and return its output.
"""
t = timeout or self._timeout
self.child.sendline(code)
self.child.expect(self.PROMPT, timeout=t)
return self.child.before.strip()
def run_block(self, code: str) -> str:
"""
Send a multi-line block (function / class / if) and return output.
Automatically sends blank line to close indented blocks.
"""
lines = code.strip().split("\n")
for line in lines:
self.child.sendline(line)
self.child.expect([self.CONT, self.PROMPT])
self.child.sendline("") # end block
self.child.expect(self.PROMPT)
return self.child.before.strip()
# ─────────────────────────────────────────────────────────────────────────────
# 6. Expect pattern utils
# ─────────────────────────────────────────────────────────────────────────────
def expect_or_timeout(
child: pexpect.spawn,
patterns: list[str],
timeout: int = 30,
on_timeout: str | None = None,
) -> int:
"""
Expect one of patterns; return index or raise RuntimeError on timeout.
on_timeout: error message to include in RuntimeError.
"""
idx = child.expect(patterns + [TIMEOUT, EOF], timeout=timeout)
if idx >= len(patterns):
msg = on_timeout or f"Timeout waiting for one of {patterns}"
raise TimeoutError(msg)
return idx
def capture_until(
child: pexpect.spawn,
end_pattern: str,
timeout: int = 30,
) -> str:
"""Capture all output until end_pattern matches."""
child.expect(end_pattern, timeout=timeout)
return child.before or ""
def extract_match(child: pexpect.spawn, pattern: str, group: int = 1) -> str | None:
"""
Run expect with a regex and extract a capture group.
Returns None if the pattern doesn't match before timeout.
"""
try:
child.expect(re.compile(pattern))
m = child.match
return m.group(group) if m else None
except (TIMEOUT, EOF):
return None
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== pexpect.run ===")
out = pexpect.run("echo Hello from pexpect", encoding="utf-8").strip()
print(f" Output: {out!r}")
print("\n=== Python REPL automation ===")
with PythonREPL() as repl:
repl.run("import math")
pi_out = repl.run("print(round(math.pi, 4))")
print(f" π ≈ {pi_out!r}")
list_out = repl.run("print([x**2 for x in range(5)])")
print(f" Squares: {list_out!r}")
print("\n=== Conversation wrapper ===")
with Conversation("python3") as c:
c.wait(r">>> ").send("x = 'hello world'").wait(r">>> ")
c.send("print(x.upper())").wait(r">>> ")
print(f" Output: {c.before!r}")
print("\n=== run_command ===")
output, code = run_command("ls /tmp", timeout=5)
print(f" Exit code: {code}")
print(f" First line: {output.splitlines()[0]!r}")For the paramiko alternative — paramiko provides a full SSH2 protocol implementation with SFTP, public-key auth, port forwarding, and SSH agent support; pexpect automates any interactive program (not just SSH) by treating its I/O as a stream, making it the right tool when you need to drive a CLI that has an interactive session model (database CLIs, FTP, legacy apps, test frameworks); use paramiko for production SSH automation, pexpect for anything that’s interactive-but-not-SSH-protocol. For the subprocess alternative — subprocess.run() captures stdout/stderr of non-interactive programs; pexpect handles programs that require interactive back-and-forth (expect a password prompt, send the password, read the result), which subprocess can’t do because the child process detects it’s not a TTY and may disable prompts or buffer differently. The Claude Skills 360 bundle includes pexpect skill sets covering pexpect.spawn() with timeout/encoding, expect()/sendline()/send()/sendcontrol(), before/after match capture, pexpect.run() one-shot execution, pexpect.TIMEOUT/EOF constants, Conversation fluent wrapper, PythonREPL automated REPL testing, ssh_run() via pxssh, ssh_upload() SCP automation, mysql_query()/psql_query() database CLI, expect_or_timeout() safe expect, and capture_until()/extract_match() helpers. Start with the free tier to try interactive CLI automation code generation.