routes/elevenlabs_hybrid.py
Rama main · réplica de lectura del repositorio original.
"""
routes/elevenlabs_hybrid.py — ElevenLabs + OpenClaw Hybrid Blueprint (P7-T5)
Provides two endpoints that form the bridge between ElevenLabs Conversational
AI (voice layer) and OpenClaw / Clawdbot Gateway (brain layer):
POST /api/elevenlabs-llm
Custom LLM endpoint configured in the ElevenLabs hybrid agent.
Receives the conversation context, extracts the latest user message,
forwards it to the Clawdbot Gateway using the persistent WebSocket
connection, strips canvas/HTML markers from the response, then
returns clean text to ElevenLabs in OpenAI-compatible SSE format
so ElevenLabs TTS can begin speaking as the first tokens arrive.
GET /api/canvas-pending
Side-channel for canvas commands extracted from OpenClaw responses.
Returns and clears the pending canvas command queue so the frontend
adapter (ElevenLabsHybridAdapter._startCanvasPolling) can load the
correct iframe without the agent reading HTML aloud.
Architecture:
Browser (ElevenLabs SDK)
→ POST /api/elevenlabs-llm (this module)
→ gateway_connection.stream_to_queue(session='voice-elevenlabs-hybrid')
← streaming text chunks
→ SSE to ElevenLabs TTS
OpenClaw response: "Dashboard ready! {canvas:present,url:/pages/stats.html} Check it out."
Spoken text: "Dashboard ready! Check it out."
Canvas queue: [{"action": "present", "url": "/pages/stats.html"}]
GET /api/canvas-pending returns → {"commands": [{"action": "present", "url": "..."}]}
Ref: future-dev-plans/16-ELEVENLABS-OPENCLAW-HYBRID.md
Ref: ADR-008 (Fallback chains — graceful degradation on Gateway unavailability)
"""
import json
import logging
import os
import queue
import re
import threading
from collections import deque
from flask import Blueprint, Response, jsonify, request
from services.gateway import gateway_connection
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Blueprint
# ---------------------------------------------------------------------------
elevenlabs_hybrid_bp = Blueprint('elevenlabs_hybrid', __name__)
# ---------------------------------------------------------------------------
# Session key
# Separate from ClawdBot's voice-main-N key so histories don't collide.
# ---------------------------------------------------------------------------
HYBRID_SESSION_KEY = os.getenv('ELEVENLABS_HYBRID_SESSION_KEY', 'voice-elevenlabs-hybrid')
# Optional shared secret for validating requests from ElevenLabs
HYBRID_LLM_SECRET = os.getenv('ELEVENLABS_HYBRID_LLM_SECRET', '')
# ---------------------------------------------------------------------------
# Canvas command side-channel
# Thread-safe deque; items are dicts: {"action": "present"|"close", "url": str}
# ---------------------------------------------------------------------------
_canvas_pending: deque = deque()
_canvas_lock = threading.Lock()
# ---------------------------------------------------------------------------
# Canvas extraction helpers
# ---------------------------------------------------------------------------
_CANVAS_PATTERN = re.compile(r'\{canvas:(\w+),url:([^}]+)\}')
_HTML_BLOCK_PATTERN = re.compile(r'```html[\s\S]*?```', re.IGNORECASE)
def _extract_canvas_commands(text: str) -> list:
"""
Parse {canvas:action,url:path} markers from OpenClaw response text.
Returns a list of {"action": str, "url": str} dicts.
"""
commands = []
for match in _CANVAS_PATTERN.finditer(text):
commands.append({
'action': match.group(1), # 'present', 'close', etc.
'url': match.group(2).strip(),
})
return commands
def _strip_canvas_markers(text: str) -> str:
"""
Remove {canvas:...} markers and raw ```html``` blocks from spoken text
so ElevenLabs TTS doesn't read them aloud.
"""
text = _CANVAS_PATTERN.sub('', text)
text = _HTML_BLOCK_PATTERN.sub('', text)
return text.strip()
def _queue_canvas_commands(commands: list) -> None:
"""Append extracted canvas commands to the pending queue (thread-safe)."""
if not commands:
return
with _canvas_lock:
for cmd in commands:
_canvas_pending.append(cmd)
# ---------------------------------------------------------------------------
# POST /api/elevenlabs-llm — custom LLM endpoint
# ---------------------------------------------------------------------------
@elevenlabs_hybrid_bp.route('/api/elevenlabs-llm', methods=['POST'])
def elevenlabs_custom_llm():
"""
Bridge ElevenLabs voice to OpenClaw brain.
ElevenLabs sends the full conversation context in OpenAI chat format:
{"messages": [{"role": "system"/"user"/"assistant", "content": "..."}]}
We extract the latest user message, forward to Gateway, stream the
response back as OpenAI-compatible SSE so ElevenLabs TTS can start
speaking on the first sentence rather than waiting for the full reply.
Fallback (ADR-008): if Gateway is unreachable, returns a graceful
error response so ElevenLabs speaks an apology rather than hanging.
"""
# ── Optional shared-secret auth ──────────────────────────────────────────
if HYBRID_LLM_SECRET:
auth_header = request.headers.get('Authorization', '')
if auth_header != f'Bearer {HYBRID_LLM_SECRET}':
return jsonify({'error': 'Unauthorized'}), 401
# ── Parse request ─────────────────────────────────────────────────────────
data = request.get_json(silent=True) or {}
messages = data.get('messages', [])
if not messages:
return _openai_error_response("No messages provided"), 400
# Extract the latest user turn (last message with role 'user')
user_message = ''
for msg in reversed(messages):
if msg.get('role') == 'user':
user_message = msg.get('content', '')
break
if not user_message:
return _openai_error_response("No user message found in context"), 400
logger.info(f"[ElevenLabsHybrid] Custom LLM request: {user_message[:80]!r}")
# ── Check Gateway availability ────────────────────────────────────────────
if not gateway_connection.is_configured():
logger.warning('[ElevenLabsHybrid] Gateway not configured — returning fallback response')
return _openai_sync_response(
"Sorry, my connection to the server brain is not configured right now. "
"Please check the CLAWDBOT_AUTH_TOKEN environment variable."
)
# ── Stream from Gateway → SSE to ElevenLabs ──────────────────────────────
def generate():
"""Generator: reads Gateway events, yields OpenAI SSE chunks."""
event_queue: queue.Queue = queue.Queue()
captured_actions = []
full_response_parts = []
# Run stream_to_queue in a background thread so we can yield from
# the main thread (Flask's response generator must be synchronous).
stream_thread = threading.Thread(
target=gateway_connection.stream_to_queue,
args=(event_queue, user_message, HYBRID_SESSION_KEY, captured_actions),
daemon=True,
)
stream_thread.start()
try:
while True:
try:
event = event_queue.get(timeout=60)
except queue.Empty:
logger.warning('[ElevenLabsHybrid] Gateway stream timeout')
break
etype = event.get('type')
if etype == 'delta':
chunk_text = event.get('text', '')
if chunk_text:
full_response_parts.append(chunk_text)
# Strip canvas markers from streaming chunks
clean_chunk = _strip_canvas_markers(chunk_text)
if clean_chunk:
yield _sse_delta(clean_chunk)
elif etype == 'text_done':
full_response = event.get('response') or ''.join(full_response_parts)
# Extract canvas commands from the full response
canvas_cmds = _extract_canvas_commands(full_response)
_queue_canvas_commands(canvas_cmds)
if canvas_cmds:
logger.info(f'[ElevenLabsHybrid] Queued {len(canvas_cmds)} canvas command(s)')
break
elif etype == 'error':
error_msg = event.get('error', 'Unknown Gateway error')
logger.error(f'[ElevenLabsHybrid] Gateway error: {error_msg}')
yield _sse_delta("I'm having trouble connecting right now. Please try again.")
break
elif etype == 'handshake':
# Connection established — nothing to yield to ElevenLabs
continue
# 'action' events: tool use / lifecycle events — log only
elif etype == 'action':
action_name = event.get('action', {}).get('type', 'unknown')
logger.debug(f'[ElevenLabsHybrid] Gateway action: {action_name}')
finally:
yield 'data: [DONE]\n\n'
stream_thread.join(timeout=5)
return Response(
generate(),
mimetype='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'X-Accel-Buffering': 'no', # disable nginx buffering for SSE
},
)
# ---------------------------------------------------------------------------
# GET /api/canvas-pending — canvas command side-channel
# ---------------------------------------------------------------------------
@elevenlabs_hybrid_bp.route('/api/canvas-pending', methods=['GET'])
def canvas_pending():
"""
Return and clear the pending canvas command queue.
The ElevenLabsHybridAdapter frontend polls this endpoint every second
during a hybrid conversation. When OpenClaw creates a canvas page, the
command appears here; the frontend then loads the iframe.
Response:
{"commands": [{"action": "present", "url": "/pages/stats.html"}, ...]}
Commands are consumed (cleared) on each call.
"""
with _canvas_lock:
commands = list(_canvas_pending)
_canvas_pending.clear()
return jsonify({'commands': commands})
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _sse_delta(text: str) -> str:
"""Format a text chunk as an OpenAI-compatible SSE delta event."""
payload = json.dumps({
'choices': [{
'delta': {
'role': 'assistant',
'content': text,
},
'finish_reason': None,
}]
})
return f'data: {payload}\n\n'
def _openai_sync_response(text: str):
"""
Return a non-streaming OpenAI-compatible response.
Used for fallback / error paths where we have the full text immediately.
"""
return jsonify({
'choices': [{
'message': {
'role': 'assistant',
'content': text,
},
'finish_reason': 'stop',
}]
})
def _openai_error_response(message: str):
"""Return an OpenAI-compatible error body."""
return jsonify({'error': {'message': message, 'type': 'invalid_request_error'}})