使用asyncio.subprocess调用含input的Python脚本时提示不显示

问题原因 #

input()的提示文本（如name: ）末尾没有换行符，而你的代码使用process.stdout.readline()读取输出——该方法会持续等待换行符才返回数据。这导致提示文本被滞留在子进程的stdout缓冲区中，直到用户输入完成后，子进程输出带换行的内容时，才会把之前的提示文本一并返回，最终出现提示不可见的情况。

解决方案（无需修改test.py） #

放弃按行读取的方式，改用逐字节读取stdout，确保无换行的提示文本能被及时捕获并显示。同时单独处理输入转发，避免输入输出阻塞。

修改后的main.py代码：

import asyncio
import sys

async def read_stream(stream, output_callback):
    while True:
        # 逐字节读取，保证无换行的提示能被及时获取
        chunk = await stream.read(1)
        if not chunk:
            break
        output_callback(chunk.decode())

async def main():
    process = await asyncio.create_subprocess_exec(
        'python3', '-u', 'test.py',
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE
    )

    # 启动异步任务读取子进程stdout，实时输出
    asyncio.create_task(read_stream(
        process.stdout,
        lambda content: print(f"; {content}", end="", flush=True)
    ))

    # 处理输入：将终端输入转发给子进程（Web场景可替换为客户端输入）
    loop = asyncio.get_running_loop()
    while True:
        try:
            # 同步读取终端输入（异步环境下需用线程池）
            user_input = await loop.run_in_executor(None, sys.stdin.readline)
        except EOFError:
            break
        if not user_input:
            break
        # 写入子进程stdin，保留输入的换行符
        process.stdin.write(user_input.encode())
        await process.stdin.drain()

    await process.wait()

asyncio.run(main())

关键改动说明 #

1. stdout读取方式：用read(1)逐字节读取，替代readline()，确保无换行的提示文本能被实时捕获。
2. 异步任务分离：将stdout读取和stdin处理拆分为独立逻辑，避免互相阻塞。
3. 输入转发控制：将stdin改为asyncio.subprocess.PIPE，主动控制输入传递，适配后续Web场景的输入对接（如WebSocket消息）。
4. 实时输出：打印时使用end=""保持原格式，flush=True确保内容立即显示到终端。

测试运行结果 #

执行后会得到符合预期的输出：

python3 main.py
; hello, world!
; name: s
; hey s!

（显示; name: 时程序会等待用户输入，输入完成后继续执行）

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