bruno
09984d49fa
- Serveur WebSocket direct avec websockets 16.0 - Interface HTML/CSS/JS dark theme avec chat - Relay vers API Hermes sur port 8642 - Historique de conversation (20 messages) - Clé API via Infisical - Testé et fonctionnel sur openclaw1.dev.home:7081
68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Quick test: start bot, hit WS, see result."""
|
|
import subprocess, os, json, time, sys, asyncio, threading, signal
|
|
|
|
os.chdir("/home/openclaw/workspace/jarvis-pipecat")
|
|
|
|
# Get key
|
|
creds = json.load(open("/opt/infisical/admin-credentials.json"))
|
|
r = subprocess.run(
|
|
["/home/openclaw/.local/bin/infisical", "secrets", "get",
|
|
"API_SERVER_KEY", "--domain", "https://secret.dracodev.net",
|
|
"--projectId", "192c1eed-6666-41ee-80df-a4636a2da0ad",
|
|
"--env", "prod", "--path", "/hermes-claw", "--plain",
|
|
"--token", creds["token"]],
|
|
capture_output=True, text=True, timeout=15)
|
|
os.environ["HERMES_API_KEY"] = r.stdout.strip()
|
|
os.environ["PYTHONUNBUFFERED"] = "1"
|
|
|
|
# Import bot (this triggers the key check print)
|
|
import bot
|
|
|
|
# Start uvicorn in thread
|
|
import uvicorn
|
|
def run_server():
|
|
uvicorn.run(bot.app, host="0.0.0.0", port=7081, log_level="info")
|
|
|
|
t = threading.Thread(target=run_server, daemon=True)
|
|
t.start()
|
|
|
|
# Wait for server
|
|
for i in range(20):
|
|
try:
|
|
import urllib.request
|
|
urllib.request.urlopen("http://localhost:7081/", timeout=1)
|
|
break
|
|
except:
|
|
time.sleep(0.3)
|
|
else:
|
|
print("FAIL: Server didn't start")
|
|
sys.exit(1)
|
|
|
|
# Test WS
|
|
async def test():
|
|
import websockets
|
|
try:
|
|
async with websockets.connect("ws://localhost:7081/ws", open_timeout=5) as ws:
|
|
msg = await asyncio.wait_for(ws.recv(), timeout=5)
|
|
data = json.loads(msg)
|
|
print(f"WELCOME: {data.get('message', msg)[:120]}")
|
|
|
|
await ws.send("Bonjour! Quelle heure est-il?")
|
|
print("SENT question")
|
|
|
|
resp = await asyncio.wait_for(ws.recv(), timeout=20)
|
|
data2 = json.loads(resp)
|
|
content = data2.get("content", str(data2))
|
|
print(f"REPLY: {content[:300]}")
|
|
print("\n✅✅✅ SUCCESS! JARVIS v3 works! ✅✅✅")
|
|
return True
|
|
except Exception as e:
|
|
print(f"WS FAIL: {type(e).__name__}: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
result = asyncio.run(test())
|
|
sys.exit(0 if result else 1)
|