Omar SobhandClaude Opus 4.8
413230255e
serve.py gains /tts (ElevenLabs synthesis) + /config (advertises tts mode). Client plays the returned MP3 when ElevenLabs is on, else falls back to browser speechSynthesis; barge-in stops in-flight audio. Default voice Sarah (free-tier usable); key read from ELEVENLABS_API_KEY env, never committed. README documents the free-tier library-voice 402 gotcha. Verified: 200 audio/mpeg, 38KB MP3. Co-Authored-By: Claude Opus 4.8 <[email protected]>
110 lines
5.0 KiB
Python
110 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""APESS voice client host + gateway proxy.
|
|
|
|
Serves index.html AND proxies POST /webhook to the node's ZeroClaw gateway on the
|
|
same origin — so the browser needs no CORS and no bearer token, and we reuse the
|
|
reliable /webhook path (the /ws/chat path drops the peripheral matrix tool).
|
|
|
|
Usage:
|
|
NODE_URL=http://127.0.0.1:8080 NODE_TOKEN=zc_xxx python3 serve.py [port]
|
|
|
|
- NODE_URL node gateway base (default http://127.0.0.1:8080; via `adb forward
|
|
tcp:8080 tcp:8080` over USB, or the board's LAN IP:8080 on the day).
|
|
- NODE_TOKEN gateway bearer token (kept server-side, never sent to the browser).
|
|
- port local port to serve on (default 8090). Open http://localhost:<port>.
|
|
|
|
localhost is a secure context, so the browser grants mic access; the proxy hop is
|
|
server-side, so there is no CORS. One process, one origin.
|
|
"""
|
|
import os, sys, json, urllib.request
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
NODE_URL = os.environ.get("NODE_URL", "http://127.0.0.1:8080").rstrip("/")
|
|
NODE_TOKEN = os.environ.get("NODE_TOKEN", "")
|
|
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8090
|
|
# Optional ElevenLabs TTS (server-side; key never reaches the browser). If unset,
|
|
# the client falls back to the browser's built-in speechSynthesis voice.
|
|
ELEVEN_KEY = os.environ.get("ELEVENLABS_API_KEY", "")
|
|
ELEVEN_VOICE = os.environ.get("ELEVENLABS_VOICE_ID", "EXAVITQu4vr4xnSDxMaL") # Sarah (free-tier usable)
|
|
ELEVEN_MODEL = os.environ.get("ELEVENLABS_MODEL", "eleven_turbo_v2_5")
|
|
|
|
class H(BaseHTTPRequestHandler):
|
|
def log_message(self, *a): pass # quiet
|
|
|
|
def _send(self, code, body, ctype="application/json"):
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
if self.path == "/ping":
|
|
return self._send(200, b'{"ok":true}')
|
|
if self.path == "/config":
|
|
mode = "elevenlabs" if ELEVEN_KEY else "browser"
|
|
return self._send(200, json.dumps({"tts": mode}).encode())
|
|
path = "/index.html" if self.path in ("/", "") else self.path.split("?")[0]
|
|
fp = os.path.normpath(os.path.join(HERE, path.lstrip("/")))
|
|
if not fp.startswith(HERE) or not os.path.isfile(fp):
|
|
return self._send(404, b"not found", "text/plain")
|
|
ctype = "text/html" if fp.endswith(".html") else "text/plain"
|
|
with open(fp, "rb") as f:
|
|
self._send(200, f.read(), ctype)
|
|
|
|
def do_POST(self):
|
|
if self.path == "/tts":
|
|
return self._tts()
|
|
if not self.path.startswith("/webhook"):
|
|
return self._send(404, b'{"error":"not found"}')
|
|
n = int(self.headers.get("Content-Length", 0))
|
|
body = self.rfile.read(n)
|
|
q = self.path[len("/webhook"):] # keep ?agent=...
|
|
req = urllib.request.Request(
|
|
f"{NODE_URL}/webhook{q}", data=body, method="POST",
|
|
headers={"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {NODE_TOKEN}"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
self._send(resp.status, resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
self._send(e.code, e.read() or b'{"error":"upstream"}')
|
|
except Exception as e:
|
|
self._send(502, json.dumps({"error": str(e)}).encode())
|
|
|
|
def _tts(self):
|
|
if not ELEVEN_KEY:
|
|
return self._send(503, b'{"error":"tts disabled"}')
|
|
n = int(self.headers.get("Content-Length", 0))
|
|
try:
|
|
text = json.loads(self.rfile.read(n)).get("text", "").strip()
|
|
except Exception:
|
|
text = ""
|
|
if not text:
|
|
return self._send(400, b'{"error":"no text"}')
|
|
payload = json.dumps({
|
|
"text": text, "model_id": ELEVEN_MODEL,
|
|
"voice_settings": {"stability": 0.5, "similarity_boost": 0.75},
|
|
}).encode()
|
|
req = urllib.request.Request(
|
|
f"https://api.elevenlabs.io/v1/text-to-speech/{ELEVEN_VOICE}",
|
|
data=payload, method="POST",
|
|
headers={"xi-api-key": ELEVEN_KEY, "Content-Type": "application/json",
|
|
"Accept": "audio/mpeg"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
self._send(200, resp.read(), "audio/mpeg")
|
|
except urllib.error.HTTPError as e:
|
|
self._send(e.code, e.read() or b'{"error":"tts upstream"}')
|
|
except Exception as e:
|
|
self._send(502, json.dumps({"error": str(e)}).encode())
|
|
|
|
if __name__ == "__main__":
|
|
if not NODE_TOKEN:
|
|
print("WARN: NODE_TOKEN is empty — gateway calls will 401.", file=sys.stderr)
|
|
print(f"APESS voice client → {NODE_URL}")
|
|
print(f"TTS: {'ElevenLabs ('+ELEVEN_VOICE+')' if ELEVEN_KEY else 'browser (speechSynthesis)'}")
|
|
print(f"open http://localhost:{PORT}")
|
|
ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever()
|