muse-chat-api.py: CDP chat client (send/messages/wait) for smoke headless

bin/muse-chat-api.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+"""Muse.ai chat API via CDP (headless smoke profile).
+Usage:
+  muse-chat-api.py send "hello"          # send message to current chat
+  muse-chat-api.py messages             # print recent messages
+  muse-chat-api.py wait [timeout]       # wait for new response
+Requires: websocket-client (pip install --break-system-packages websocket-client)
+CDP relay must be up: http://10.201.87.2:9410/json/list
+"""
+import json, sys, time, urllib.request
+import websocket
+
+CDP_URL = "http://10.201.87.2:9410/json/list"
+
+def get_page():
+    with urllib.request.urlopen(CDP_URL, timeout=5) as r:
+        targets = json.load(r)
+    pages = [t for t in targets if t.get('type')=='page' and 'muse.ai' in t.get('url','')]
+    if not pages:
+        raise RuntimeError("no muse.ai page found")
+    return pages[0]
+
+def connect():
+    page = get_page()
+    ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=10)
+    return ws
+
+def ev(ws, expr, await_promise=False):
+    ws.send(json.dumps({"id":1,"method":"Runtime.evaluate",
+        "params":{"expression":expr,"returnByValue":True,"awaitPromise":await_promise}}))
+    r = json.loads(ws.recv())
+    return r.get('result',{}).get('result',{}).get('value')
+
+def send_message(text):
+    ws = connect()
+    try:
+        result = ev(ws, f"""
+(async () => {{
+  const ta = document.querySelector('textarea[aria-label="Message"]');
+  if (!ta) return 'ERROR: no composer';
+  ta.focus();
+  document.execCommand('insertText', false, {json.dumps(text)});
+  await new Promise(r => setTimeout(r, 200));
+  ta.dispatchEvent(new KeyboardEvent('keydown', {{key:'Enter', code:'Enter', keyCode:13, bubbles:true}}));
+  return 'sent';
+}})()
+""", await_promise=True)
+        return result
+    finally:
+        ws.close()
+
+def get_messages(limit=10):
+    ws = connect()
+    try:
+        return ev(ws, f"""
+Array.from(document.querySelectorAll('p')).slice(-{limit}).map(e=>e.innerText).join('\\n---\\n')
+""")
+    finally:
+        ws.close()
+
+def wait_for_response(timeout=60, poll=2):
+    """Wait for a new assistant message after sending."""
+    ws = connect()
+    try:
+        before = ev(ws, "document.body.innerText.length")
+        start = time.time()
+        while time.time() - start < timeout:
+            time.sleep(poll)
+            # Check if there's a "stop" button (indicates generating)
+            generating = ev(ws, """
+!!document.querySelector('button[aria-label*="Stop"], button[aria-label*="stop"]')
+""")
+            if not generating:
+                # Give it a moment to finish rendering
+                time.sleep(2)
+                return get_messages(5)
+        return "TIMEOUT waiting for response"
+    finally:
+        ws.close()
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print(__doc__)
+        sys.exit(1)
+    cmd = sys.argv[1]
+    if cmd == "send" and len(sys.argv) > 2:
+        print(send_message(sys.argv[2]))
+    elif cmd == "messages":
+        print(get_messages(int(sys.argv[2]) if len(sys.argv)>2 else 10))
+    elif cmd == "wait":
+        print(wait_for_response(int(sys.argv[2]) if len(sys.argv)>2 else 60))
+    else:
+        print("unknown command")