Working with stdin, stdout and Pipes

The difference between a script and a tool people build on is usually composition: whether your command can sit in the middle of a pipeline without corrupting it, hanging, or crashing when the downstream program exits early. All of that comes down to how you treat three streams.

TL;DR

• stdout is the result. Anything a pipe should carry goes there and nothing else does.
• stderr is narration. Progress, warnings, errors and prompts — nothing should parse it.
• Support - for reading stdin and writing stdout; it costs a few lines.
• Catch BrokenPipeError at the boundary and exit quietly — | head is not a crash.
• Check isatty before colour, progress bars and prompts.

Three streams, three jobs

Almost every pipeline bug in a CLI is something written to the wrong stream. The test is simple: if piping your command into another program should carry it, it belongs on stdout. Everything else — including the progress bar that makes the tool pleasant interactively — belongs on stderr, where it still reaches the user's terminal but never enters the pipeline.

In practice that means two output objects created once and imported everywhere:

# src/mytool/console.py
import sys
from rich.console import Console

out = Console(file=sys.stdout)        # results
err = Console(file=sys.stderr)        # progress, warnings, errors

Every print in the program then goes through one of them, and the question "which stream is this on" is answered at the call site rather than by grepping for stray print() calls later.

Reading input that may be piped

A tool that reads stdin unconditionally appears to hang when run with no arguments, because it is waiting for input a person at a prompt has no idea they are meant to type. Check first:

import sys
from pathlib import Path

def read_document(source: Path | None) -> str:
    """Read from SOURCE, or from stdin when SOURCE is '-' or omitted with data piped in."""
    if source is not None and str(source) != "-":
        return source.read_text(encoding="utf-8")

    if sys.stdin.isatty():
        raise typer.BadParameter("no input: pass a file, or pipe a document on stdin")

    return sys.stdin.read()

The - convention is worth honouring in both directions. mytool apply - reads the document from stdin; mytool export -o - writes results to stdout. Users of Unix tools expect it, and supporting it is what lets your command appear anywhere in a pipeline rather than only at the start.

One detail that bites across platforms: name the encoding. sys.stdin.read() uses the platform default, which differs between Linux, macOS and Windows and between locales. For text, reconfigure explicitly; for binary data, use the underlying buffer:

sys.stdin.reconfigure(encoding="utf-8", errors="strict")
raw_bytes = sys.stdin.buffer.read()          # when the input is not text at all

Output other programs can parse

The moment someone wants to script your tool, they will parse whatever you print. Giving them a real format instead means the human-readable output stops being an interface you have to keep stable.

@app.command()
def status(
    json_output: Annotated[bool, typer.Option("--json", help="Emit JSON on stdout.")] = False,
) -> None:
    rows = core.collect_status()
    if json_output:
        typer.echo(json.dumps({"version": 1, "data": [r.as_dict() for r in rows], "warnings": []}))
        return
    render_table(rows)

Use one envelope for every command. A tool where status --json returns a list and list --json returns an object forces consumers to write per-command parsers, which is exactly the friction the flag was meant to remove.

For long or streaming results, newline-delimited JSON is the better shape: one object per line, no enclosing array, so a consumer can start work before your command has finished — and your memory stays flat.

Surviving a closed pipe

mytool list | head -n 10 is an ordinary thing to type. When head has what it needs it exits and closes the pipe, and your next write raises BrokenPipeError. The default behaviour — a traceback, plus a second warning from the interpreter at shutdown — makes a completely normal shell idiom look like a crash.

def main() -> None:
    try:
        app()
    except BrokenPipeError:
        # Redirect stdout so the interpreter's own flush at shutdown cannot fail too.
        devnull = os.open(os.devnull, os.O_WRONLY)
        os.dup2(devnull, sys.stdout.fileno())
        sys.exit(0)

Exit 0: nothing went wrong. The downstream consumer decided it had enough, which is the pipeline working as designed. The broken pipe guide covers the platform differences and why the dup2 line is needed.

Behaving differently when nobody is watching

Five behaviours should change when output is not a terminal, and each is one isatty check. Rich and Click handle most of them automatically; what they cannot decide is whether a progress bar should become periodic plain lines or disappear entirely, and whether a prompt should fail or use a default.

Route the check through one helper so the answer is consistent and testable:

def interactive() -> bool:
    return sys.stdin.isatty() and sys.stderr.isatty() and not os.environ.get("CI")

Streaming rather than collecting

The shape of a command that produces many records decides how it behaves in a pipeline. Collecting everything and printing at the end is simpler to write and worse in three ways: memory grows with the input, the downstream program waits for the whole run, and a failure halfway through produces nothing at all.

# collects — memory grows, nothing appears until the end
def export(period: str) -> None:
    rows = list(core.collect(period))
    typer.echo(json.dumps([row.as_dict() for row in rows]))

# streams — flat memory, output starts immediately
def export(period: str) -> None:
    for row in core.collect(period):          # a generator
        typer.echo(json.dumps(row.as_dict()))

The streaming version needs core.collect to be a generator rather than a function returning a list, which is usually a small change and occasionally a design improvement in its own right — the logic layer stops deciding how much to hold in memory, and the caller decides instead.

Two details make streaming work properly in practice. Flush on a sensible boundary: Python buffers stdout when it is not a terminal, which is what you want for throughput, but it means a long-running command produces nothing for a while. If a downstream consumer needs records promptly, flush every few hundred lines rather than every line:

for index, row in enumerate(core.collect(period), start=1):
    typer.echo(json.dumps(row.as_dict()))
    if index % 500 == 0:
        sys.stdout.flush()

And report progress on stderr, where it does not enter the pipeline:

if index % 1000 == 0:
    err.print(f"[dim]{index} rows exported…[/]")

Composing with the shell, deliberately

A tool that reads stdin and writes stdout gains a whole vocabulary of uses you never wrote code for. It is worth designing for them explicitly, because a few small decisions decide whether these work:

mytool export --json | jq '.data[] | select(.size > 1e6)'      # filter
mytool export --json | mytool import -                          # round-trip
cat jobs.ndjson | mytool apply - --dry-run                      # bulk apply
mytool list --json | head -5                                    # sample
mytool export -o - | gzip > backup.json.gz                      # stream to storage
diff <(mytool export --json) <(ssh host mytool export --json)   # compare environments

Four properties make all six possible, and each has already appeared above: results on stdout, - supported in both directions, a stable JSON shape, and a clean exit when the downstream closes. The last example — process substitution comparing two environments — is the kind of use nobody plans for and everybody appreciates.

The round-trip case (export | import -) is worth testing explicitly, because it is both a genuinely useful operation and a strong check that your output format contains everything your input format needs.

Buffering, ordering and the surprises

Three behaviours around streams surprise people often enough to be worth stating.

stdout and stderr are buffered differently. stdout is line-buffered on a terminal and block- buffered when redirected; stderr is unbuffered or line-buffered. The visible consequence is that when both are redirected to the same file, the interleaving does not match the order you wrote them. If ordering matters — a log people read alongside results — send everything to one stream, or flush deliberately.

Redirection changes behaviour, not just destination. mytool status on a terminal and mytool status > out.txt produce different bytes if you are colouring output, sizing tables to the window, or drawing progress. That is correct and intended, but it means a bug report that says "the output is wrong in a file" is usually about detection rather than about formatting.

Exit code and output are separate channels. A command can print a perfectly good JSON document and still exit non-zero — a partial failure, say. Consumers should check the status before parsing, and your documentation should say what output accompanies each code. Putting a "status": "error" field in the JSON instead of using the exit code is a common mistake: shell scripts branch on the status, not on the payload.

Testing pipeline behaviour

All of this is testable in-process, which means it belongs in the ordinary suite rather than in a shell script somebody runs by hand.

def test_results_go_to_stdout_and_narration_to_stderr(cli):
    result = cli.invoke(app, ["export", "--json"])

    assert result.exit_code == 0
    json.loads(result.stdout)                      # stdout is parseable on its own
    assert "exported" in result.stderr             # progress went elsewhere

def test_reads_a_document_from_stdin(cli):
    result = cli.invoke(app, ["apply", "-"], input='{"replicas": 3}')

    assert result.exit_code == 0

def test_empty_stdin_is_a_clear_error(cli):
    result = cli.invoke(app, ["apply", "-"], input="")

    assert result.exit_code == 65
    assert "no input" in result.stderr

The first test is the important one and the one people leave out. json.loads(result.stdout) fails loudly the moment anything non-JSON is written to stdout — a stray print, a progress line, a warning that went to the wrong stream — which is exactly the regression that breaks every consumer at once.

For broken-pipe behaviour, an in-process test is awkward, so use a subprocess in one test:

def test_survives_a_closed_downstream():
    tool = subprocess.Popen(["mytool", "list"], stdout=subprocess.PIPE)
    head = subprocess.Popen(["head", "-n", "5"], stdin=tool.stdout, stdout=subprocess.DEVNULL)
    tool.stdout.close()
    head.wait()

    assert tool.wait() == 0        # not a crash, not a non-zero status

Designing a command that filters

The most composable shape a command can have is a filter: read records from stdin, transform them, write records to stdout. Tools built that way chain with each other and with everything else in the shell.

@app.command()
def filter_failed(
    source: Annotated[Path, typer.Argument(help="NDJSON input, or - for stdin.")] = Path("-"),
) -> None:
    """Read deployment records and emit only the failed ones."""
    for line in read_lines(source):
        record = json.loads(line)
        if not record.get("healthy", True):
            typer.echo(json.dumps(record))

def read_lines(source: Path) -> Iterator[str]:
    if str(source) == "-":
        if sys.stdin.isatty():
            raise typer.BadParameter("no input: pipe NDJSON on stdin or pass a file")
        yield from sys.stdin
    else:
        with source.open(encoding="utf-8") as handle:
            yield from handle

Four properties make it a good citizen. It defaults to stdin, so it works mid-pipeline with no arguments. It streams, so memory stays flat and output appears immediately. It emits the same shape it consumes, so it can be chained with itself. And it writes nothing but records to stdout, so the next command's parser never chokes.

The same shape covers a surprising range of commands: filtering, enriching, reformatting, validating. When a command does not fit it — because it needs the whole input to produce one answer, for instance — that is worth noticing rather than forcing, and the honest version reads everything and emits one document.

What to do when both a file and stdin are possible

Commands often accept a path and support stdin, which raises a small design question people answer inconsistently: what happens when neither is given?

Three defensible answers, and the one to avoid:

Read stdin when data is waiting, otherwise error. The most Unix-like: mytool apply with a pipe behaves like mytool apply -, and without one it fails immediately with a message naming the argument. This is the behaviour of most filters.

Require the argument, with - as the explicit stdin form. Slightly more verbose for users and much harder to get wrong, because there is no implicit mode. Good for destructive commands, where "it read something I did not mean to give it" is expensive.

Default to a conventional file (./mytool.toml, ./jobs.ndjson), documented in the help text. Convenient for a tool with an obvious project-local input.

What to avoid is blocking on stdin with no indication. A command that hangs at a prompt with no output is indistinguishable from a crash, and the user's next move is Ctrl-C — which, if you have not handled it cleanly, prints a traceback on top of the confusion.

Where to go next

• Reading piped input in Python CLIs — stdin detection, the - convention, encodings and streaming large inputs.
• Emitting JSON output for scripting — envelope design, NDJSON, versioning the shape and testing it.
• Handling broken pipe and SIGPIPE — why the traceback appears, and the boundary that removes it.
• Detecting a TTY and adapting output — colour, progress, prompts and the environment variables worth honouring.

Key takeaways

• Results go to stdout; everything else goes to stderr, without exception.
• Support - in both directions so the tool composes in either position.
• One JSON envelope across every command beats a shape per command.
• A closed downstream is a normal end, not an error — exit 0, quietly.
• One interactive() helper, consulted everywhere, keeps TTY behaviour consistent.

Frequently asked questions

Related

• Up: Advanced Input Parsing & User Experience
• Sideways: Interactive terminal UI with Rich
• Sideways: Error handling and exit codes for CLIs
• Related: Structured logging for CLI apps
• Related: CLI help output and documentation