asyncssh

import asyncio import asyncssh import os async def run_client(): host = os.environ.get('SSH_HOST', 'localhost') port = int(os.environ.get('SSH_PORT', 22)) username = os.environ.get('SSH_USERNAME', 'testuser') password = os.environ.get('SSH_PASSWORD', '') # Or use client_keys=['/path/to/id_rsa'] try: # Connect to the SSH server conn = await asyncssh.connect(host, port=port, username=username, password=password) # Run a command and get the result result = await conn.run('echo "Hello from AsyncSSH!"', check=True) print(f"Command: {result.command}") print(f"Stdout: {result.stdout.strip()}") print(f"Stderr: {result.stderr.strip()}") print(f"Return code: {result.returncode}") # Close the connection (async with handles this implicitly if used for conn) conn.close() except (asyncssh.Error, OSError) as exc: print(f'SSH connection or command failed: {exc}') if __name__ == '__main__': # Note: For 'localhost' testing, you might need a local SSH server configured # to accept connections for the specified username/password or client_key. # Example to run: # SSH_HOST=your_ssh_server SSH_USERNAME=your_user SSH_PASSWORD=your_pass python your_script.py asyncio.run(run_client())