如何通过Python原生主机实现AHK热键与Chrome扩展通信？

我正尝试构建一个最小可用系统：按下AutoHotkey (AHK)热键后，通过Python原生消息主机将消息发送至Chrome扩展。

架构如下：

AHK → 命名管道 → Python (原生主机) → Chrome Extension (Manifest V3)

最终目标很简单：按下热键（如NumpadAdd）时触发扩展内的事件。

已实现功能 #

• ✅ AHK打开命名管道\\.\pipe\NativeEventPipe并写入JSON。
• ✅ Python（原生主机）通过win32pipe.ReadFile(...)接收AHK消息。
• ✅ Python记录消息并调用sys.stdout.buffer.write(...) + flush()发送至Chrome。
• ✅ Chrome连接至主机并发送握手信息（{ cmd: "start", attempt: 1 }）。
• ❌ Chrome始终无法接收AHK事件，握手后onMessage从未触发。

原生消息边界以下的所有环节均正常工作——AHK输入已到达Python，但Chrome无法接收原生主机发送的消息。

Python代码片段（原生主机） #

def send_message(message):
    try:
        encoded = json.dumps(message).encode("utf-8")
        sys.stdout.buffer.write(struct.pack("I", len(encoded)))
        sys.stdout.buffer.write(encoded)
        sys.stdout.buffer.flush()
        return True
    except Exception as e:
        # 记录错误
        return False

扩展代码片段 #

port = chrome.runtime.connectNative("com.nativebridge.test");

port.onMessage.addListener((msg) => {
  console.log("✅ [SW] AHK Event:", msg);
});

需求说明 #

能否有人协助我成功完成该系统？我并非执着于命名管道——仅需一个最小可用解决方案，满足：

• AHK触发事件
• 该事件可被Chrome扩展识别
• 以某种形式使用原生消息通信（管道/文件/标准输入等）

同时还有以下疑问：

• 在该IPC模型中，我对sys.stdout.buffer.write(...)的使用是否正确？
• 管道模型与stdio原生消息结合是否存在根本性缺陷？
• 更简洁的设计（如从AHK调用短期子进程）是否更可靠？

文件与仓库 #

我已构建该项目的极简版本并上传至GitHub公开仓库。

系统环境 #

• Windows 11 Pro
• Brave Browser (Chromium)
• Chrome Extension (Manifest V3, service worker)
• AutoHotkey v1
• Python 3.11
• 使用win32pipe、win32file进行管道访问

问题排查与解决方案 #

1. sys.stdout.buffer.write使用正确性验证

你的send_message函数格式符合Chrome原生消息规范：先发送4字节无符号整数（小端序）表示消息长度，再发送UTF-8编码的JSON内容，最后flush。但需注意：

• 显式指定小端字节序（struct.pack("<I", len(encoded))），避免跨平台或环境差异问题
• 禁止向stdout输出任何调试信息（比如print语句），Chrome会严格校验消息格式，额外输出会直接中断通信

2. 管道模型与stdio原生消息的兼容性问题

原架构存在核心缺陷：Python主机同时阻塞监听AHK命名管道和处理Chrome的stdio通信，当Python卡在win32pipe.ReadFile等待AHK消息时，无法及时响应Chrome的通信心跳，甚至会导致Chrome判定主机无响应，消息无法被正确推送。

3. 更可靠的极简方案：AHK直接调用短期原生主机

放弃长驻Python主机，改为AHK触发热键时直接启动Python子进程，让Python作为短期原生主机向Chrome发送消息，流程更简单可靠：

Python原生主机修改

import sys
import struct
import json

def send_to_chrome(message):
    encoded = json.dumps(message).encode("utf-8")
    sys.stdout.buffer.write(struct.pack("<I", len(encoded)))
    sys.stdout.buffer.write(encoded)
    sys.stdout.buffer.flush()

if __name__ == "__main__":
    # 接收AHK传递的热键参数
    msg = {"event": "hotkey_triggered", "key": sys.argv[1]}
    send_to_chrome(msg)

AHK脚本修改

NumpadAdd::
    Run, pythonw.exe "你的脚本绝对路径\native_host.py" "NumpadAdd"
Return

扩展端修改

function handleNativeMessage() {
    const port = chrome.runtime.connectNative("com.nativebridge.test");
    port.onMessage.addListener((msg) => {
        console.log("✅ [SW] AHK Event:", msg);
    });
    port.onDisconnect.addListener(() => {
        // 短期主机发送消息后自动断开，属于正常流程
        console.log("ℹ️ [SW] Native host disconnected");
    });
}
// 扩展启动时初始化连接
handleNativeMessage();

4. 原架构修复方案（坚持长驻主机）

如果必须使用长驻Python主机，需用多线程拆分任务：

• 主线程处理Chrome的stdio通信，保持连接活跃
• 子线程非阻塞监听AHK命名管道，收到消息后交给主线程发送给Chrome

示例代码片段：

import sys
import struct
import json
import threading
import win32pipe
import win32file

def send_message(message):
    try:
        encoded = json.dumps(message).encode("utf-8")
        sys.stdout.buffer.write(struct.pack("<I", len(encoded)))
        sys.stdout.buffer.write(encoded)
        sys.stdout.buffer.flush()
        return True
    except Exception as e:
        print(f"发送消息错误: {e}", file=sys.stderr)
        return False

def listen_named_pipe():
    pipe_name = r"\\.\pipe\NativeEventPipe"
    while True:
        try:
            pipe = win32pipe.CreateNamedPipe(
                pipe_name,
                win32pipe.PIPE_ACCESS_DUPLEX,
                win32pipe.PIPE_TYPE_MESSAGE | win32pipe.PIPE_READMODE_MESSAGE | win32pipe.PIPE_WAIT,
                1, 65536, 65536, 0, None
            )
            win32pipe.ConnectNamedPipe(pipe, None)
            result, data = win32file.ReadFile(pipe, 65536)
            if result == 0:
                ahk_msg = json.loads(data.decode("utf-8"))
                send_message(ahk_msg)
            win32file.CloseHandle(pipe)
        except Exception as e:
            print(f"管道错误: {e}", file=sys.stderr)

def handle_chrome_messages():
    while True:
        # 读取Chrome的消息以保持连接活跃
        raw_length = sys.stdin.buffer.read(4)
        if not raw_length:
            break
        length = struct.unpack("<I", raw_length)[0]
        message = sys.stdin.buffer.read(length).decode("utf-8")
        # 调试信息输出到stderr，避免干扰stdout的消息格式
        print(f"收到Chrome消息: {message}", file=sys.stderr)

if __name__ == "__main__":
    pipe_thread = threading.Thread(target=listen_named_pipe, daemon=True)
    pipe_thread.start()
    handle_chrome_messages()

关键注意事项

• 确保Chrome原生主机注册表配置正确（HKCU\Software\Google\Chrome\NativeMessagingHosts\com.nativebridge.test指向合法的manifest文件）
• manifest文件的path需指向Python启动器+脚本绝对路径（比如"C:\\Python311\\pythonw.exe C:\\path\\to\\your\\script.py"）

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