hermes-agentvibes-hook

Hermes ↔ AgentVibes TTS Hook

Send every Hermes response as spoken audio to a remote machine via the AgentVibes SSH receiver.

Architecture

Hermes (server/Docker)
  └── agent:end event fires
      └── hooks/agentvibes-tts/handler.py
          └── strip markdown, truncate at word boundary
          └── JSON payload → base64-encode
          └── SSH → agentvibes-receiver@<host>:<port>
              └── Remote machine queues & plays via Piper TTS

Setup

1. Install AgentVibes on the target machine (the one with speakers)

npx agentvibes install

2. Generate an SSH key on the Hermes server

ssh-keygen -t ed25519 -f /absolute/path/to/id_ed25519_agentvibes -N ""

Use an absolute path — tilde (~) expansion does not work inside Python subprocess argument lists.

Register the public key on the target machine's agentvibes-receiver user.

3. Create the hook directory and files

Create <hermes_home>/hooks/agentvibes-tts/HOOK.yaml:

name: agentvibes-tts
description: Send agent responses to AgentVibes TTS remotely
events:
  - agent:end

Create <hermes_home>/hooks/agentvibes-tts/handler.py — update the four config constants at the top:

"""
AgentVibes TTS Hook — fires on agent:end to speak the agent's response.

Sends the response via SSH to the AgentVibes receiver using the full JSON
payload format (supports voice, project, effects metadata).
"""

import asyncio
import base64
import json
import logging
import os
import re
import subprocess
import time

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

AGENTVIBES_SSH_KEY  = "/absolute/path/to/id_ed25519"   # no tilde — subprocess won't expand it
AGENTVIBES_HOST     = "<target-ip-or-hostname>"
AGENTVIBES_PORT     = "<receiver-port>"
AGENTVIBES_USER     = "agentvibes-receiver"
AGENTVIBES_VOICE    = "en_US-libritts-high::Leo-8"
AGENTVIBES_PROJECT  = "hermes"
MAX_CONTENT_LEN     = 200          # chars of spoken content (prefix NOT counted)
PREFIX              = "Hermes here, "
_KNOWN_HOSTS        = "/absolute/path/to/known_hosts"   # persistent across restarts
_LOG_FILE           = os.path.join(os.path.dirname(os.path.abspath(__file__)), "tts-hook.log")

# ---------------------------------------------------------------------------
# File logger — writes to tts-hook.log next to handler.py, never to stdout
# ---------------------------------------------------------------------------

_log = logging.getLogger("agentvibes-tts")
if not _log.handlers:
    try:
        _h = logging.FileHandler(_LOG_FILE, encoding="utf-8")
        _h.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
        _log.addHandler(_h)
    except OSError:
        _log.addHandler(logging.NullHandler())
    _log.setLevel(logging.INFO)

# ---------------------------------------------------------------------------
# Rate limiting — prevent queue flooding on rapid consecutive turns
# ---------------------------------------------------------------------------

_last_sent: float = 0.0
_MIN_INTERVAL_S: float = 3.0


def _tts_speak(text: str) -> None:
    """Build JSON payload and send to AgentVibes receiver over SSH."""
    global _last_sent

    if not text or not text.strip():
        return

    now = time.monotonic()
    if now - _last_sent < _MIN_INTERVAL_S:
        _log.info("rate-limited: %.1fs since last send — skipping", now - _last_sent)
        return
    _last_sent = now

    # Truncate at word boundary so the spoken text never cuts mid-syllable
    full_text = PREFIX + text.strip()
    max_len = len(PREFIX) + MAX_CONTENT_LEN
    if len(full_text) > max_len:
        truncated = full_text[:max_len].rsplit(" ", 1)[0]
        full_text = truncated + "..."

    payload = base64.b64encode(
        json.dumps({
            "text":     full_text,
            "voice":    AGENTVIBES_VOICE,
            "project":  AGENTVIBES_PROJECT,
            "provider": "piper",
            "pretext":  "",   # prefix already prepended above
            "effects":  "",
            "music":    "",
            "volume":   "",
            "speed":    "",
        }).encode()
    ).decode()

    cmd = [
        "ssh",
        "-i", AGENTVIBES_SSH_KEY,
        "-o", "ConnectTimeout=5",
        # accept-new: trust the host key on first connection, reject changes
        # (MITM protection without interactive prompts)
        "-o", "StrictHostKeyChecking=accept-new",
        "-o", f"UserKnownHostsFile={_KNOWN_HOSTS}",
        "-o", "BatchMode=yes",
        "-p", AGENTVIBES_PORT,
        f"{AGENTVIBES_USER}@{AGENTVIBES_HOST}",
        payload,
    ]

    try:
        result = subprocess.run(cmd, capture_output=True, timeout=15)
        if result.returncode != 0:
            stderr = result.stderr.decode(errors="replace").strip()
            _log.warning("ssh exit %d: %s", result.returncode, stderr or "(no stderr)")
        else:
            stdout = result.stdout.decode(errors="replace").strip()
            _log.info("queued OK: %s", stdout)
    except subprocess.TimeoutExpired:
        _log.warning("ssh timed out after 15s to %s:%s", AGENTVIBES_HOST, AGENTVIBES_PORT)
    except Exception as exc:
        _log.warning("ssh error: %s", exc)


async def handle(event_type: str, context: dict) -> None:
    """Async handler — only processes agent:end events."""
    if event_type != "agent:end":
        return

    response = (context.get("response") or "").strip()
    if not response:
        return

    response = _strip_markdown(response)

    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, _tts_speak, response)


def _strip_markdown(text: str) -> str:
    """Remove markdown artifacts that sound unnatural in TTS."""
    # Fenced code blocks — always unlistenable
    text = re.sub(r"```[\s\S]*?```", "", text)
    # Inline code — unwrap
    text = re.sub(r"`([^`]+)`", r"\1", text)
    # Bold / italic markers
    text = re.sub(r"\*{1,2}([^\*]+)\*{1,2}", r"\1", text)
    # Links — keep label, drop URL
    text = re.sub(r"\[([^\]]+)\]\([^\)]+\)", r"\1", text)
    # Emoji — strip everywhere, not just line-start
    text = re.sub(
        r"[\U0001F300-\U0001F9FF"
        r"\U00002600-\U000027FF"
        r"\U0000FE00-\U0000FE0F"
        r"•→↓↑]+",
        "",
        text,
    )
    text = re.sub(r"\s+", " ", text).strip()
    return text