如何用asyncssh实现类似Paramiko Channel的交互式SSH Shell？

你当前的AsyncSSH代码仅完成了SSH连接和读取一行欢迎信息，缺少交互式Shell必需的PTY申请以及双向IO处理，这就是无法发送数据的原因。以下是实现交互式功能的完整方案，效果和你期望的一致——在控制台显示Shell提示符并响应标准输入：

完整实现代码 #

import asyncio
import asyncssh
import sys

async def handle_ssh_io(stdin, stdout, stderr):
    # 将SSH输出实时打印到本地终端
    async def read_ssh_output():
        # 处理stdout
        while True:
            data = await stdout.read(4096)
            if not data:
                break
            sys.stdout.write(data.decode())
            sys.stdout.flush()
        
        # 处理stderr
        while True:
            data = await stderr.read(4096)
            if not data:
                break
            sys.stderr.write(data.decode())
            sys.stderr.flush()

    # 将本地输入实时发送到SSH会话
    async def write_ssh_input():
        loop = asyncio.get_event_loop()
        while True:
            try:
                # 用线程池处理阻塞的标准输入读取，避免卡住事件循环
                input_data = await loop.run_in_executor(None, sys.stdin.readline)
                if not input_data:
                    break
                stdin.write(input_data)
                await stdin.drain()
            except EOFError:
                break

    # 并发执行输出读取和输入写入
    await asyncio.gather(read_ssh_output(), write_ssh_input())

async def run_client():
    async with asyncssh.connect(
        HOST, port=PORT, username=USER, password=PASSWORD,
        known_hosts=None
    ) as conn:
        # 申请PTY并指定终端类型，这是交互式Shell的核心配置
        stdin, stdout, stderr = await conn.open_session(
            term_type='xterm',
            request_pty=True
        )
        # 启动双向IO交互
        await handle_ssh_io(stdin, stdout, stderr)

if __name__ == "__main__":
    try:
        asyncio.run(run_client())
    except (OSError, asyncssh.Error) as exc:
        sys.exit(f'SSH连接失败: {str(exc)}')

关键说明 #

• request_pty=True：必须申请伪终端，否则远程Shell不会进入交互式模式（不会显示提示符，也不响应输入）
• term_type='xterm'：指定终端类型，让远程Shell正确处理换行、光标移动等控制字符
• 双向IO并发处理：用asyncio.gather同时运行输出读取和输入写入任务，保证交互的实时性；用run_in_executor处理本地标准输入的阻塞读取，避免阻塞AsyncIO事件循环
• 实时刷新输出：每次写入本地终端后调用flush()，确保内容立即显示，不会被缓冲区积压

内容的提问来源于stack exchange，提问作者johnc