Python’s pty module (Unix/macOS only) allocates pseudo-terminal pairs and forks child processes attached to them, enabling interactive-terminal simulation, ANSI escape capture, and password-aware process automation. import pty. openpty: master_fd, slave_fd = pty.openpty() — allocate a pty pair; child uses slave_fd, parent uses master_fd. fork: pid, fd = pty.fork() — fork; in the child (pid=0), stdin/stdout/stderr are the slave pty; in the parent, fd is the master. spawn: pty.spawn(argv) or pty.spawn(argv, master_read=fn, stdin_read=fn) — runs a command attached to the current terminal’s pty or a new one; blocks until completion. Master fd behavior: reads return bytes from child’s stdout; writes send bytes to child’s stdin; the kernel echoes input by default. Use os.read(master_fd, 1024) — non-blocking with select; os.write(master_fd, b"input\n"). tty.setraw(sys.stdin.fileno()) — disable line buffering for raw key capture. termios — read/write terminal attributes. select.select([master_fd], [], [], timeout) — poll for output. Windows: not available; use winpty (PyPI) or pywinpty. Claude Code generates interactive process wrappers, terminal recorders, pexpect-style automators, and SSH tunnel handlers.
CLAUDE.md for pty
## pty Stack
- Stdlib: import pty, os, select, tty, termios # Unix/macOS only
- Pair: master_fd, slave_fd = pty.openpty()
- Fork: pid, master_fd = pty.fork()
if pid == 0: os.execvp(cmd, args) # child
else: os.read(master_fd, 1024) # parent reads child output
- Spawn: pty.spawn(["ssh", "host"]) # interactive pass-through
- Read: select.select([master_fd], [], [], 0.1); data = os.read(master_fd, 1024)
- Write: os.write(master_fd, b"command\n")pty Pseudo-Terminal Pipeline
# app/ptyutil.py — spawn, capture, interact, record, expect-style automation
from __future__ import annotations
import errno
import io
import os
import platform
import select
import signal
import sys
import time
from dataclasses import dataclass, field
from typing import Callable
# pty is Unix only
_PTY_AVAILABLE = platform.system() != "Windows"
if _PTY_AVAILABLE:
import pty
import termios
import tty
# ─────────────────────────────────────────────────────────────────────────────
# 1. Basic low-level helpers
# ─────────────────────────────────────────────────────────────────────────────
def allocate_pty() -> tuple[int, int]:
"""
Allocate a pseudo-terminal pair.
Returns (master_fd, slave_fd).
The child process should use slave_fd; the parent uses master_fd.
Example:
master_fd, slave_fd = allocate_pty()
"""
if not _PTY_AVAILABLE:
raise OSError("pty is not available on Windows")
return pty.openpty()
def set_nonblocking(fd: int) -> None:
"""Set a file descriptor to non-blocking mode."""
import fcntl
flags = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
def read_available(fd: int, max_bytes: int = 4096, timeout: float = 0.1) -> bytes:
"""
Read available bytes from fd with a timeout.
Returns empty bytes if nothing available.
Example:
data = read_available(master_fd)
"""
r, _, _ = select.select([fd], [], [], timeout)
if not r:
return b""
try:
return os.read(fd, max_bytes)
except OSError as e:
if e.errno in (errno.EIO, errno.EBADF):
return b""
raise
def read_until_eof(fd: int, timeout: float = 30.0, chunk: int = 4096) -> bytes:
"""
Read from fd until EIO (child closed) or timeout.
Returns all accumulated bytes.
Example:
output = read_until_eof(master_fd, timeout=10)
"""
buf = io.BytesIO()
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
r, _, _ = select.select([fd], [], [], 0.1)
if r:
try:
data = os.read(fd, chunk)
if not data:
break
buf.write(data)
except OSError as e:
if e.errno == errno.EIO:
break
raise
return buf.getvalue()
# ─────────────────────────────────────────────────────────────────────────────
# 2. Process spawner with output capture
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class PtyResult:
output: bytes
exit_code: int
duration: float
def text(self, encoding: str = "utf-8", errors: str = "replace") -> str:
return self.output.decode(encoding, errors=errors)
def __str__(self) -> str:
return (f"exit={self.exit_code} "
f"bytes={len(self.output)} "
f"time={self.duration:.2f}s")
def run_in_pty(
args: list[str],
stdin_data: bytes | None = None,
timeout: float = 30.0,
env: dict | None = None,
cwd: str | None = None,
) -> PtyResult:
"""
Run a command in a pseudo-terminal and capture all output.
Useful for commands that behave differently without a tty (e.g. color output,
interactive prompts, terminal-aware pagers).
Example:
result = run_in_pty(["python3", "-c", "import sys; print(sys.stdout.isatty())"])
print(result.text()) # "True"
"""
if not _PTY_AVAILABLE:
raise OSError("pty not available on Windows")
t0 = time.monotonic()
master_fd, slave_fd = pty.openpty()
pid = os.fork()
if pid == 0:
# Child: replace stdio with slave pty
os.setsid()
import fcntl
fcntl.ioctl(slave_fd, termios.TIOCSCTTY, 0)
for fd in (0, 1, 2):
os.dup2(slave_fd, fd)
if slave_fd > 2:
os.close(slave_fd)
os.close(master_fd)
if cwd:
os.chdir(cwd)
if env is not None:
os.execvpe(args[0], args, env)
else:
os.execvp(args[0], args)
os._exit(127)
# Parent
os.close(slave_fd)
set_nonblocking(master_fd)
buf = io.BytesIO()
deadline = time.monotonic() + timeout
if stdin_data:
try:
os.write(master_fd, stdin_data)
except OSError:
pass
while True:
if time.monotonic() > deadline:
try:
os.kill(pid, signal.SIGKILL)
except ProcessLookupError:
pass
break
r, _, _ = select.select([master_fd], [], [], 0.1)
if r:
try:
data = os.read(master_fd, 4096)
if data:
buf.write(data)
continue
except OSError as e:
if e.errno == errno.EIO:
break
raise
# Check if child exited
result_pid, status = os.waitpid(pid, os.WNOHANG)
if result_pid == pid:
# Drain any remaining output
while True:
r2, _, _ = select.select([master_fd], [], [], 0.05)
if not r2:
break
try:
data = os.read(master_fd, 4096)
if data:
buf.write(data)
else:
break
except OSError:
break
os.close(master_fd)
exit_code = os.waitstatus_to_exitcode(status)
return PtyResult(output=buf.getvalue(), exit_code=exit_code,
duration=time.monotonic() - t0)
os.close(master_fd)
_, status = os.waitpid(pid, 0)
exit_code = os.waitstatus_to_exitcode(status)
return PtyResult(output=buf.getvalue(), exit_code=exit_code,
duration=time.monotonic() - t0)
# ─────────────────────────────────────────────────────────────────────────────
# 3. Terminal recorder (output capture with timestamps)
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class TerminalEvent:
ts: float # seconds since start
kind: str # "output" or "input"
data: bytes
@dataclass
class TerminalRecording:
events: list[TerminalEvent] = field(default_factory=list)
duration: float = 0.0
exit_code: int = -1
def full_output(self) -> bytes:
return b"".join(e.data for e in self.events if e.kind == "output")
def text(self, encoding: str = "utf-8") -> str:
return self.full_output().decode(encoding, errors="replace")
def __str__(self) -> str:
out_bytes = sum(len(e.data) for e in self.events if e.kind == "output")
return (f"events={len(self.events)} "
f"output_bytes={out_bytes} "
f"duration={self.duration:.2f}s "
f"exit={self.exit_code}")
def record_pty(
args: list[str],
inputs: list[tuple[float, bytes]] | None = None,
timeout: float = 30.0,
) -> TerminalRecording:
"""
Run a command in a pty, recording all output events with timestamps.
inputs: list of (delay_seconds, bytes_to_send) for scripted input.
Example:
rec = record_pty(["bash", "-i"], inputs=[(0.5, b"ls -1\n"), (1.0, b"exit\n")])
print(rec.text())
"""
if not _PTY_AVAILABLE:
raise OSError("pty not available on Windows")
recording = TerminalRecording()
t0 = time.monotonic()
master_fd, slave_fd = pty.openpty()
pid = os.fork()
if pid == 0:
os.setsid()
import fcntl
fcntl.ioctl(slave_fd, termios.TIOCSCTTY, 0)
for fd in (0, 1, 2):
os.dup2(slave_fd, fd)
if slave_fd > 2:
os.close(slave_fd)
os.close(master_fd)
os.execvp(args[0], args)
os._exit(127)
os.close(slave_fd)
set_nonblocking(master_fd)
pending_inputs = list(inputs or [])
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
now = time.monotonic() - t0
# Send scheduled inputs
while pending_inputs and now >= pending_inputs[0][0]:
delay, data = pending_inputs.pop(0)
try:
os.write(master_fd, data)
recording.events.append(TerminalEvent(ts=now, kind="input", data=data))
except OSError:
pass
r, _, _ = select.select([master_fd], [], [], 0.05)
if r:
try:
chunk = os.read(master_fd, 4096)
if chunk:
recording.events.append(
TerminalEvent(ts=time.monotonic() - t0, kind="output", data=chunk)
)
continue
except OSError as e:
if e.errno == errno.EIO:
break
raise
result_pid, status = os.waitpid(pid, os.WNOHANG)
if result_pid == pid:
recording.exit_code = os.waitstatus_to_exitcode(status)
break
os.close(master_fd)
try:
_, status = os.waitpid(pid, os.WNOHANG)
if status:
recording.exit_code = os.waitstatus_to_exitcode(status)
except ChildProcessError:
pass
recording.duration = time.monotonic() - t0
return recording
# ─────────────────────────────────────────────────────────────────────────────
# 4. Expect-style pattern waiter
# ─────────────────────────────────────────────────────────────────────────────
class PtySession:
"""
Interactive pty session with send/expect API.
Example:
with PtySession(["python3", "-i"]) as sess:
sess.expect(">>> ", timeout=3)
sess.send("1 + 1\n")
output = sess.expect(">>> ", timeout=3)
print(output) # "2\\n>>> "
"""
def __init__(self, args: list[str], timeout: float = 10.0) -> None:
if not _PTY_AVAILABLE:
raise OSError("pty not available on Windows")
self._args = args
self._timeout = timeout
self._master_fd: int = -1
self._pid: int = -1
self._buf = b""
def __enter__(self) -> "PtySession":
master_fd, slave_fd = pty.openpty()
pid = os.fork()
if pid == 0:
os.setsid()
import fcntl
fcntl.ioctl(slave_fd, termios.TIOCSCTTY, 0)
for fd in (0, 1, 2):
os.dup2(slave_fd, fd)
if slave_fd > 2:
os.close(slave_fd)
os.close(master_fd)
os.execvp(self._args[0], self._args)
os._exit(127)
os.close(slave_fd)
set_nonblocking(master_fd)
self._master_fd = master_fd
self._pid = pid
return self
def __exit__(self, *args) -> None:
try:
os.kill(self._pid, signal.SIGTERM)
except ProcessLookupError:
pass
try:
os.close(self._master_fd)
except OSError:
pass
try:
os.waitpid(self._pid, 0)
except ChildProcessError:
pass
def send(self, data: str | bytes) -> None:
"""Send data to the process stdin."""
if isinstance(data, str):
data = data.encode()
os.write(self._master_fd, data)
def expect(self, pattern: str | bytes, timeout: float | None = None) -> bytes:
"""
Wait until pattern appears in output. Returns accumulated output up to
and including the pattern. Raises TimeoutError on timeout.
"""
if isinstance(pattern, str):
pattern = pattern.encode()
deadline = time.monotonic() + (timeout or self._timeout)
while True:
if pattern in self._buf:
idx = self._buf.index(pattern) + len(pattern)
result = self._buf[:idx]
self._buf = self._buf[idx:]
return result
if time.monotonic() > deadline:
raise TimeoutError(f"Pattern {pattern!r} not found within timeout")
r, _, _ = select.select([self._master_fd], [], [], 0.05)
if r:
try:
chunk = os.read(self._master_fd, 4096)
self._buf += chunk
except OSError as e:
if e.errno == errno.EIO:
raise EOFError("Process closed pty") from e
raise
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
if not _PTY_AVAILABLE:
print("pty not available on Windows — skipping demo")
raise SystemExit(0)
print("=== pty demo ===")
# ── run_in_pty: verify isatty() ────────────────────────────────────────────
print("\n--- run_in_pty: isatty() check ---")
result = run_in_pty(
["python3", "-c", "import sys; print(sys.stdout.isatty())"],
timeout=10
)
print(f" {result}")
print(f" isatty output: {result.text().strip()!r}") # True when in pty
# ── run_in_pty: color output ───────────────────────────────────────────────
print("\n--- run_in_pty: ANSI color check (ls --color) ---")
result2 = run_in_pty(["ls", "--color=always", "/"], timeout=5)
print(f" {result2}")
has_ansi = b"\x1b[" in result2.output
print(f" ANSI escapes present: {has_ansi}")
# ── run_in_pty: echo command ───────────────────────────────────────────────
print("\n--- run_in_pty: echo ---")
r3 = run_in_pty(["echo", "Hello from pty!"], timeout=5)
print(f" output: {r3.text().strip()!r}")
# ── record_pty ────────────────────────────────────────────────────────────
print("\n--- record_pty: python3 -c ---")
rec = record_pty(
["python3", "-c", "print('recorded output'); import time; time.sleep(0.1)"],
timeout=5,
)
print(f" {rec}")
print(f" text: {rec.text().strip()!r}")
# ── PtySession: python REPL interaction ───────────────────────────────────
print("\n--- PtySession: python3 -i ---")
try:
with PtySession(["python3", "-i"], timeout=5) as sess:
# Wait for initial prompt
sess.expect(">>> ", timeout=5)
sess.send("2 + 3\n")
out = sess.expect(">>> ", timeout=5)
val = out.decode("utf-8", errors="replace").strip().split("\n")[0]
print(f" 2 + 3 = {val!r}")
sess.send("exit()\n")
except Exception as e:
print(f" PtySession demo: {e}")
print("\n=== done ===")For the subprocess alternative — subprocess.Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE) captures subprocess I/O through pipes without a terminal — use subprocess for the vast majority of process spawning needs where the child doesn’t check isatty() and where you don’t need ANSI escapes, interactive prompts, or a controlling terminal; use pty when the child process behaves differently without a tty (e.g. disables color output, buffers aggressively, or requires terminal-style echo), when automating interactive TUI programs, or when capturing terminal-aware output including ANSI escape sequences. For the pexpect alternative — pexpect (PyPI) provides a higher-level spawn/expect/sendline API built on top of pty with pattern-matching, timeout handling, and cross-platform Windows support via pexpect.popen_spawn — use pexpect when building robust interactive automation (SSH, FTP, passwd, database shells) where you need reliable pattern matching, \r\n handling, and tested edge cases; use pty directly when you need low-level control over the master fd, want to implement a terminal recorder or multiplexer, or want zero dependencies. The Claude Skills 360 bundle includes pty skill sets covering allocate_pty()/set_nonblocking()/read_available()/read_until_eof() low-level helpers, PtyResult with run_in_pty() capture-with-timeout runner, TerminalEvent/TerminalRecording with record_pty() timestamped recorder, and PtySession context-manager with send()/expect() pattern waiter. Start with the free tier to try pseudo-terminal patterns and pty pipeline code generation.