Python subprocess and asyncio: Building Robust CLI Automation in 2026

Most Python automation scripts eventually need to run something outside Python. A shell command, a system utility, a compiled binary, or another script in a different language. The subprocess module handles this. When you need to run multiple external commands at the same time, asyncio gives you non-blocking execution. Together, they form the foundation of practical CLI automation in Python.

This is not a theoretical overview. It is a practical guide to the patterns that work, the mistakes that cause problems, and the code you will actually write when you need to automate real workflows.

The Basics: subprocess.run()

The subprocess module provides several ways to run external commands. For most cases, subprocess.run() is the right choice. It runs a command, waits for it to finish, and returns a CompletedProcess object with the return code, stdout, and stderr.

import subprocess

result = subprocess.run(
    ["ls", "-la", "/tmp"],
    capture_output=True,
    text=True
)

print(f"Return code: {result.returncode}")
print(f"Output: {result.stdout}")

The key parameters are capture_output (which captures both stdout and stderr), text (which returns strings instead of bytes), and timeout (which raises TimeoutExpired if the command takes too long). Always use a timeout for any command that might hang or run indefinitely.

The text parameter is important because without it, subprocess returns bytes objects. This means you need to decode them manually before using them as strings, which adds friction and introduces encoding errors if the subprocess outputs text in a different encoding than your Python script expects. Setting text=True handles the decoding for you using the system’s default encoding, which is usually UTF-8 on modern systems.

The return code tells you whether the command succeeded. A return code of zero means success on Unix-like systems. On Windows, zero also means success, but some commands use different conventions. Always check the return code before assuming the command worked, even if it did not raise an exception. Subprocess only raises exceptions for things like the command not being found or the timeout expiring, not for commands that run successfully but produce an error exit code.

try:
    result = subprocess.run(
        ["curl", "-s", "https://example.com"],
        capture_output=True,
        text=True,
        timeout=30
    )
except subprocess.TimeoutExpired:
    print("Command timed out after 30 seconds")

The timeout parameter accepts seconds as a float. For long-running processes, set it generously. For commands that should respond quickly, keep it tight. The timeout applies to the entire execution, not per-output, so a command that produces output slowly but steadily will not time out as long as it keeps producing.

Security: Why Shell=True Is Dangerous

The most common security mistake with subprocess is using shell=True when you do not need to. When shell=True, Python passes your command string to the system shell for interpretation. This means shell metacharacters like pipes, redirects, and semicolons are interpreted, which can lead to command injection if any part of your command comes from user input.

# Dangerous with user input
user_input = "file.txt; rm -rf /"
subprocess.run(f"cat {user_input}", shell=True)  # DANGER

# Safe: pass as a list
subprocess.run(["cat", user_input])  # No shell interpretation

When you pass arguments as a list, each element becomes a separate argument to the program. The shell is not involved, so metacharacters are treated as literal characters. This is safer and more predictable.

The exception is when you genuinely need shell features like wildcards, pipes, or environment variable expansion. In those cases, use shell=True but sanitize any user input thoroughly, or better yet, restructure your code to avoid shell features entirely.

The security risk is not theoretical. Command injection vulnerabilities are among the most common and most dangerous web application security issues. Even in internal automation scripts, shell injection can cause data loss, unauthorized access, or system compromise. The cost of using list arguments instead of shell strings is minimal, and the security benefit is substantial. There is no reason to take the risk.

# Shell features needed: use shlex.quote()
import shlex
safe_arg = shlex.quote(user_input)
subprocess.run(f"cat {safe_arg}", shell=True)

Streaming Output

capture_output=True buffers the entire output before returning. For commands that produce a lot of output or run for a long time, this wastes memory and prevents you from processing output in real time.

Use subprocess.Popen() when you need to read output as it is produced:

import subprocess

process = subprocess.Popen(
    ["ping", "-c", "5", "google.com"],
    stdout=subprocess.PIPE,
    text=True
)

for line in process.stdout:
    print(f"Received: {line.strip()}")

process.wait()

The for line in process.stdout loop reads one line at a time. This is memory-efficient and lets you process output as it arrives. The process.wait() call at the end ensures the process is properly cleaned up.

Streaming is particularly useful for long-running commands like file transfers, builds, or data processing tasks. Instead of waiting for the entire output to accumulate, you can display progress, filter for specific lines, or take action based on intermediate output. For example, you might watch a build process and stop it early if you see a specific error message.

For stderr, you can redirect it to stdout using stderr=subprocess.STDOUT, or read it separately. If you need both simultaneously, consider using asyncio, which handles concurrent reads without deadlocking.

One practical consideration: line buffering depends on the subprocess. Some programs buffer their output when they detect they are writing to a pipe instead of a terminal. If you are not seeing output as expected, the program may be buffering. Some programs provide flags to disable buffering, like Python’s -u flag or GCC’s -pipe option. Others do not, and you may need to work around the buffering by reading in fixed-size chunks instead of by line.

Running Commands Concurrently with asyncio

When you need to run multiple independent commands at the same time, asyncio gives you non-blocking execution. The asyncio.create_subprocess_exec() function starts a process without waiting for it to complete. You can then await multiple processes simultaneously.

import asyncio

async def run_commands():
    commands = [
        ["ping", "-c", "3", "google.com"],
        ["ping", "-c", "3", "cloudflare.com"],
        ["ping", "-c", "3", "github.com"],
    ]

    processes = []
    for cmd in commands:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        processes.append(proc)

    results = await asyncio.gather(
        *[p.communicate() for p in processes]
    )

    for cmd, (stdout, stderr) in zip(commands, results):
        print(f"--- {cmd[-1]} ---")
        print(stdout.decode())

asyncio.run(run_commands())

The critical detail here is that all processes are started before any of them are awaited. If you await each process sequentially, you lose the concurrency benefit. Start them all, then gather the results.

The asyncio approach works well when your processes are I/O-bound, meaning they spend most of their time waiting for external resources like network responses or file system operations. If your processes are CPU-bound, meaning they spend most of their time doing computation, asyncio does not help because Python’s Global Interpreter Lock prevents true parallel execution of Python code. However, external processes are separate from the Python interpreter, so they do run in parallel even with asyncio. The GIL only affects Python code, not subprocess calls.

Another advantage of asyncio over threading for subprocess management is that asyncio does not suffer from the same deadlocking issues that threads can cause when reading from multiple pipes. The asyncio event loop handles the concurrent reads cleanly, which is why asyncio is generally preferred over threading for subprocess-heavy automation.

Handling Multiple Processes Safely

The asyncio.gather() approach works well for a fixed number of processes. For more dynamic scenarios where you need to start and stop processes based on conditions, use asyncio.wait():

import asyncio

async def managed_processes():
    tasks = []
    for i in range(5):
        proc = await asyncio.create_subprocess_exec(
            "sleep", str(i + 1),
            stdout=asyncio.subprocess.PIPE
        )
        tasks.append(asyncio.create_task(proc.wait()))

    # Wait for the first 3 to complete
    done, pending = await asyncio.wait(
        tasks, return_when=asyncio.FIRST_COMPLETED
    )

    print(f"{len(done)} processes finished")
    print(f"{len(pending)} still running")

    # Cancel remaining
    for task in pending:
        task.cancel()

asyncio.run(managed_processes())

asyncio.wait() accepts a return_when parameter that controls when it returns: FIRST_COMPLETED, FIRST_EXCEPTION, or ALL_COMPLETED. This gives you fine-grained control over process lifecycle management.

Pipelines: Connecting Process Outputs

Sometimes you need the output of one command to feed into the input of another, like shell pipes. In Python, you can do this by connecting process stdout to stdin:

import asyncio

async def pipeline():
    grep = await asyncio.create_subprocess_exec(
        "grep", "error",
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE
    )

    find = await asyncio.create_subprocess_exec(
        "find", "/var/log", "-name", "*.log",
        stdout=grep.stdin
    )

    # Let find feed into grep
    await find.wait()
    grep.stdin.close()

    stdout, _ = await grep.communicate()
    print(stdout.decode())

asyncio.run(pipeline())

The pattern is: start the downstream process first, connect its stdin to the upstream process’s stdout, then start the upstream process. The asyncio event loop handles the data flow between them.

This pattern is directly analogous to shell pipes, but with more control. You can inspect data as it flows between processes, apply filters, or make decisions based on intermediate output. You can also handle errors in individual pipeline stages independently, which is difficult to do in a shell pipeline where a failure in one stage can silently affect downstream stages.

Error Handling and Retry Logic

Real-world automation needs to handle failures gracefully. The subprocess module does not retry failed commands automatically. You need to build that logic yourself.

import subprocess
import time

def run_with_retry(cmd, max_retries=3, timeout=30, backoff=2):
    for attempt in range(max_retries):
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=timeout
            )
            if result.returncode == 0:
                return result
            print(f"Attempt {attempt + 1} failed: {result.stderr.strip()}")
        except subprocess.TimeoutExpired:
            print(f"Attempt {attempt + 1} timed out")

        if attempt < max_retries - 1:
            wait_time = backoff ** attempt
            print(f"Retrying in {wait_time} seconds...")
            time.sleep(wait_time)

    raise RuntimeError(f"Command failed after {max_retries} attempts: {cmd}")

The backoff parameter implements exponential backoff, which is important for commands that fail due to temporary conditions like network issues or resource contention. Starting with a short wait and increasing it reduces load on whatever is causing the failure.

The retry pattern is essential for any automation that interacts with external systems. Network requests, API calls, database operations, and file system operations can all fail transiently. A script that gives up on the first failure is brittle. A script that retries with backoff is robust.

However, not all failures should be retried. A command that fails because a file does not exist will not succeed on retry. A command that fails because of a syntax error will not succeed on retry. The retry logic should be reserved for transient failures, which you can often identify by the exit code or the error message. Many commands use specific exit codes for different error types, and checking these codes lets you decide whether to retry or give up immediately.

For asyncio, the same pattern works but uses asyncio.sleep() instead of time.sleep():

import asyncio

async def run_with_retry_async(cmd, max_retries=3, timeout=30, backoff=2):
    for attempt in range(max_retries):
        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(), timeout=timeout
            )
            if proc.returncode == 0:
                return stdout.decode()
            print(f"Attempt {attempt + 1} failed: {stderr.decode().strip()}")
        except asyncio.TimeoutError:
            print(f"Attempt {attempt + 1} timed out")

        if attempt < max_retries - 1:
            await asyncio.sleep(backoff ** attempt)

    raise RuntimeError(f"Command failed after {max_retries} attempts: {cmd}")

Real-World Example: Backup Script

Here is a practical example that combines these patterns into a backup script that compresses directories, checks their integrity, and uploads them to remote storage:

import asyncio
import subprocess
from pathlib import Path

async def backup_directory(source, dest):
    """Compress a directory and verify the archive."""
    archive_name = f"{source.name}.tar.gz"
    archive_path = dest / archive_name

    # Compress
    proc = await asyncio.create_subprocess_exec(
        "tar", "-czf", str(archive_path), "-C", str(source.parent), source.name,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    _, stderr = await proc.communicate()
    if proc.returncode != 0:
        raise RuntimeError(f"Compression failed: {stderr.decode()}")

    # Verify
    verify = await asyncio.create_subprocess_exec(
        "tar", "-tzf", str(archive_path),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    await verify.communicate()
    if verify.returncode != 0:
        raise RuntimeError(f"Verification failed for {archive_path}")

    return archive_path

async def upload_archive(archive_path, remote):
    """Upload archive to remote storage."""
    proc = await asyncio.create_subprocess_exec(
        "rsync", "-avz", str(archive_path), remote,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await proc.communicate()
    if proc.returncode != 0:
        raise RuntimeError(f"Upload failed: {stderr.decode()}")
    return stdout.decode()

async def main():
    sources = [Path(d) for d in ["/home/user/projects", "/home/user/documents"]]
    dest = Path("/tmp/backups")
    dest.mkdir(exist_ok=True)

    backups = [backup_directory(s, dest) for s in sources]
    archives = await asyncio.gather(*backups)

    uploads = [upload_archive(a, "user@remote:/backups/") for a in archives]
    results = await asyncio.gather(*uploads)

    for archive, result in zip(archives, results):
        print(f"Uploaded {archive.name}: {result.strip().split(chr(10))[-1]}")

asyncio.run(main())

This script compresses two directories in parallel, verifies each archive, and uploads them. The asyncio.gather() calls run independent operations concurrently while keeping failures isolated to individual tasks.

Common Pitfalls

Deadlocking on stdout and stderr. If a process writes to both stdout and stderr and you try to read both synchronously, one buffer can fill up and block the process while you are reading the other. Use asyncio to read both concurrently, or redirect stderr to stdout with stderr=subprocess.STDOUT if you do not need them separated.

Forgetting to close file handles. When using Popen with PIPE, the stdout and stderr file objects need to be closed after reading. Using context managers or calling communicate() handles this automatically.

Ignoring return codes. A return code of zero means success. Any other value indicates an error. Always check returncode before assuming a command succeeded, even if it did not raise an exception.

Not cleaning up zombie processes. On Unix systems, child processes that have finished but whose parent has not called wait() become zombies. Always call wait() or communicate() on processes you create, and use try/finally blocks to ensure cleanup even when errors occur.

The subprocess and asyncio modules are not glamorous, but they are the backbone of Python automation. Mastering them means your scripts can interact with any tool on the system, run tasks in parallel, and handle the messy reality of external process execution without breaking.

The patterns covered here apply to a wide range of automation tasks: deploying code, processing data, managing infrastructure, running tests, and building CI pipelines. Once you understand how to run external commands, capture their output, handle errors, and run multiple commands concurrently, you can automate almost anything that has a command-line interface. And in practice, almost everything does.