#!/usr/bin/env python3
"""
JARVIS v3 Phase 1 — Relay WebSocket ↔ Hermes API
==================================================
PoC minimal : WebSocket texte → API Hermes → réponse texte.
Pas de Pipecat — intégration directe pour valider le concept.
"""
import asyncio
import json
import os
import subprocess
import sys
import httpx
from websockets.asyncio.server import serve
from websockets.http11 import Response, Headers
PORT = int(os.environ.get("JARVIS_PORT", "7081"))
# ─── Clé API ─────────────────────────────────────────────────
def get_hermes_key():
key = os.environ.get("HERMES_API_KEY", "") or os.environ.get("HERMES_ALEXA_KEY", "")
if key: return key
try:
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)
return r.stdout.strip()
except Exception as e:
print(f"⚠️ Infisical: {e}"); return ""
HERMES_KEY = get_hermes_key()
HERMES_URL = "http://localhost:8642/v1/chat/completions"
SYSTEM_PROMPT = (
"Tu es JARVIS, l'assistant vocal de Bruno. "
"Parle en français, sois concis (3 phrases max), chaleureux et utile. "
"Pas de markdown. Réponds comme si tu parlais à l'oral."
)
# ─── Page HTML ───────────────────────────────────────────────
HTML_PAGE = """
JARVIS v3 — Phase 1 PoC
⚡ JARVIS v3 — Phase 1 (Hermes direct)
🔴 Déconnecté
"""
# ─── HTTP Handler ────────────────────────────────────────────
async def http_handler(websocket):
"""Serve the test page over raw HTTP (before upgrade)."""
# This is handled by serve()'s process_request
pass
async def process_request(connection, request):
"""Serve HTML for HTTP requests, upgrade for WebSocket."""
if request.path == "/" or request.path == "/test.html":
headers = Headers([("Content-Type", "text/html; charset=utf-8")])
return Response(200, "OK", headers, HTML_PAGE.encode())
return None # Let WebSocket handler take over
# ─── WebSocket Handler ───────────────────────────────────────
async def ws_handler(websocket):
"""Handle WebSocket connection: relay messages to Hermes API."""
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
client = httpx.AsyncClient(timeout=45.0)
try:
# Send welcome
await websocket.send(json.dumps({
"type": "welcome",
"content": "👋 Bonjour Bruno ! Je suis JARVIS v3 (Phase 1). Pose-moi une question !"
}))
print("✅ Client connecté")
async for raw in websocket:
try:
data = json.loads(raw)
text = data.get("text", "").strip()
except:
text = raw.strip()
if not text:
continue
print(f"👤 {text[:80]}")
messages.append({"role": "user", "content": text})
# Keep last 20 messages + system prompt
if len(messages) > 21:
messages = [messages[0]] + messages[-20:]
# Send thinking indicator
await websocket.send(json.dumps({"type": "thinking"}))
# Call Hermes API
try:
resp = await client.post(
HERMES_URL,
headers={
"Authorization": f"Bearer {HERMES_KEY}",
"Content-Type": "application/json",
},
json={
"model": "Hermes Agent",
"messages": messages,
"temperature": 0.8,
"max_tokens": 300,
"stream": False,
},
)
resp.raise_for_status()
body = resp.json()
reply = body["choices"][0]["message"]["content"]
messages.append({"role": "assistant", "content": reply})
print(f"🤖 {reply[:80]}")
await websocket.send(json.dumps({
"type": "response",
"content": reply,
}))
except Exception as e:
print(f"❌ API error: {e}")
await websocket.send(json.dumps({
"type": "error",
"content": f"Erreur API: {str(e)[:200]}",
}))
except Exception as e:
print(f"👋 Déconnecté: {e}")
finally:
await client.aclose()
# ─── Main ────────────────────────────────────────────────────
async def main():
if not HERMES_KEY:
print("❌ Pas de clé API Hermes"); sys.exit(1)
print(f"🔑 {HERMES_KEY[:15]}...")
print(f"🌐 {HERMES_URL}")
print(f"🚀 JARVIS v3 PoC → http://0.0.0.0:{PORT}")
print(f" WebSocket → ws://0.0.0.0:{PORT}/ws")
async with serve(ws_handler, "0.0.0.0", PORT, process_request=process_request):
print(f"✅ Serveur prêt sur le port {PORT}")
await asyncio.get_running_loop().create_future() # run forever
if __name__ == "__main__":
asyncio.run(main())