Execute commands using SSH protocol by python script with AsyncSSH library

ssh-commands-by-python-asyncssh.py

#!/usr/bin/env python3
#

# https://github.com/ronf/asyncssh

import argparse
import asyncio
import time

import asyncssh
import sys

async def run_command(username: str, password: str, host: str, command: str) -> int:
    """
    Connect to a remote host via SSH, execute the given command,
    print headers, stdout, stderr, and return the exit status.
    """
    try:
        async with asyncssh.connect(host, username=username, password=password) as conn:
            result = await conn.run(command)

    except (asyncssh.Error, OSError) as e:
        print(f"Error: {e}", file=sys.stderr)
        # Use 1 to indicate a connection/execution failure
        return 1

    # Print stdout under a clear header if present
    if result.stdout:
        print("=== STDOUT ===")
        print(result.stdout, end='')

    # Print stderr under a clear header if present
    if result.stderr:
        print("=== STDERR ===", file=sys.stderr)
        print(result.stderr, end='', file=sys.stderr)

    # Finally, print the exit status
    print(f"\nExit status: {result.exit_status}")
    return result.exit_status


def main():
    parser = argparse.ArgumentParser(
        prog='ssh_command_asyncssh.py',
        description='Execute a command via SSH using asyncssh'
    )
    parser.add_argument('-u', '--username', required=True, help='SSH username')
    parser.add_argument('-p', '--password', required=True, help='SSH password')
    parser.add_argument('-H', '--host', required=True, help='Remote host or IP')
    parser.add_argument('-c', '--command', required=True, help='Bash command to execute')

    args = parser.parse_args()

    start_time = time.perf_counter()

    # Run the coroutine and get the exit status
    exit_code = asyncio.run(
        run_command(
            username=args.username,
            password=args.password,
            host=args.host,
            command=args.command
        )
    )

    elapsed_time = time.perf_counter() - start_time
    print(f"\n=== ELAPSED TIME ===\n{elapsed_time:.4f} seconds")

    # Propagate the exit code from the remote command (or 1 on failure)
    sys.exit(exit_code)

if __name__ == '__main__':
    main()