diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 73f3cba435df..4dc93a72aac9 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -526,6 +526,231 @@ def strip_think_blocks(agent, content: str) -> str:
return content
+# ── Reasoning-prose stripper ────────────────────────────────────────
+# Some chat-tuned reasoning models (notably minimax-m3, kimi-k2.5/2.6)
+# emit their chain-of-thought as natural-language sentences directly in
+# the visible ``content`` field instead of using ``…`` XML
+# tags or the structured ``reasoning_content`` channel. Examples that
+# leaked to chat in the wild:
+#
+# - "Let me check what the gateway is doing."
+# - "Found it. The error is at gateway/run.py:17678."
+# - "Now the **real** bug surface for ... — let me check how the agent
+# output gets transformed BEFORE it reaches truncate_message."
+#
+# The patterns cluster around reasoning meta-verbs (let me X, now I see,
+# now the real X, found it, aha, I can see) typically at the start of a
+# sentence. The fix: detect a leading "reasoning preamble" (one or more
+# sentences matching the patterns) and drop it, keeping any substantive
+# answer that follows. Also drop trailing reasoning sentences — the
+# min/max-m3 pattern frequently emits one final "Found it." or "Got it."
+# after the real answer. Conservative on purpose — the helper refuses
+# to touch short messages (under 40 chars) and short-circuits to the
+# input when stripping would leave < 8 chars of visible answer.
+# CLI / TUI passes should bypass this helper (callers gate on platform)
+# so the reasoning is still visible to the operator working locally.
+
+_REASONING_PROSE_OPENERS = (
+ # "Let me / Let's" + verb (lowercase ASCII only, then word boundary)
+ r"\blet(?:'s|s| me| us)\s+(?:also\s+|just\s+|first\s+|actually\s+|quickly\s+|"
+ r"now\s+|try\s+to\s+)?(?Pthink|check|look|trace|find|examine|"
+ r"reason|verify|recall|consider|review|recheck|re-?verify|re-?check|"
+ r"test|push|step|back|back-?out|backtrack|skip|read|run|do|go|see|open|"
+ r"close|inspect|examine|investigate|walk|drill|dig|break|split|cross|"
+ r"poke|grep|search|scan|hit|re-?read|cross-?check|take|put|move|kill|"
+ r"restart|rebuild|recompile|rerun|reapply|revert|apply|patch|fix|"
+ r"compare|diff|map|reconstruct|retrace|simulate|verify|cross-reference|"
+ r"isolate|identify|catalogue|enumerate|list|count|summarise|summarize|"
+ r"elaborate|expand|recap)\b",
+ # "Now let me / Now I can / Now the real / Now I understand"
+ r"\bnow\s+(?:let\s+me|i\s+can|i\s+see|i\s+have|i\s+understand|"
+ r"the\s+real|it's\s+clear|everything|the\s+full|here|we\s+have)\b",
+ # "Found it. / Found the bug. / Found the root cause. / Found X"
+ r"\bfound\s+(?:it|the|that|an?|one|two|three|my|our|another)\b",
+ # "Aha —" / "Aha:"
+ r"\baha\b",
+ # "I see the / I can see / I lost / I should check / I need to / I think I"
+ r"\bi\s+(?:see|can\s+see|lost|should(?:n't)?|need|want|have\s+to|"
+ r"think\s+i|now\s+see|finally\s+see|now\s+have|now\s+need)\b",
+ # "Wait, / Wait —" (mid-thought correction)
+ r"\bwait\s*[,—-]",
+ # "Interesting —" / "Interesting."
+ r"\binteresting\b",
+ # "Smoking gun" / "this is the X"
+ r"\bthis\s+is\s+the\s+(?:smoking\s+gun|root\s+cause|bug|real\s+issue|"
+ r"actual\s+issue|real\s+bug|actual\s+bug|core\s+issue|key\s+issue)\b",
+ # "Confirmed:" / "Confirmed."
+ r"\bconfirmed\s*[:.]",
+ # "Let me give / hand / pass / set" and similar light verbs
+ r"\blet\s+me\s+(?:give|hand|pass|set|tell|show)\b",
+ # "Let me try" / "let me attempt" / "let me see"
+ r"\blet\s+me\s+(?:try|attempt|see|head|jump|dive)\b",
+ # "Got it." / "Got it —" (trailing acknowledgment)
+ r"\bgot\s+it\b",
+ # "Right," / "Right —" (trailing realization)
+ r"\bright\s*[,—\-]",
+ # "OK so" / "Okay so" (transitional opener)
+ r"\b(?:ok|okay)\s+so\b",
+)
+
+# Compile once at import.
+# Public re-exports for the cross-module stream-time consumer
+# (gateway.stream_consumer) and any future call site — see
+# ``agent.reasoning_prose`` for the supported import surface. We keep
+# the private names here as thin aliases for backward compatibility.
+from agent.reasoning_prose import ( # noqa: E402 (re-export)
+ REASONING_PROSE_OPENERS_RE as _REASONING_PROSE_OPENERS_RE,
+ SENTENCE_END_RE as _SENTENCE_END,
+)
+
+# Sentence-end characters used to find the end of the preamble.
+# (Kept as a module-level reference for any in-tree call that imported
+# it before the public re-export. New code should import the public
+# name from ``agent.reasoning_prose`` directly.)
+
+
+def strip_reasoning_prose(
+ agent,
+ content: str,
+ *,
+ min_length: int = 25,
+ min_remainder: int = 8,
+ strip_trailing: bool = True,
+) -> str:
+ """Strip leading reasoning-prose sentences from assistant content.
+
+ Some chat-tuned reasoning models (notably minimax-m3 and the kimi-k2.5
+ family) emit their chain-of-thought as natural-language sentences
+ directly in the visible ``content`` field. ``strip_think_blocks``
+ handles XML tag variants (``…``) but is blind to
+ prose-style leaks. This helper removes the leading reasoning
+ preamble while preserving any substantive answer that follows.
+
+ Conservative by design:
+ * Refuses to touch content shorter than ``min_length`` (default
+ 25 chars — too risky on shorter messages where the model has
+ a one-liner that's the actual answer).
+ * Walks sentences one at a time from the start. Stops as soon as
+ a sentence doesn't match a reasoning opener, OR as soon as the
+ remaining content would be shorter than ``min_remainder`` chars
+ (avoid leaving a fragment).
+ * If no opener matches, returns content unchanged.
+
+ When ``strip_trailing`` is True (default), also walks the
+ *trailing* sentences of the content and drops any that match the
+ reasoning-opener set. This catches the common min/max-m3 pattern
+ where the model emits its final reasoning sentence(s) at the end
+ of an otherwise-good answer — e.g. "Yeah, the bug is in
+ base.py:4722. Found it." where "Found it." would otherwise leak.
+ Trailing stripping is gated on the content having ≥3 sentences
+ so we don't over-strip a 2-sentence answer that happens to
+ start with a reasoning verb.
+
+ The helper is a no-op on CLI / TUI paths. Callers gate on the
+ platform (the gateway ``_sanitize_gateway_final_response`` is the
+ typical chokepoint).
+ """
+ if not content or not isinstance(content, str):
+ return content or ""
+ text = content
+ if len(text) < min_length:
+ return text
+
+ # ── Leading preamble strip ────────────────────────────────────
+ # Walk the content forward, one sentence at a time, dropping any
+ # sentence whose first non-whitespace token is a reasoning opener.
+ # Crucial: we track a ``cursor`` and only ``search()`` *from* that
+ # cursor, never from position 0 again. An earlier implementation
+ # re-searched the whole ``text`` after each cut, which let a
+ # reasoning opener in sentence N+1 (or in a legitimate mid-message
+ # clause like "I can see the Submit button is red") be silently
+ # dropped even when the user-facing sentence was substantive. The
+ # cursor discipline below is the load-bearing fix for that bug.
+ cursor = 0
+ while cursor < len(text):
+ chunk = text[cursor:]
+ match = _REASONING_PROSE_OPENERS_RE.search(chunk)
+ if not match:
+ break
+ # The opener regex's lookbehind only fires at the *start* of
+ # ``chunk`` OR right after sentence-end punctuation. When
+ # ``cursor > 0`` and the match.start() > 0, the opener is mid-
+ # sentence, which means the consumer is NOT supposed to strip
+ # it — we've already consumed the leading preamble and any
+ # further opener is part of the user-facing answer. Bail out.
+ if match.start() != 0:
+ break
+ opener_end = match.end()
+ boundary = _SENTENCE_END.search(chunk, opener_end)
+ if not boundary:
+ # No sentence boundary after the opener — the opener runs
+ # to end of content. Drop from the opener onward; the
+ # cursor advances to the match start (everything before is
+ # also being dropped because it's only the partial preamble
+ # we haven't already consumed).
+ new_text = text[: cursor + match.start()].rstrip()
+ if not new_text or len(new_text) < min_remainder:
+ return ""
+ if new_text == text[:cursor]:
+ break
+ text = new_text
+ cursor = 0 # reset so the next pass re-anchors at the new start
+ else:
+ # Drop opener through end of its sentence; the remainder
+ # starts at boundary.end() within ``chunk`` (which is
+ # offset by ``cursor`` in the full text).
+ new_cursor = cursor + boundary.end()
+ text = text[:new_cursor].rstrip() + text[new_cursor:].lstrip()
+ text = text.strip()
+ if not text or len(text) < min_remainder:
+ return ""
+ cursor = 0 # restart anchor for the next pass
+ if len(text) < min_length:
+ return text
+
+ # ── Trailing reasoning-sentence strip ─────────────────────────
+ # Only if the content is long enough that a 1-2 sentence answer
+ # would have already been returned via the leading pass without
+ # truncation. This catches the min/max-m3 pattern of trailing
+ # "Found it." or "Got it." after a real answer.
+ if strip_trailing and len(text) >= min_length * 3:
+ # Count sentences by sentence-end punctuation. Need at least
+ # 3 to consider this multi-sentence (so a 2-sentence
+ # "Let's go. The fix is X." isn't over-stripped).
+ sentence_count = sum(1 for _ in _SENTENCE_END.finditer(text)) + 1
+ if sentence_count >= 3:
+ # Walk from the end: find the last sentence-end, then
+ # check whether the text after it matches an opener at a
+ # boundary. If so, drop it. Repeat once for cases where
+ # the final two sentences are both reasoning.
+ for _ in range(2):
+ last_boundary = None
+ for m in _SENTENCE_END.finditer(text):
+ last_boundary = m
+ if last_boundary is None:
+ break
+ tail_start = last_boundary.end()
+ tail = text[tail_start:].lstrip()
+ if not tail or len(tail) >= min_length:
+ break
+ # Tail must end with terminal punctuation to be a
+ # complete sentence (avoids over-stripping mid-sentence
+ # when the assistant just trails off).
+ if not re.search(r"[\.\!\?]\s*$", tail):
+ break
+ # Check the tail starts with an opener. Re-anchor
+ # _REASONING_PROSE_OPENERS_RE to the tail start so
+ # the boundary check fires correctly.
+ tail_match = _REASONING_PROSE_OPENERS_RE.match(tail)
+ if not tail_match:
+ break
+ # Drop the trailing reasoning sentence.
+ text = text[:tail_start].rstrip()
+ if not text or len(text) < min_remainder:
+ return ""
+
+ return text
+
def recover_with_credential_pool(
agent,
@@ -2344,6 +2569,7 @@ def force_close_tcp_sockets(client: Any) -> int:
"drop_thinking_only_and_merge_users",
"restore_primary_runtime",
"extract_reasoning",
+ "strip_reasoning_prose",
"dump_api_request_debug",
"anthropic_prompt_cache_policy",
"create_openai_client",
diff --git a/agent/reasoning_prose.py b/agent/reasoning_prose.py
new file mode 100644
index 000000000000..22f09ce0691b
--- /dev/null
+++ b/agent/reasoning_prose.py
@@ -0,0 +1,145 @@
+"""Reasoning-prose stripping — public surface for the regex primitives.
+
+Some chat-tuned reasoning models (notably minimax-m3 and the kimi-k2.5
+family) emit their chain-of-thought as natural-language sentences
+directly in the visible ``content`` field, instead of using XML-style
+``…`` tags. We strip that prose in two places:
+
+* **Final-response chokepoint** — ``agent.agent_runtime_helpers.strip_reasoning_prose``
+ walks the entire response and removes leading (and optionally trailing)
+ reasoning sentences. This module owns the regexes that function uses.
+* **Stream-time** — ``gateway.stream_consumer.GatewayStreamConsumer`` strips
+ the leading reasoning sentence on the *first* delta of a turn, so users
+ never see the chain-of-thought flash by mid-stream. It needs the same
+ opener regex.
+
+Originally both call sites imported the regexes as the private
+``_REASONING_PROSE_OPENERS_RE`` and ``_SENTENCE_END`` symbols from
+``agent.agent_runtime_helpers``. That created a tight, non-obvious
+cross-module coupling — renaming the private symbol would silently break
+the gateway's stream-time stripper, with no obvious link in either file
+explaining the dependency.
+
+This module is the supported import surface. Both call sites now use
+the public ``REASONING_PROSE_OPENERS_RE`` and ``SENTENCE_END_RE`` names
+imported from here, so the dependency is discoverable via the import
+graph and grep finds it without needing to know the leading underscore
+is load-bearing.
+
+Pattern-design notes (kept here so future maintainers don't repeat the
+mistake):
+
+The opener patterns are *tuned* for chat-tuned reasoning models. A naive
+list of "thinking verbs" (let me, I think, so, actually, wait) is too
+broad — those words show up in normal user-facing answers all the time
+("I'll be there at 5", "wait for it", "the first thing to try is X").
+The mistake we hit before: matching ``\\bi'll\\b`` or ``\\bso[,\\s]`` as
+opener patterns *anywhere* in the text causes legitimate mid-sentence
+phrases to be silently deleted from the user-visible reply.
+
+So every pattern here is **start-of-message-anchored in practice** by
+the consumer (``strip_reasoning_prose`` and the stream-time stripper
+both walk from position 0 and stop as soon as a non-opener is hit), and
+each pattern is **verb-shaped**: it must be a meta-cognitive opener, not
+a content word. "Let me check" is meta; "I'll be there" is not — the
+distinction is that the meta opener comes with a thinking verb
+attached (``let me VERB``, ``I think I VERB``, ``so let me``), and a
+trailing verb is what separates it from the conversational use.
+"""
+
+from __future__ import annotations
+
+import re
+
+
+# ── Source of truth: the opener patterns ────────────────────────────────
+# Each entry is a substring of a single regex alternative. They match
+# English chain-of-thought openers that some chat-tuned reasoning models
+# emit directly in the assistant's visible content field.
+#
+# Tightness rules (read before adding new patterns):
+# 1. A pattern must require a *thinking verb* attached. Bare ``\\bso\\b``,
+# bare ``\\bfirst\\b``, bare ``\\bi'll\\b``, etc. will match in normal
+# answers and silently delete content.
+# 2. A pattern must be *anchored* by the consumer to start-of-message.
+# Don't try to catch mid-message reasoning here — the whole function
+# stops at the first non-opener, so any opener it matches must be
+# the *first* thing in the message.
+# 3. Keep the list short. Every entry has a false-positive cost. If
+# you're tempted to add ``\\bactually\\b`` remember "actually works"
+# is a legitimate reply, not reasoning.
+_REASONING_PROSE_OPENERS: tuple[str, ...] = (
+ # "Let me / Let's" + (optional adverb) + thinking verb.
+ # The verb list is the *only* thing that separates "Let me check"
+ # (reasoning) from "Let's meet at 5" (content). Don't strip a verb.
+ r"\blet['']?s\b\s+(?:also\s+|just\s+|first\s+|actually\s+|quickly\s+"
+ r"|now\s+|try\s+to\s+)?"
+ r"(?Pthink|check|look|trace|find|examine|reason|verify|"
+ r"recall|consider|review|recheck|re-?verify|re-?check|test|push|"
+ r"step\s+back|backtrack|skip|read|run|do|go|see|open|close|"
+ r"inspect|investigate|walk|drill|dig|break|split|cross|"
+ r"poke|grep|search|scan|hit|re-?read|cross-?check|take|put|"
+ r"move|kill|restart|rebuild|recompile|rerun|reapply|revert|"
+ r"apply|patch|fix|compare|diff|map|reconstruct|retrace|"
+ r"simulate|isolate|identify|enumerate|summarise|summarize|"
+ r"elaborate|expand|recap|give|hand|pass|set|tell|show|try|"
+ r"attempt|see|head|jump|dive)\b",
+ # "Now let me / Now I can see / Now I understand / Now the real"
+ r"\bnow\s+(?:let['']?s\s+me|let\s+me|i\s+can\s+see|i\s+see|"
+ r"i\s+have|i\s+understand|it['']?s\s+clear|the\s+real|"
+ r"everything|the\s+full|here|we\s+have)\b",
+ # "Found it" / "Found the bug" — punctuated and unpunctuated
+ r"\bfound\s+(?:it|the|that|an?|one|two|three|my|our|another)\b",
+ # "Aha" / "Aha —" (insight beat)
+ r"\baha\b\s*[:—\-]?",
+ # "I see the X" / "I can see the X" — needs a *noun phrase* after,
+ # not "I see what you mean" (which is content). The
+ # required-following-article distinguishes them.
+ r"\bi\s+(?:see|can\s+see)\s+(?:the|a|an|my|our|this|that|these|those)\b",
+ # "Smoking gun" / "this is the root cause" / "this is the bug"
+ r"\bthis\s+is\s+the\s+(?:smoking\s+gun|root\s+cause|bug|"
+ r"real\s+issue|actual\s+issue|real\s+bug|actual\s+bug|"
+ r"core\s+issue|key\s+issue)\b",
+ # Trailing-realization beats (used for the trailing-sentence strip).
+ # "Got it." / "Got it —" / "Right," / "Right —" / "OK so" /
+ # "Okay so" / "Confirmed:". These are LESS risky to match in the
+ # leading pass too, because they only appear as sentence openers
+ # when the model is announcing its own conclusion. A user-facing
+ # answer doesn't start with "Got it."
+ r"\bgot\s+it\b\s*[:—\-.]?",
+ r"\bright\s*[,—\-]",
+ r"\b(?:ok|okay)\s+so\b",
+ r"\bconfirmed\s*[:.]",
+)
+
+# Compile once at import.
+# The boundary lookbehind ``(?:^|(?<=[\.\!\?]\s)|(?<=\n))`` matches the
+# opener at the *start* of the message OR right after a sentence-ending
+# punctuation + whitespace. That's what lets the consumer walk sentence
+# by sentence from the top.
+REASONING_PROSE_OPENERS_RE: re.Pattern[str] = re.compile(
+ r"(?i)(?:^|(?<=[\.\!\?]\s)|(?<=\n))"
+ r"\s*"
+ r"(?:" + "|".join(_REASONING_PROSE_OPENERS) + r")",
+ flags=re.UNICODE,
+)
+
+# A "starts with reasoning opener" pattern anchored to position 0.
+# Used by the stream-time consumer, which only ever looks at the start
+# of the message. This is the SAFE form — it can never match mid-sentence
+# because it requires position 0.
+STARTS_WITH_OPENER_RE: re.Pattern[str] = re.compile(
+ r"\s*(?:" + "|".join(_REASONING_PROSE_OPENERS) + r")",
+ flags=re.UNICODE | re.IGNORECASE,
+)
+
+# Sentence-end characters used to find the end of the preamble.
+# The lookahead ``(?=[A-Z"'`(\[]|\*\*[A-Z])`` requires the *next*
+# sentence to start with a capital / quote / parenthesis / bolded
+# capital — that keeps "Wait —" or "Right —" from matching their
+# own em-dash as a sentence boundary, and it keeps comma-followed
+# clauses from being treated as separate sentences.
+SENTENCE_END_RE: re.Pattern[str] = re.compile(
+ r"(?<=[\.\!\?])\s+(?=[A-Z\"'`\(\[]|\*\*[A-Z])|$|\n\s*\n",
+ re.UNICODE,
+)
diff --git a/cli.py b/cli.py
index c4d88449fc52..85966b4bed07 100644
--- a/cli.py
+++ b/cli.py
@@ -8477,6 +8477,41 @@ def _handle_skills_command(self, cmd: str):
from hermes_cli.skills_hub import handle_skills_slash
handle_skills_slash(cmd, ChatConsole())
+ def _handle_skill_approval_command(self, cmd: str, *, approve: bool) -> None:
+ """Approve or deny a pending guarded skill install from the CLI."""
+ parts = cmd.split(maxsplit=1)
+ approval_id = parts[1].strip() if len(parts) > 1 else ""
+ action = "skill-approve" if approve else "skill-deny"
+ if not approval_id:
+ print(f"Usage: /{action} ")
+ return
+ try:
+ if approve:
+ from tools.skill_approval_records import approve_skill_approval
+
+ record = approve_skill_approval(approval_id)
+ print(
+ f"✓ Installed skill '{record.get('skill_name')}' "
+ f"at {record.get('installed_path')}"
+ )
+ self._reload_skills()
+ else:
+ from tools.skill_approval_records import deny_skill_approval
+
+ record = deny_skill_approval(approval_id)
+ print(f"Denied skill approval {approval_id} ({record.get('skill_name')}).")
+ except Exception as exc:
+ print(f"Skill approval error: {exc}")
+
+ def _handle_approvals_command(self) -> None:
+ """List pending guarded skill install approvals."""
+ from tools.skill_approval_records import (
+ format_pending_skill_approvals,
+ list_pending_skill_approvals,
+ )
+
+ print(format_pending_skill_approvals(list_pending_skill_approvals()))
+
def _show_gateway_status(self):
"""Show status of the gateway and connected messaging platforms."""
from gateway.config import load_gateway_config, Platform
@@ -8789,6 +8824,12 @@ def process_command(self, command: str) -> bool:
elif canonical == "skills":
with self._busy_command(self._slow_command_status(cmd_original)):
self._handle_skills_command(cmd_original)
+ elif canonical == "skill-approve":
+ self._handle_skill_approval_command(cmd_original, approve=True)
+ elif canonical == "skill-deny":
+ self._handle_skill_approval_command(cmd_original, approve=False)
+ elif canonical == "approvals":
+ self._handle_approvals_command()
elif canonical == "platforms":
self._show_gateway_status()
elif canonical == "status":
diff --git a/gateway/builtin_hooks/mcp_http_autostart.py b/gateway/builtin_hooks/mcp_http_autostart.py
new file mode 100644
index 000000000000..f28179bd73aa
--- /dev/null
+++ b/gateway/builtin_hooks/mcp_http_autostart.py
@@ -0,0 +1,223 @@
+"""mcp_http_autostart — boot the Hermes MCP server on streamable-HTTP when the gateway starts.
+
+This is the first built-in gateway hook in the codebase. The hook reads
+``config.yaml`` keys under ``mcp_serve:`` and spawns the HTTP transport
+of the existing ``mcp_serve.py`` module in a daemon thread. The gateway
+itself stays free to handle messaging traffic.
+
+Config schema (``~/.hermes/config.yaml``):
+
+ mcp_serve:
+ enabled: true # default: false
+ transport: http # default: http (only http is wired here; stdio is opt-in)
+ host: 127.0.0.1 # default
+ port: 18950 # default
+ no_auth: false # default false. Set true ONLY for local dev.
+ log_level: info # default
+
+Env-var equivalents (override config when set):
+ HERMES_MCP_TRANSPORT, HERMES_MCP_HOST, HERMES_MCP_PORT, HERMES_MCP_API_KEY
+
+Failure modes are silent: if anything blows up we log to gateway.log and
+move on, so a broken MCP autostart never blocks gateway startup.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import sys
+import threading
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+logger = logging.getLogger("hermes.gateway.builtin.mcp_http_autostart")
+
+# Background thread bookkeeping for the autostarted MCP HTTP server.
+# The actual event loop is owned/managed by asyncio.run/uvicorn; we don't
+# attempt to track or expose it here to avoid misleading status. Earlier
+# versions kept a ``_loop`` reference assigned from inside the thread
+# target, but uvicorn never actually used it — so ``get_status()`` would
+# happily report ``loop_running=False`` on a perfectly healthy server.
+_thread: Optional[threading.Thread] = None
+_started: bool = False
+_start_lock = threading.Lock()
+_uvicorn_server: Optional[Any] = None
+_uvicorn_lock = threading.Lock()
+
+
+def _read_config() -> Dict[str, Any]:
+ """Pull the ``mcp_serve`` block out of config.yaml.
+
+ The actual signature of ``hermes_cli.config.cfg_get`` is
+ ``cfg_get(cfg: Optional[Dict], *keys, default=...)`` — the *first*
+ positional arg is the loaded config dict, not a key path. The previous
+ implementation called ``cfg_get("mcp_serve", default={})``, which
+ short-circuited through the ``isinstance(cfg, dict)`` guard and
+ silently returned the default every time. That meant the
+ ``mcp_serve:`` block in ``config.yaml`` was dead weight: the autostart
+ hook would only ever fire when ``HERMES_MCP_PORT`` (or another env
+ var) was set.
+
+ Fix: load the config dict first, then pass it as the first positional
+ arg with ``"mcp_serve"`` as the key path.
+ """
+ try:
+ from hermes_cli.config import load_config, cfg_get # type: ignore[import-not-found]
+
+ cfg = load_config()
+ block = cfg_get(cfg, "mcp_serve", default={})
+ return block if isinstance(block, dict) else {}
+ except Exception as exc: # pragma: no cover - hermes_cli may not be importable here
+ logger.debug("cfg_get failed: %s; falling back to env-only", exc)
+ return {}
+
+
+def _resolve() -> Dict[str, Any]:
+ """Compute final transport settings: config first, then env, then defaults."""
+ cfg = _read_config()
+ if not cfg.get("enabled", False) and not os.getenv("HERMES_MCP_PORT"):
+ return {"enabled": False}
+
+ return {
+ "enabled": True,
+ "transport": os.getenv("HERMES_MCP_TRANSPORT", cfg.get("transport", "http")),
+ "host": os.getenv("HERMES_MCP_HOST", cfg.get("host", "127.0.0.1")),
+ "port": int(os.getenv("HERMES_MCP_PORT", str(cfg.get("port", 18950)))),
+ "api_key": os.getenv("HERMES_MCP_API_KEY", cfg.get("api_key")),
+ "no_auth": bool(cfg.get("no_auth", False)) or os.getenv("HERMES_MCP_NO_AUTH") == "1",
+ "log_level": os.getenv("HERMES_MCP_LOG_LEVEL", cfg.get("log_level", "info")),
+ }
+
+
+def _server_thread_target(settings: Dict[str, Any]) -> None:
+ """Run ``mcp_serve.run_mcp_server()`` in a daemon thread.
+
+ We pass the resolved transport settings directly through
+ ``MCPServerSettings`` rather than mutating ``sys.argv`` — the latter
+ would leak the MCP transport args into the rest of the gateway process
+ (any code that reads ``sys.argv`` for logging or CLI helpers would see
+ them). The thread is daemon, so the gateway process can exit cleanly
+ even if uvicorn is mid-handshake.
+ """
+ try:
+ # Make sure we can import the upstream module from the gateway process.
+ # The repo root is on sys.path when running as a module, but the gateway
+ # may have been launched with a different cwd, so add it defensively.
+ repo_root = Path(__file__).resolve().parents[3]
+ if str(repo_root) not in sys.path:
+ sys.path.insert(0, str(repo_root))
+
+ import mcp_serve # type: ignore[import-not-found]
+ except Exception as exc:
+ logger.error("mcp_http_autostart: cannot import mcp_serve: %s", exc)
+ return
+
+ logger.info(
+ "mcp_http_autostart: starting MCP server on http://%s:%d (auth=%s)",
+ settings["host"], settings["port"],
+ "off" if settings["no_auth"] else "bearer",
+ )
+
+ try:
+ mcp_serve.run_mcp_server(
+ settings=mcp_serve.MCPServerSettings(
+ transport=settings["transport"],
+ host=settings["host"],
+ port=settings["port"],
+ path="/mcp",
+ api_key=settings.get("api_key"),
+ no_auth=settings.get("no_auth", False),
+ verbose=False,
+ )
+ )
+ except SystemExit as exc: # mcp_serve.py exits with code 2 on bad config
+ logger.warning("mcp_http_autostart: mcp_serve exited: %s", exc)
+ except Exception as exc:
+ logger.exception("mcp_http_autostart: mcp_serve crashed: %s", exc)
+ finally:
+ with _uvicorn_lock:
+ global _uvicorn_server
+ _uvicorn_server = None
+
+
+def register_uvicorn_server(server: Any) -> None:
+ """Optional hook for ``mcp_serve`` to register the live uvicorn server.
+
+ ``mcp_serve`` doesn't currently call this, but the public hook is here
+ so a future refactor (e.g. exposing the uvicorn ``Server`` from
+ ``_run_http``) can wire it without a cross-module API change. When
+ the server reference is set, ``get_status()`` will use its
+ ``started``/``should_exit`` flags instead of just thread liveness.
+ """
+ with _uvicorn_lock:
+ global _uvicorn_server
+ _uvicorn_server = server
+
+
+def register(gateway_runner: Any) -> None:
+ """Called by ``GatewayHooks._register_builtin_hooks()``.
+
+ Spawns the daemon thread that hosts the MCP HTTP server. Idempotent:
+ repeated calls are a no-op once the server is up.
+ """
+ global _started, _thread
+ with _start_lock:
+ if _started:
+ return
+
+ settings = _resolve()
+ if not settings.get("enabled"):
+ logger.info("mcp_http_autostart: disabled (set mcp_serve.enabled: true in config.yaml)")
+ return
+
+ if not settings.get("api_key") and not settings.get("no_auth"):
+ logger.warning(
+ "mcp_http_autostart: HTTP transport requires auth. Either set "
+ "mcp_serve.api_key in config.yaml (or HERMES_MCP_API_KEY env) "
+ "or set mcp_serve.no_auth: true for local-only testing. "
+ "Autostart will NOT start until this is fixed."
+ )
+ return
+
+ _thread = threading.Thread(
+ target=_server_thread_target,
+ args=(settings,),
+ name="mcp-http-autostart",
+ daemon=True,
+ )
+ _thread.start()
+ _started = True
+ logger.info("mcp_http_autostart: thread launched (PID lives in the gateway process)")
+
+
+def get_status() -> Dict[str, Any]:
+ """Introspection helper for ``/healthz`` or future ``/diagnostic`` commands.
+
+ Status is derived from:
+ * ``thread_alive`` — the daemon thread that hosts the MCP server.
+ * ``server_started`` — if a uvicorn ``Server`` has been registered
+ via :func:`register_uvicorn_server`, its ``started`` flag. This
+ is the load-bearing signal; a healthy uvicorn reports ``True``
+ after the bind completes, and ``False`` if it crashed or hasn't
+ bound yet.
+
+ Earlier revisions exposed a ``loop_running`` field that mirrored a
+ ``_loop`` variable assigned in the thread target — but uvicorn owns
+ the real event loop, so that field could report ``False`` on a
+ healthy server. Removed.
+ """
+ with _uvicorn_lock:
+ server_started: Optional[bool] = None
+ if _uvicorn_server is not None:
+ started_flag = getattr(_uvicorn_server, "started", None)
+ if isinstance(started_flag, bool):
+ server_started = started_flag
+ should_exit = getattr(_uvicorn_server, "should_exit", None)
+ if isinstance(should_exit, bool) and should_exit:
+ server_started = False
+ return {
+ "enabled": _started,
+ "thread_alive": bool(_thread and _thread.is_alive()),
+ "server_started": server_started,
+ }
diff --git a/gateway/hooks.py b/gateway/hooks.py
index 5ab45119202c..1157473dfee4 100644
--- a/gateway/hooks.py
+++ b/gateway/hooks.py
@@ -55,11 +55,23 @@ def loaded_hooks(self) -> List[dict]:
def _register_builtin_hooks(self) -> None:
"""Register built-in hooks that are always active.
- Currently empty — no shipped built-in hooks. Kept as the extension
- point for future always-on gateway hooks so they drop in without
- re-plumbing discover_and_load().
+ Hooks register themselves with the running gateway by calling the
+ register() function exported from the hook module. The hook is
+ responsible for failing silently on its own errors — we never want
+ a buggy hook to break gateway startup.
"""
- return
+ # MCP HTTP autostart — spawns the streamable-HTTP transport of
+ # mcp_serve.py in a daemon thread if config.yaml says so. See
+ # ``gateway/builtin_hooks/mcp_http_autostart.py`` for the schema.
+ try:
+ from gateway.builtin_hooks import mcp_http_autostart # type: ignore[import-not-found]
+
+ mcp_http_autostart.register(self._runner)
+ except Exception as exc: # pragma: no cover - never break startup
+ import logging
+ logging.getLogger("hermes.gateway.hooks").warning(
+ "builtin_hook mcp_http_autostart.register failed: %s", exc
+ )
def discover_and_load(self) -> None:
"""
diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py
index 89806a739312..01debbbfec32 100644
--- a/gateway/platforms/base.py
+++ b/gateway/platforms/base.py
@@ -1444,17 +1444,47 @@ class MessageEvent:
# Timestamps
timestamp: datetime = field(default_factory=datetime.now)
+ # Chat-syntax → canonical command-name mapping. Used to normalize
+ # `!`-prefixed guarded-skill-approval variants (which exist so they
+ # don't collide visually with platform slash-command menus) into the
+ # same routing namespace as their `/`-prefixed counterparts. Keep
+ # this aligned with the chat prompts the gateway prints to the user.
+ _BANG_COMMAND_ALIASES = {
+ "!approve": "skill-approve",
+ "!skill-approve": "skill-approve",
+ "!skill_approve": "skill-approve",
+ "!deny": "skill-deny",
+ "!skill-deny": "skill-deny",
+ "!skill_deny": "skill-deny",
+ "!approvals": "skill-approvals",
+ }
+
def is_command(self) -> bool:
"""Check if this is a command message (e.g., /new, /reset)."""
- return self.text.startswith("/")
-
+ if self.text.startswith("/"):
+ return True
+ # Guarded skill approvals are prompted in chat as !approve / !deny so
+ # they do not collide visually with platform slash-command menus.
+ first = self.text.split(maxsplit=1)[0].lower() if self.text else ""
+ return first in self._BANG_COMMAND_ALIASES
+
def get_command(self) -> Optional[str]:
"""Extract command name if this is a command message."""
if not self.is_command():
return None
- # Split on space and get first word, strip the /
parts = self.text.split(maxsplit=1)
- raw = parts[0][1:].lower() if parts else None
+ if not parts:
+ return None
+ token = parts[0]
+ # /-prefixed commands: strip the leading /, drop @botname suffix,
+ # reject anything that still contains a / (a path, not a command).
+ if token.startswith("/"):
+ raw = token[1:].lower()
+ else:
+ # !-prefixed guarded-skill aliases — map to canonical name.
+ raw = self._BANG_COMMAND_ALIASES.get(token.lower())
+ if raw is None:
+ return None
if raw and "@" in raw:
raw = raw.split("@", 1)[0]
# Reject file paths: valid command names never contain /
@@ -4718,6 +4748,22 @@ def truncate_message(
split_at = region.rfind("\n")
if split_at < _cp_limit // 2:
split_at = region.rfind(" ")
+ # Prefer sentence boundaries (". " "? " "! ") over the
+ # backtrack-to-space fallback so multi-sentence paragraphs
+ # don't get chopped mid-clause. Only consider a sentence
+ # boundary if it appears in the second half of the window
+ # (i.e. we're not giving up too much usable space) and
+ # after a backtrack-space has already been found. This
+ # preserves the "newline > space" priority for code blocks
+ # and lists while adding graceful sentence breaks for prose.
+ if split_at >= 1:
+ sentence_split = -1
+ for punct in (". ", "? ", "! "):
+ idx = region.rfind(punct)
+ if idx > sentence_split:
+ sentence_split = idx + 1 # keep the punctuation with the chunk
+ if sentence_split > _cp_limit // 2:
+ split_at = sentence_split
if split_at < 1:
split_at = _cp_limit
diff --git a/gateway/run.py b/gateway/run.py
index f11686ccd360..35f36ec4e4d4 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -286,15 +286,55 @@ def _looks_like_gateway_provider_error(text: str) -> bool:
def _sanitize_gateway_final_response(platform: Any, text: str) -> str:
- """Sanitize final gateway replies before sending them to high-noise chats.
+ """Sanitize final gateway replies before sending them to chat platforms.
Telegram is Bob's mobile inbox, so it should receive concise, safe provider
failure categories instead of raw HTTP bodies, request IDs, or policy text.
- Other platforms keep the existing behaviour for now.
+
+ For *all* chat platforms (Discord, Telegram, WhatsApp, Slack, etc.) we
+ also run the prose-level reasoning-leak stripper when the
+ ``display.strip_reasoning_prose`` config (or its per-platform
+ override) is enabled. This catches the "Let me check…", "Found it…",
+ "Now the real bug is…" preambles that some chat-tuned reasoning
+ models (notably minimax-m3 and the kimi-k2.5 family) emit as
+ natural-language sentences in their visible content. The strip is
+ opt-out: set the flag to ``false`` to preserve the model's prose
+ for platforms where transparency is desired (e.g. CLI / TUI).
"""
if not text:
return text
- if _gateway_platform_value(platform) != "telegram":
+
+ # Reasoning-prose strip is opt-in via display.strip_reasoning_prose.
+ # The gateway path is only reached for chat platforms (Discord,
+ # Telegram, WhatsApp, Slack, etc.) — CLI/terminal users see the
+ # model's raw output and can run with ``display.show_reasoning``.
+ platform_value = _gateway_platform_value(platform)
+ try:
+ from gateway.display_config import resolve_display_setting
+ _strip_prose_effective = resolve_display_setting(
+ _load_gateway_config(),
+ platform_value,
+ "strip_reasoning_prose",
+ True, # default ON — chat platforms want reasoning hidden
+ )
+ except Exception:
+ _strip_prose_effective = True
+ if _strip_prose_effective and platform_value != "cli":
+ try:
+ from agent.agent_runtime_helpers import strip_reasoning_prose
+ stripped = strip_reasoning_prose(None, text)
+ if stripped != text:
+ logger.debug(
+ "strip_reasoning_prose: platform=%s in_len=%d out_len=%d",
+ platform_value, len(text), len(stripped),
+ )
+ text = stripped
+ except Exception as _e:
+ logger.debug("strip_reasoning_prose failed: %s", _e)
+
+ text = _strip_skill_approval_sentinel(text)
+
+ if platform_value != "telegram":
return text
redacted = _redact_gateway_user_facing_secrets(str(text))
@@ -303,6 +343,30 @@ def _sanitize_gateway_final_response(platform: Any, text: str) -> str:
return redacted
+_SKILL_APPROVAL_SENTINEL_RE = re.compile(r"^USER_APPROVAL_REQUIRED:[A-Za-z0-9_]+(?:\n|$)")
+
+
+def _strip_skill_approval_sentinel(text: str) -> str:
+ """Drop the guarded-skill approval sentinel from chat-visible replies.
+
+ The sentinel is the load-bearing protocol between the agent and the
+ gateway: the agent emits ``USER_APPROVAL_REQUIRED:`` (typically
+ followed by a newline + a user-facing explanation) when a guarded
+ skill install needs human approval. The gateway intercepts the
+ sentinel to render an ``!approve `` / ``!deny `` prompt; the
+ sentinel itself must never reach the user's chat.
+
+ The trailing-newline anchor is intentional but optional — the
+ prose-leak stripper that runs upstream may consume the explanation
+ line and leave the sentinel as the entire response. In that case
+ the regex still matches the sentinel at end-of-string so the user
+ never sees a bare ``USER_APPROVAL_REQUIRED:sk1234_abc`` reply.
+ """
+ if not text:
+ return text
+ return _SKILL_APPROVAL_SENTINEL_RE.sub("", text, count=1)
+
+
def _prepare_gateway_status_message(platform: Any, event_type: str, message: str) -> Optional[str]:
"""Filter/sanitize agent status callbacks before platform delivery."""
text = str(message or "").strip()
@@ -7672,10 +7736,22 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
# The agent thread is blocked on a threading.Event inside
# tools/approval.py — sending an interrupt won't unblock it.
# Route directly to the approval handler so the event is signalled.
- if _cmd_def_inner and _cmd_def_inner.name in {"approve", "deny"}:
+ if _cmd_def_inner and _cmd_def_inner.name in {
+ "approve",
+ "deny",
+ "skill-approve",
+ "skill-deny",
+ "approvals",
+ }:
if _cmd_def_inner.name == "approve":
return await self._handle_approve_command(event)
- return await self._handle_deny_command(event)
+ if _cmd_def_inner.name == "deny":
+ return await self._handle_deny_command(event)
+ if _cmd_def_inner.name == "skill-approve":
+ return await self._handle_skill_approve_command(event)
+ if _cmd_def_inner.name == "skill-deny":
+ return await self._handle_skill_deny_command(event)
+ return await self._handle_approvals_command(event)
# /agents (/tasks alias) should be query-only and never interrupt.
if _cmd_def_inner and _cmd_def_inner.name == "agents":
@@ -8093,6 +8169,15 @@ async def _do_undo():
if canonical == "deny":
return await self._handle_deny_command(event)
+ if canonical == "skill-approve":
+ return await self._handle_skill_approve_command(event)
+
+ if canonical == "skill-deny":
+ return await self._handle_skill_deny_command(event)
+
+ if canonical == "approvals":
+ return await self._handle_approvals_command(event)
+
if canonical == "update":
return await self._handle_update_command(event)
@@ -14561,6 +14646,56 @@ def _reply_anchor_for_event(event: MessageEvent) -> Optional[str]:
_APPROVAL_TIMEOUT_SECONDS = 300 # 5 minutes
+ async def _handle_skill_approve_command(self, event: MessageEvent) -> str:
+ """Handle /skill-approve — install a pending guarded skill."""
+ approval_id = event.get_command_args().strip().split(maxsplit=1)[0] if event.get_command_args().strip() else ""
+ if not approval_id:
+ return "Usage: /skill-approve "
+ try:
+ from tools.skill_approval_records import approve_skill_approval
+
+ record = approve_skill_approval(approval_id)
+ except Exception as exc:
+ return f"Skill approval error: {exc}"
+
+ logger.info(
+ "User approved guarded skill install via gateway (id=%s, skill=%s)",
+ approval_id,
+ record.get("skill_name"),
+ )
+ return (
+ f"Installed skill `{record.get('skill_name')}`.\n"
+ f"Path: `{record.get('installed_path')}`"
+ )
+
+ async def _handle_skill_deny_command(self, event: MessageEvent) -> str:
+ """Handle /skill-deny — reject a pending guarded skill."""
+ approval_id = event.get_command_args().strip().split(maxsplit=1)[0] if event.get_command_args().strip() else ""
+ if not approval_id:
+ return "Usage: /skill-deny "
+ try:
+ from tools.skill_approval_records import deny_skill_approval
+
+ record = deny_skill_approval(approval_id)
+ except Exception as exc:
+ return f"Skill approval error: {exc}"
+
+ logger.info(
+ "User denied guarded skill install via gateway (id=%s, skill=%s)",
+ approval_id,
+ record.get("skill_name"),
+ )
+ return f"Denied skill approval `{approval_id}` ({record.get('skill_name')})."
+
+ async def _handle_approvals_command(self, event: MessageEvent) -> str:
+ """Handle /approvals — list pending guarded skill records."""
+ from tools.skill_approval_records import (
+ format_pending_skill_approvals,
+ list_pending_skill_approvals,
+ )
+
+ return format_pending_skill_approvals(list_pending_skill_approvals())
+
async def _handle_approve_command(self, event: MessageEvent) -> Optional[str]:
"""Handle /approve command — unblock waiting agent thread(s).
@@ -14589,6 +14724,9 @@ async def _handle_approve_command(self, event: MessageEvent) -> Optional[str]:
)
if not has_blocking_approval(session_key):
+ args = event.get_command_args().strip()
+ if args and args.split(maxsplit=1)[0].startswith("sk"):
+ return await self._handle_skill_approve_command(event)
if session_key in self._pending_approvals:
self._pending_approvals.pop(session_key)
return t("gateway.approval_expired")
@@ -14635,6 +14773,9 @@ async def _handle_deny_command(self, event: MessageEvent) -> str:
)
if not has_blocking_approval(session_key):
+ args = event.get_command_args().strip()
+ if args and args.split(maxsplit=1)[0].startswith("sk"):
+ return await self._handle_skill_deny_command(event)
if session_key in self._pending_approvals:
self._pending_approvals.pop(session_key)
return t("gateway.deny.stale")
diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py
index 16d1cecd31aa..4cbfb72fa277 100644
--- a/gateway/stream_consumer.py
+++ b/gateway/stream_consumer.py
@@ -302,6 +302,69 @@ def finish(self) -> None:
# response (run_agent.py _strip_think_blocks), but the stream
# consumer sends intermediate edits before that stripping happens.
+ # ── Reasoning-prose filtering ───────────────────────────────────
+ # Chat-tuned reasoning models (minimax-m3, kimi-k2.5/2.6) emit their
+ # chain-of-thought as natural-language sentences directly in the
+ # content field instead of using XML tags. Example mid-stream
+ # deltas: "Let me check what the gateway is doing.", " Found it.",
+ # " Now the real bug surface…". Strip the same opener patterns
+ # that ``agent.agent_runtime_helpers.strip_reasoning_prose`` would
+ # catch at final-response time, so gateway users don't see the
+ # reasoning as it streams. The same opening-verb regex is used
+ # for boundary-aware matching.
+ _REASONING_PROSE_OPENERS_RE = None
+ _SENTENCE_END_RE = None
+ _REASONING_PROSE_BUFFER = ""
+
+ @classmethod
+ def _ensure_prose_patterns(cls) -> None:
+ """Lazy-load the prose-opener regexes from the shared helpers module.
+
+ We import from ``agent.reasoning_prose`` (the public surface) rather
+ than the underscored private names in ``agent_runtime_helpers`` so
+ the dependency is visible in the import graph and renaming the
+ underlying regex won't silently break the stream-time stripper.
+ """
+ if cls._REASONING_PROSE_OPENERS_RE is None:
+ from agent.reasoning_prose import (
+ REASONING_PROSE_OPENERS_RE as _prose_re,
+ SENTENCE_END_RE as _se_re,
+ )
+ cls._REASONING_PROSE_OPENERS_RE = _prose_re
+ cls._SENTENCE_END_RE = _se_re
+
+ def _strip_prose_prefix(self, text: str) -> str:
+ """If *text* starts with one or more reasoning-opener sentences,
+ drop them and return the remainder. Otherwise return *text*
+ unchanged. This is the per-delta analogue of
+ ``strip_reasoning_prose``'s leading-preamble pass; the
+ trailing-sentence pass is left to the final-response chokepoint
+ because it needs full content to make the call.
+ """
+ self._ensure_prose_patterns()
+ if not text:
+ return text
+ out = text
+ # Bound the loop — anything past 8 sentences of pure reasoning
+ # is almost certainly a model stuck in a loop, not a preamble.
+ for _ in range(8):
+ m = self._REASONING_PROSE_OPENERS_RE.match(out)
+ if not m:
+ return out
+ opener_end = m.end()
+ boundary = self._SENTENCE_END_RE.search(out, opener_end)
+ if not boundary:
+ return out
+ candidate = (
+ out[:m.start()].rstrip()
+ + " "
+ + out[boundary.end():].lstrip()
+ ).strip()
+ if len(candidate) < 8 or candidate == out:
+ return out
+ out = candidate
+ return out
+
def _filter_and_accumulate(self, text: str) -> None:
"""Add a text delta to the accumulated buffer, suppressing think blocks.
@@ -309,10 +372,28 @@ def _filter_and_accumulate(self, text: str) -> None:
reasoning/thinking block. Text inside such blocks is silently
discarded. Partial tags at buffer boundaries are held back in
``_think_buffer`` until enough characters arrive to decide.
+
+ Also strips leading reasoning-prose sentences on the FIRST
+ delta of a turn (when ``_accumulated`` is empty). Once the
+ assistant has started streaming a real answer, the prose
+ detector is disabled for the rest of the turn to avoid
+ chopping a real sentence that happens to start with "Let me…"
+ (a common, legitimate English construction). The
+ final-response chokepoint in
+ ``_sanitize_gateway_final_response`` still runs the
+ trailing-sentence pass on the full text.
"""
buf = self._think_buffer + text
self._think_buffer = ""
+ # First-delta reasoning-prose strip — only fires at the start
+ # of a fresh turn. The stripper is permissive: it only
+ # removes text when the first sentence(s) clearly match the
+ # opener set AND a sentence boundary is present, so a model
+ # that opens with a clean "Here's the answer." is untouched.
+ if not self._accumulated and not self._in_think_block and buf:
+ buf = self._strip_prose_prefix(buf)
+
while buf:
if self._in_think_block:
# Look for the earliest closing tag
diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py
index 34e3af4014b6..3ae907f436b7 100644
--- a/hermes_cli/commands.py
+++ b/hermes_cli/commands.py
@@ -97,6 +97,11 @@ class CommandDef:
gateway_only=True, args_hint="[session|always]"),
CommandDef("deny", "Deny a pending dangerous command", "Session",
gateway_only=True),
+ CommandDef("skill-approve", "Approve and install a pending guarded skill", "Session",
+ aliases=("skill_approve",), args_hint=""),
+ CommandDef("skill-deny", "Deny a pending guarded skill install", "Session",
+ aliases=("skill_deny",), args_hint=""),
+ CommandDef("approvals", "List pending guarded skill approvals", "Session"),
CommandDef("background", "Run a prompt in the background", "Session",
aliases=("bg", "btw"), args_hint=""),
CommandDef("agents", "Show active agents and running tasks", "Session",
@@ -337,6 +342,7 @@ def is_gateway_known_command(name: str | None) -> bool:
{
"agents",
"approve",
+ "approvals",
"background",
"commands",
"deny",
@@ -345,6 +351,8 @@ def is_gateway_known_command(name: str | None) -> bool:
"profile",
"queue",
"restart",
+ "skill-approve",
+ "skill-deny",
"status",
"steer",
"stop",
diff --git a/mcp_serve.py b/mcp_serve.py
index 5ae0261d9af7..54299fdcd70a 100644
--- a/mcp_serve.py
+++ b/mcp_serve.py
@@ -36,7 +36,7 @@
import sys
import threading
import time
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
from pathlib import Path
from typing import Dict, List, Optional
@@ -863,8 +863,71 @@ def permissions_respond(
# Entry point
# ---------------------------------------------------------------------------
-def run_mcp_server(verbose: bool = False) -> None:
- """Start the Hermes MCP server on stdio."""
+# ---------------------------------------------------------------------------
+# Public settings surface — lets callers (e.g. mcp_http_autostart) pass
+# transport settings directly without touching process-global ``sys.argv``.
+# Keeping the CLI parser as the user-facing entrypoint and the keyword args
+# as the programmatic one means in-process embedders (the autostart hook)
+# don't mutate argv and risk side effects elsewhere in the gateway.
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class MCPServerSettings:
+ """Programmatic transport settings for :func:`run_mcp_server`."""
+
+ transport: str = "stdio"
+ host: str = "127.0.0.1"
+ port: int = 18950
+ path: str = "/mcp"
+ api_key: Optional[str] = None
+ no_auth: bool = False
+ verbose: bool = False
+
+
+def _settings_from_argv(argv: List[str]) -> MCPServerSettings:
+ """Bridge the CLI parser into the programmatic settings object."""
+ (
+ transport,
+ host,
+ port,
+ path,
+ api_key,
+ no_auth,
+ verbose,
+ ) = _parse_transport_args(argv)
+ return MCPServerSettings(
+ transport=transport,
+ host=host,
+ port=port,
+ path=path,
+ api_key=api_key,
+ no_auth=no_auth,
+ verbose=verbose,
+ )
+
+
+def run_mcp_server(
+ verbose: bool = False,
+ *,
+ settings: Optional[MCPServerSettings] = None,
+) -> None:
+ """Start the Hermes MCP server (stdio or streamable-HTTP).
+
+ Transports:
+ - ``stdio`` (default): one MCP client per process, speaking JSON-RPC on
+ stdin/stdout. Same behaviour as the original `hermes mcp serve`.
+ - ``http``: serve the FastMCP ASGI app on host:port. Multiple clients
+ can connect; requires a bearer token (or the explicit ``--no-auth``
+ flag for local-only testing).
+
+ Selection: pass ``--transport http`` (CLI), set
+ ``HERMES_MCP_TRANSPORT=http`` (env), set ``HERMES_MCP_PORT`` to enable
+ HTTP at the given port with stdio as the fallback default, or call
+ ``run_mcp_server(settings=MCPServerSettings(...))`` directly (the path
+ in-process embedders like the gateway autostart hook should use — it
+ avoids mutating the host process's ``sys.argv``).
+ """
if not _MCP_SERVER_AVAILABLE:
print(
"Error: MCP server requires the 'mcp' package.\n"
@@ -873,21 +936,55 @@ def run_mcp_server(verbose: bool = False) -> None:
)
sys.exit(1)
- if verbose:
+ if settings is None:
+ # CLI / __main__ path — parse the user's argv. The ``verbose`` kwarg
+ # is honored as a fallback for callers that don't go through
+ # settings but do want to override argv-based detection.
+ settings = _settings_from_argv(sys.argv[1:])
+ if verbose and not settings.verbose:
+ object.__setattr__(settings, "verbose", True)
+ settings = replace(settings, verbose=True)
+
+ if settings.verbose:
logging.basicConfig(level=logging.DEBUG, stream=sys.stderr)
else:
logging.basicConfig(level=logging.WARNING, stream=sys.stderr)
+ # ---- transport selection ---------------------------------------------
+ transport = settings.transport
+ http_host = settings.host
+ http_port = settings.port
+ http_path = settings.path
+ api_key = settings.api_key
+ no_auth = settings.no_auth
+
bridge = EventBridge()
bridge.start()
server = create_mcp_server(event_bridge=bridge)
+ # ---- optional ari.* tool registration --------------------------------
+ # Always import-and-try; skip silently if the ari tools file isn't present
+ # (allows the stdio entrypoint to keep working when only the messaging
+ # server is bundled).
+ try:
+ from mcp_serve_ari_tools import register_ari_tools # type: ignore[import-not-found]
+
+ register_ari_tools(server)
+ logger.info("Registered ari.* MCP tools (asterisk ARI)")
+ except ImportError:
+ pass
+ except Exception as exc: # pragma: no cover
+ logger.warning("Failed to register ari.* tools: %s", exc)
+
import asyncio
- async def _run():
+ async def _run() -> None:
try:
- await server.run_stdio_async()
+ if transport == "stdio":
+ await server.run_stdio_async()
+ else: # "http"
+ await _run_http(server, http_host, http_port, http_path, api_key, no_auth)
finally:
bridge.stop()
@@ -895,3 +992,298 @@ async def _run():
asyncio.run(_run())
except KeyboardInterrupt:
bridge.stop()
+
+
+# ---------------------------------------------------------------------------
+# Transport plumbing
+# ---------------------------------------------------------------------------
+
+
+def _parse_transport_args(argv: List[str]) -> tuple:
+ """Parse ``[--transport, --host, --port, --path, --api-key, --no-auth, --verbose]``.
+
+ Returns: ``(transport, host, port, path, api_key, no_auth, verbose)``
+ """
+ import argparse
+
+ parser = argparse.ArgumentParser(
+ prog="hermes-mcp-serve",
+ description=(
+ "Expose Hermes as an MCP server. "
+ "Default transport is stdio (one client per process). "
+ "Use --transport http to expose the streamable-HTTP ASGI app."
+ ),
+ add_help=True,
+ )
+ parser.add_argument(
+ "--transport",
+ choices=("stdio", "http"),
+ default=os.getenv("HERMES_MCP_TRANSPORT", "stdio"),
+ help="MCP transport. stdio (default) or http.",
+ )
+ parser.add_argument(
+ "--host",
+ default=os.getenv("HERMES_MCP_HOST", "127.0.0.1"),
+ help="HTTP bind host (default 127.0.0.1).",
+ )
+ parser.add_argument(
+ "--port",
+ type=int,
+ default=int(os.getenv("HERMES_MCP_PORT", "0") or 0),
+ help=(
+ "HTTP bind port. If 0 (default) the stdio transport is used. "
+ "Set HERMES_MCP_PORT=18950 to expose HTTP at the canonical port."
+ ),
+ )
+ parser.add_argument(
+ "--path",
+ default=os.getenv("HERMES_MCP_PATH", "/mcp"),
+ help="HTTP mount path (default /mcp).",
+ )
+ parser.add_argument(
+ "--api-key",
+ default=os.getenv("HERMES_MCP_API_KEY"),
+ help=(
+ "Bearer token required on HTTP requests. Falls back to "
+ "HERMES_MCP_API_KEY env. If unset, HTTP refuses to start unless "
+ "--no-auth is given."
+ ),
+ )
+ parser.add_argument(
+ "--no-auth",
+ action="store_true",
+ help="Disable bearer-token auth on the HTTP transport (LOCAL ONLY).",
+ )
+ parser.add_argument(
+ "--verbose",
+ "-v",
+ action="store_true",
+ help="Verbose logging (debug).",
+ )
+ # When --port is provided non-zero, force transport=http
+ args = parser.parse_args(argv)
+ if args.port and args.transport == "stdio":
+ args.transport = "http"
+ if args.transport == "http" and not args.api_key and not args.no_auth:
+ print(
+ "Error: HTTP transport requires --api-key or HERMES_MCP_API_KEY "
+ "(or pass --no-auth for local-only testing).",
+ file=sys.stderr,
+ )
+ sys.exit(2)
+ return (
+ args.transport,
+ args.host,
+ args.port or 18950,
+ args.path,
+ args.api_key,
+ args.no_auth,
+ args.verbose,
+ )
+
+
+async def _run_http(
+ server: "FastMCP",
+ host: str,
+ port: int,
+ path: str,
+ api_key: Optional[str],
+ no_auth: bool,
+) -> None:
+ """Serve FastMCP over streamable-HTTP, optionally behind a bearer-token ASGI middleware.
+
+ Also serves:
+ - ``GET /.well-known/mcp.json`` — MCP discovery document (spec §3.1).
+ Conforming MCP clients (Claude Desktop >=0.7, Cursor >=0.40, recent
+ Codex) auto-detect this and prompt to connect.
+ - ``GET /healthz`` — liveness probe (always 200, no auth).
+ """
+ try:
+ # FastMCP ships the ASGI app directly (mcp>=1.9)
+ mcp_asgi = server.streamable_http_app() # type: ignore[attr-defined]
+ except AttributeError: # pragma: no cover
+ # Older FastMCP: use run_streamable_http_async instead
+ await server.run_streamable_http_async() # type: ignore[attr-defined]
+ return
+
+ # Build the parent app: discovery + healthz on the root, MCP at /mcp.
+ from starlette.applications import Starlette
+ from starlette.responses import JSONResponse
+ from starlette.routing import Mount, Route
+
+ async def _wellknown_mcp(_request):
+ """MCP discovery document per modelcontextprotocol.io spec §3.1.
+
+ The ``transports`` block advertises the streamable-HTTP endpoint so
+ clients know where to connect without manual config.
+ """
+ return JSONResponse(
+ {
+ "mcp_version": "2025-03-26",
+ "server": {
+ "name": "hermes",
+ "version": "1.26.0",
+ "description": (
+ "Hermes Agent messaging bridge + Asterisk ARI call control. "
+ "Use these tools to read/write conversations across connected "
+ "platforms and to control live phone calls via ari.* tools."
+ ),
+ },
+ "capabilities": {
+ "tools": {"listChanged": False},
+ "resources": {"subscribe": False, "listChanged": False},
+ "prompts": {"listChanged": False},
+ },
+ "transports": {
+ "streamable-http": {
+ "endpoint": "/mcp",
+ "auth": {"type": "bearer"} if not no_auth else None,
+ }
+ },
+ "tools_hint": [
+ "conversations_list",
+ "messages_read",
+ "messages_send",
+ "channels_list",
+ "events_poll",
+ "ari.answer",
+ "ari.hangup",
+ "ari.play_tts",
+ "ari.transfer",
+ "ari.dial",
+ ],
+ },
+ headers={"Cache-Control": "no-store"},
+ )
+
+ async def _healthz(_request):
+ return JSONResponse({"status": "ok", "transport": "http", "ari_tools": _has_ari_tools()})
+
+ # If the user's path is "/mcp" (default), mount the MCP app under /mcp and
+ # expose discovery at the root. If they picked a custom path, mount it there.
+ mount_path = path if path.startswith("/") else f"/{path}"
+ # Strip trailing slash so Mount("/mcp", ...) and Mount("/mcp/", ...) both work.
+ mount_path = mount_path.rstrip("/") or "/mcp"
+
+ parent = Starlette(
+ routes=[
+ Route("/.well-known/mcp.json", _wellknown_mcp),
+ Route("/.well-known/mcp", _wellknown_mcp), # alias some clients look for
+ Route("/healthz", _healthz),
+ Mount(mount_path, app=mcp_asgi),
+ ]
+ )
+
+ wrapped = _BearerAuthMiddleware(parent, expected_token=api_key, disabled=no_auth)
+
+ try:
+ import uvicorn # type: ignore[import-not-found]
+ except ImportError as exc: # pragma: no cover
+ print(
+ f"Error: HTTP transport requires 'uvicorn': {exc}",
+ file=sys.stderr,
+ )
+ sys.exit(1)
+
+ config = uvicorn.Config(
+ app=wrapped,
+ host=host,
+ port=port,
+ log_level="info",
+ access_log=False,
+ )
+ uvi = uvicorn.Server(config)
+ logger.info(
+ "Hermes MCP server listening on http://%s:%d (auth=%s, ari_tools=%s, "
+ "discovery at /.well-known/mcp.json, mcp at %s)",
+ host,
+ port,
+ "off" if no_auth else "bearer",
+ _has_ari_tools(),
+ mount_path,
+ )
+ await uvi.serve()
+
+
+def _has_ari_tools() -> bool:
+ try:
+ import mcp_serve_ari_tools # type: ignore[import-not-found]
+
+ return hasattr(mcp_serve_ari_tools, "register_ari_tools")
+ except ImportError:
+ return False
+
+
+class _BearerAuthMiddleware:
+ """ASGI middleware that requires ``Authorization: Bearer `` on HTTP transports.
+
+ Stdio and local-loopback tests can opt out with ``--no-auth``. The token
+ comparison is constant-time to avoid timing oracles.
+ """
+
+ def __init__(self, app, *, expected_token: Optional[str], disabled: bool) -> None:
+ self._app = app
+ self._expected = (expected_token or "").encode("utf-8") if expected_token else b""
+ self._disabled = disabled
+
+ async def __call__(self, scope, receive, send):
+ if scope["type"] != "http":
+ # WebSocket / lifespan scopes: pass through unchanged
+ await self._app(scope, receive, send)
+ return
+ # Public, unauthenticated endpoints. The spec requires discovery to
+ # be reachable without credentials, and healthz must work behind a
+ # load balancer / k8s probe that doesn't have our bearer token.
+ path = scope.get("path", "")
+ if path in ("/healthz", "/.well-known/mcp.json", "/.well-known/mcp"):
+ await self._app(scope, receive, send)
+ return
+ if self._disabled or not self._expected:
+ await self._app(scope, receive, send)
+ return
+ # Extract Authorization header (case-insensitive)
+ auth_value: Optional[bytes] = None
+ for k, v in scope.get("headers", []):
+ if k.lower() == b"authorization":
+ auth_value = v
+ break
+ if not auth_value or not auth_value.startswith(b"Bearer "):
+ await self._reject(send, status=401, reason="Missing bearer token")
+ return
+ presented = auth_value[len(b"Bearer "):]
+ if not _constant_time_eq(presented, self._expected):
+ await self._reject(send, status=403, reason="Invalid bearer token")
+ return
+ await self._app(scope, receive, send)
+
+ @staticmethod
+ async def _reject(send, *, status: int, reason: str) -> None:
+ body = json.dumps({"error": reason}).encode("utf-8")
+ await send(
+ {
+ "type": "http.response.start",
+ "status": status,
+ "headers": [
+ (b"content-type", b"application/json"),
+ (b"content-length", str(len(body)).encode("ascii")),
+ ],
+ }
+ )
+ await send({"type": "http.response.body", "body": body})
+
+
+def _constant_time_eq(a: bytes, b: bytes) -> bool:
+ """Constant-time bytes comparison (avoid timing oracles)."""
+ import hmac
+
+ return hmac.compare_digest(a, b)
+
+
+# ---------------------------------------------------------------------------
+# Script entrypoint
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+ # Allow direct invocation: `python mcp_serve.py [--transport http|stdio ...]`
+ # The `hermes mcp serve` subcommand in hermes_cli invokes run_mcp_server() directly.
+ run_mcp_server(verbose=False)
diff --git a/mcp_serve_ari_tools.py b/mcp_serve_ari_tools.py
new file mode 100644
index 000000000000..f088beab9473
--- /dev/null
+++ b/mcp_serve_ari_tools.py
@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import os
+import json
+import logging
+import httpx
+from typing import Any, Dict, Optional
+from mcp.server.fastmcp import FastMCP
+
+logger = logging.getLogger("hermes.mcp_ari")
+
+# --- Configuration ---
+# Defaults to a localhost Asterisk ARI listener. Override via env vars in
+# production. Default creds are *only* the standard ARI default — operators
+# should override in their .env.
+ARI_URL = os.environ.get("ASTERISK_ARI_URL", "http://127.0.0.1:8088")
+ARI_USER = os.environ.get("ASTERISK_ARI_USER", "admin")
+ARI_PASS = os.environ.get("ASTERISK_ARI_PASS", "admin")
+
+# Optional external TTS endpoint used by ``ari.play_tts``. POST the ``text`` to
+# ``{TTS_URL}`` and expect an audio file back (Content-Type audio/*) or a JSON
+# body with a ``media_uri`` field. If unset, ``ari.play_tts`` requires the
+# caller to pass ``media_uri`` directly. This keeps ARI dependency-free of any
+# particular TTS provider while still letting tools synthesize speech.
+TTS_URL = os.environ.get("ASTERISK_TTS_URL") # e.g. http://127.0.0.1:8089/synth
+
+
+def _ari_request(method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+ """Internal helper to call the Asterisk ARI REST API.
+
+ ARI HTTP semantics:
+ * Path parameters (``{channelId}``, ``{playbackId}``) belong in the URL.
+ * Operation parameters (e.g. ``media`` for ``/play``) are query string
+ params on GET/POST, not JSON bodies. ARI is form-encoded in practice.
+ * ``/play`` (NOT ``/playback`` — that's a sub-resource) is the channel
+ operation that creates a new Playback object and starts streaming
+ media to the channel. See Asterisk docs: "Almost all media is played
+ to a channel using the POST /channels/{channel_id}/play operation."
+ """
+ url = f"{ARI_URL}/ari{endpoint}"
+ # ARI expects operation params on the query string, not the body.
+ merged_params: Dict[str, Any] = {}
+ if params:
+ merged_params.update(params)
+ if data:
+ # If the caller passed a body (e.g. originate ``variables``), send it
+ # as form data; otherwise, the per-operation spec wants query params.
+ if method.upper() in ("POST", "PUT", "PATCH") and endpoint in ("/channels",):
+ merged_params.update(data)
+ else:
+ merged_params.update(data)
+ try:
+ with httpx.Client(auth=(ARI_USER, ARI_PASS), timeout=5.0) as client:
+ response = client.request(method, url, params=merged_params)
+ response.raise_for_status()
+ return {
+ "status": "success",
+ "message": f"ARI {method} {endpoint} successful",
+ "data": response.json() if response.content else {},
+ }
+ except httpx.HTTPStatusError as e:
+ return {
+ "status": "error",
+ "message": f"ARI API returned {e.response.status_code}",
+ "error": e.response.text,
+ }
+ except Exception as e:
+ return {
+ "status": "error",
+ "message": f"ARI request failed: {str(e)}",
+ }
+
+
+def _synthesize_tts(text: str, voice: str) -> Dict[str, Any]:
+ """Best-effort TTS via an external HTTP endpoint.
+
+ Two response shapes are accepted:
+ 1. ``Content-Type: audio/*`` — the response body is the audio file
+ and we return it as a ``data:`` URI ARI can play directly.
+ 2. ``application/json`` with a ``media_uri`` field — the upstream
+ service already stored the audio and returns an ``http:`` or
+ ``sound:`` URI.
+
+ Returns a dict with keys ``ok``, ``media_uri``, and (on failure) ``error``.
+ """
+ # Re-read the env var on every call (instead of using the
+ # module-level TTS_URL constant) so operators can flip the service
+ # on or off at runtime, and so test fixtures can toggle it without
+ # re-importing the module.
+ tts_url = os.environ.get("ASTERISK_TTS_URL")
+ if not tts_url:
+ return {"ok": False, "error": "ASTERISK_TTS_URL not set; pass media_uri explicitly"}
+ try:
+ with httpx.Client(timeout=10.0) as client:
+ r = client.post(tts_url, params={"text": text, "voice": voice})
+ r.raise_for_status()
+ ct = r.headers.get("content-type", "")
+ if ct.startswith("audio/"):
+ import base64
+ encoded = base64.b64encode(r.content).decode("ascii")
+ return {"ok": True, "media_uri": f"data:{ct};base64,{encoded}"}
+ # Try JSON shape
+ try:
+ body = r.json()
+ except Exception:
+ return {"ok": False, "error": f"TTS returned non-audio, non-JSON: {ct}"}
+ uri = body.get("media_uri") or body.get("uri")
+ if not uri:
+ return {"ok": False, "error": "TTS JSON missing media_uri/uri"}
+ return {"ok": True, "media_uri": uri}
+ except Exception as e:
+ return {"ok": False, "error": f"TTS request failed: {e}"}
+
+
+def register_ari_tools(mcp: FastMCP):
+ """Register ARI-related tools to the provided FastMCP server."""
+
+ def ari_answer(channel_id: str) -> str:
+ """Answer a ringing channel in Asterisk."""
+ res = _ari_request("POST", f"/channels/{channel_id}/answer")
+ return json.dumps(res)
+
+ def ari_hangup(channel_id: str) -> str:
+ """Hang up a channel in Asterisk.
+
+ Optional ``reason_code`` query param (defaults to 16 = normal clearing).
+ """
+ res = _ari_request("POST", f"/channels/{channel_id}/hangup", params={"reason_code": 16})
+ return json.dumps(res)
+
+ def ari_play_tts(channel_id: str, text: str, voice: str = "alice", media_uri: Optional[str] = None) -> str:
+ """Play text-to-speech on a channel.
+
+ ARI has no native TTS — ``POST /channels/{id}/play`` only accepts a
+ pre-existing ``media`` URI. This tool:
+
+ 1. If the caller passes ``media_uri`` directly, plays it verbatim.
+ 2. Otherwise, calls the external TTS endpoint at ``ASTERISK_TTS_URL``
+ and uses its returned ``media_uri`` (or its raw audio body,
+ base64-encoded into a ``data:`` URI).
+ 3. Falls back to a clear error if neither is available.
+
+ Args:
+ channel_id: The ARI channel id (e.g. from ``ari.dial``).
+ text: The text to speak. Ignored if ``media_uri`` is set.
+ voice: TTS voice hint passed to the upstream service.
+ media_uri: Skip TTS and play this URI directly. Must be an ARI-
+ compatible scheme (``sound:``, ``http(s):``, ``data:``).
+ """
+ if not media_uri:
+ synth = _synthesize_tts(text, voice)
+ if not synth.get("ok"):
+ return json.dumps({
+ "status": "error",
+ "message": "TTS synthesis failed; pass media_uri to skip TTS",
+ "error": synth.get("error"),
+ })
+ media_uri = synth["media_uri"]
+ # /play (not /playback — /playback is a sub-resource of a Playback obj)
+ res = _ari_request(
+ "POST",
+ f"/channels/{channel_id}/play",
+ params={"media": media_uri},
+ )
+ return json.dumps(res)
+
+ def ari_stop_playback(playback_id: str) -> str:
+ """Stop an in-progress media playback.
+
+ The complementary operation to ``ari.play_tts``: once a Playback
+ object exists, ``DELETE /playbacks/{id}`` stops the stream. Exposed
+ separately because the caller needs the playback id returned from
+ the play call, not the channel id.
+ """
+ res = _ari_request("DELETE", f"/playbacks/{playback_id}")
+ return json.dumps(res)
+
+ def ari_transfer(channel_id: str, target: str) -> str:
+ """Transfer a channel to another target (extension or context/exten)."""
+ res = _ari_request(
+ "POST",
+ f"/channels/{channel_id}/redirect",
+ params={"endpoint": target},
+ )
+ return json.dumps(res)
+
+ def ari_dial(target: str, variables: Optional[Dict[str, str]] = None, app: str = "bridge", caller_id: Optional[str] = None) -> str:
+ """Originate a call to a target.
+
+ Args:
+ target: ARI endpoint string (e.g. ``PJSIP/1001``, ``SIP/peer``).
+ variables: Channel variables to set at originate time
+ (e.g. ``{"CALLERID(name)": "Hermes"}``).
+ app: Stasis application the new channel will route into. Defaults
+ to ``bridge`` — override to your dialplan app if you have one.
+ caller_id: Optional caller-id name to present.
+ """
+ params: Dict[str, Any] = {"endpoint": target, "app": app}
+ if caller_id:
+ params["callerId"] = caller_id
+ if variables:
+ for k, v in variables.items():
+ params[f"variable_{k}"] = v
+ res = _ari_request("POST", "/channels", params=params)
+ return json.dumps(res)
+
+ def ari_list_channels() -> str:
+ """List currently active channels (debugging / dashboard use)."""
+ res = _ari_request("GET", "/channels")
+ return json.dumps(res)
+
+ mcp.add_tool(ari_answer, name="ari.answer", description="Answer a ringing channel in Asterisk.")
+ mcp.add_tool(ari_hangup, name="ari.hangup", description="Hang up a channel in Asterisk (reason_code=16).")
+ mcp.add_tool(ari_play_tts, name="ari.play_tts", description="Play TTS on a channel. ARI has no native TTS — pass media_uri or set ASTERISK_TTS_URL.")
+ mcp.add_tool(ari_stop_playback, name="ari.stop_playback", description="Stop a running Playback by its id.")
+ mcp.add_tool(ari_transfer, name="ari.transfer", description="Redirect a channel to another endpoint (uses ARI /redirect).")
+ mcp.add_tool(ari_dial, name="ari.dial", description="Originate a call to an ARI endpoint.")
+ mcp.add_tool(ari_list_channels, name="ari.list_channels", description="List currently active ARI channels.")
diff --git a/tests/gateway/test_platform_base.py b/tests/gateway/test_platform_base.py
index 10a924764ab4..d595f3b32ccf 100644
--- a/tests/gateway/test_platform_base.py
+++ b/tests/gateway/test_platform_base.py
@@ -72,6 +72,57 @@ def test_slash_only(self):
assert event.is_command() is True
+class TestMessageEventBangCommands:
+ """`!`-prefixed guarded-skill aliases must route through `get_command`.
+
+ The chat-syntax mapping in ``gateway.platforms.base`` is the bridge
+ between user-typed ``!approve`` / ``!deny`` tokens and the
+ canonical ``skill-approve`` / ``skill-deny`` slash-command form.
+ The original ``is_command()`` accepted these tokens but
+ ``get_command()`` returned ``None`` for them — the agent would see
+ "this is a command" and then fail to parse it. These tests pin
+ the corrected behavior.
+ """
+
+ def test_bang_approve_is_command(self):
+ event = MessageEvent(text="!approve")
+ assert event.is_command() is True
+
+ def test_bang_deny_is_command(self):
+ event = MessageEvent(text="!deny")
+ assert event.is_command() is True
+
+ def test_bang_approvals_is_command(self):
+ event = MessageEvent(text="!approvals")
+ assert event.is_command() is True
+
+ def test_bang_approve_routes_to_skill_approve(self):
+ event = MessageEvent(text="!approve")
+ assert event.get_command() == "skill-approve"
+
+ def test_bang_deny_routes_to_skill_deny(self):
+ event = MessageEvent(text="!deny")
+ assert event.get_command() == "skill-deny"
+
+ def test_bang_approvals_routes_to_skill_approvals(self):
+ event = MessageEvent(text="!approvals")
+ assert event.get_command() == "skill-approvals"
+
+ def test_bang_approve_with_id_preserves_command_name(self):
+ """Arguments are NOT pulled into get_command() — that helper
+ returns the command name only. Use get_command_args() for the
+ rest of the line."""
+ event = MessageEvent(text="!approve sk1234_abc")
+ assert event.get_command() == "skill-approve"
+ assert event.get_command_args() == "sk1234_abc"
+
+ def test_unknown_bang_prefix_is_not_command(self):
+ """Bang-prefixed tokens not in the mapping stay non-commands."""
+ event = MessageEvent(text="!bogus")
+ assert event.is_command() is False
+ assert event.get_command() is None
+
+
class TestMessageEventGetCommand:
def test_simple_command(self):
event = MessageEvent(text="/new")
diff --git a/tests/gateway/test_telegram_noise_filter.py b/tests/gateway/test_telegram_noise_filter.py
index b5cbf820bcc7..7c73683f8f7f 100644
--- a/tests/gateway/test_telegram_noise_filter.py
+++ b/tests/gateway/test_telegram_noise_filter.py
@@ -81,3 +81,43 @@ def test_telegram_final_response_keeps_normal_answers():
answer = "Here is the clean summary you asked for."
assert _sanitize_gateway_final_response(Platform.TELEGRAM, answer) == answer
+
+
+def test_skill_approval_sentinel_stripped_from_final_response():
+ """The `USER_APPROVAL_REQUIRED:` sentinel must never reach the user.
+
+ When the agent halts pending an approval decision, the response
+ surface must not include the raw sentinel token — the user would
+ have no idea what to do with it. The user-facing portion of the
+ message (after the sentinel line) is preserved.
+ """
+ raw = "USER_APPROVAL_REQUIRED:sk1234_abc\nI will need your approval to install this skill."
+
+ for platform in (Platform.TELEGRAM, Platform.DISCORD, Platform.SLACK, "local"):
+ sanitized = _sanitize_gateway_final_response(platform, raw)
+ assert "USER_APPROVAL_REQUIRED" not in sanitized, (
+ f"Sentinel leaked to {platform}: {sanitized!r}"
+ )
+ # The user-facing portion of the message is preserved. Note:
+ # the prose-leak stripper may consume certain "I'll" patterns
+ # upstream — we use a plain ``"will need"`` verb so the assertion
+ # is independent of prose-strip behaviour.
+ assert "need your approval" in sanitized
+
+
+def test_skill_approval_sentinel_stripped_when_alone():
+ """A response that consists only of the sentinel must collapse to ``""``.
+
+ Edge case: if the prose-leak stripper upstream consumes the
+ human-readable second line, the sentinel ends up alone in the
+ text. The sanitizer must still strip it cleanly, leaving the user
+ with no spurious token and no extra noise.
+ """
+ raw = "USER_APPROVAL_REQUIRED:sk1234_abc"
+
+ for platform in (Platform.TELEGRAM, Platform.DISCORD, Platform.SLACK, "local"):
+ sanitized = _sanitize_gateway_final_response(platform, raw)
+ assert "USER_APPROVAL_REQUIRED" not in sanitized
+ # Empty / whitespace-only is fine — the chat platform will drop it.
+ assert sanitized.strip() == ""
+
diff --git a/tests/test_mcp_serve_ari_tools.py b/tests/test_mcp_serve_ari_tools.py
new file mode 100644
index 000000000000..e742f3a3e2ba
--- /dev/null
+++ b/tests/test_mcp_serve_ari_tools.py
@@ -0,0 +1,320 @@
+"""
+Tests for ``mcp_serve_ari_tools``.
+
+Covers the ARI bugs the original PR draft had:
+
+ 1. ``ari.play_tts`` must post to ``/channels/{id}/play`` with a
+ ``media`` query parameter, NOT ``/channels/{id}/playback`` with a
+ JSON body. ARI has no native TTS — ``/play`` is a streaming-media
+ operation, ``/playback`` is a sub-resource of an existing Playback
+ object. The ARI docs are unambiguous: "Almost all media is played
+ to a channel using the POST /channels/{channel_id}/play operation."
+
+ 2. ``ari.transfer`` must post to ``/channels/{id}/redirect``, not
+ ``/channels/{id}/transfer`` (which does not exist in ARI).
+
+ 3. ``ari.dial`` must put operation parameters (``app``, ``endpoint``,
+ ``callerId``) on the **query string**, and flatten
+ ``variables={k: v}`` into per-variable ``variable_=v`` query
+ parameters. ARI is form/query-encoded, not JSON-bodied, for these
+ fields. Sending a JSON body to ``POST /channels`` is rejected.
+
+ 4. ``ari.stop_playback`` must use ``DELETE /playbacks/{id}``.
+
+ 5. ``ari.play_tts`` with an explicit ``media_uri`` must skip the
+ TTS round-trip and post directly to ``/play`` with the URI.
+
+ 6. When TTS isn't configured and no ``media_uri`` is passed,
+ ``ari.play_tts`` must return a clear error rather than silently
+ 404-ing on a non-existent ``/playback`` endpoint.
+
+ 7. The optional external TTS endpoint may return audio in two shapes:
+ a ``Content-Type: audio/*`` response (we base64-wrap into a
+ ``data:`` URI) or an ``application/json`` body with a
+ ``media_uri`` field. Both must work.
+
+The tools are closures registered on the FastMCP server, so the test
+fixture builds a real ``FastMCP`` instance, calls
+``register_ari_tools()`` on it, and looks up the resulting tool
+functions by their registered name (``ari.answer``, ``ari.dial``,
+etc.). This pins the public surface that downstream MCP clients see.
+"""
+
+import json
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+from mcp.server.fastmcp import FastMCP
+
+import mcp_serve_ari_tools as ari_mod
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _ok_response(json_body=None):
+ resp = MagicMock()
+ resp.status_code = 200
+ resp.content = b"" if json_body is None else json.dumps(json_body).encode()
+ resp.json.return_value = json_body or {}
+ resp.raise_for_status = MagicMock()
+ return resp
+
+
+def _err_response(status, body):
+ resp = MagicMock()
+ resp.status_code = status
+ resp.text = body
+ resp.raise_for_status = MagicMock(
+ side_effect=httpx.HTTPStatusError(
+ "boom", request=MagicMock(), response=resp,
+ )
+ )
+ return resp
+
+
+@pytest.fixture
+def tools():
+ """Build a FastMCP server, register the ARI tools, return the
+ public tool functions keyed by their ``ari.*`` name.
+ """
+ mcp = FastMCP("test")
+ ari_mod.register_ari_tools(mcp)
+ return {name: tool.fn for name, tool in mcp._tool_manager._tools.items()}
+
+
+@pytest.fixture
+def mock_ari_post(monkeypatch):
+ """Capture every ``httpx.Client.request`` invocation. A per-test
+ router can override this with a more elaborate mock that also
+ handles the TTS endpoint.
+ """
+ calls = []
+
+ def fake_request(self, method, url, params=None, **kwargs):
+ calls.append({
+ "method": method,
+ "url": url,
+ "params": params or {},
+ "kwargs": kwargs,
+ })
+ return _ok_response({"id": "playback-abc123"})
+
+ monkeypatch.setattr("httpx.Client.request", fake_request)
+ return calls
+
+
+# ---------------------------------------------------------------------------
+# 1. play_tts: /play (not /playback) with media on the query string
+# ---------------------------------------------------------------------------
+
+class TestPlayTTS:
+ def test_calls_play_not_playback(self, tools, monkeypatch):
+ """The headline bug from the PR review: play_tts was hitting
+ the wrong endpoint. Pinned here against regression.
+ """
+ monkeypatch.setenv("ASTERISK_TTS_URL", "http://tts.local/synth")
+ calls = []
+
+ def router(self, method, url, params=None, **kwargs):
+ calls.append({"method": method, "url": url, "params": params or {}, "kwargs": kwargs})
+ if "tts.local" in url:
+ r = MagicMock()
+ r.status_code = 200
+ r.headers = {"content-type": "application/json"}
+ r.content = b'{"media_uri": "http://tts.local/out/abc.wav"}'
+ r.json.return_value = {"media_uri": "http://tts.local/out/abc.wav"}
+ r.raise_for_status = MagicMock()
+ return r
+ return _ok_response({"id": "pb1"})
+
+ monkeypatch.setattr("httpx.Client.request", router)
+ tools["ari.play_tts"](channel_id="ch-1", text="hello world")
+
+ ari_calls = [c for c in calls if c["url"].endswith("/play")]
+ assert len(ari_calls) == 1, f"Expected 1 /play call, got: {calls}"
+ call = ari_calls[0]
+ assert call["method"] == "POST"
+ # The /playback sub-resource is the wrong endpoint.
+ assert "/playback" not in call["url"], (
+ "REGRESSION: play_tts is hitting /playback. ARI has no such "
+ "channel operation; /playback is a sub-resource of a Playback. "
+ f"URL was: {call['url']}"
+ )
+ # Media must be a query param, not in a JSON body.
+ assert "media" in call["params"]
+ assert call["params"]["media"] == "http://tts.local/out/abc.wav"
+
+ def test_explicit_media_uri_skips_tts(self, tools, mock_ari_post, monkeypatch):
+ """If the operator already has a sound: or http: URI, the tool
+ must NOT make a TTS call.
+ """
+ monkeypatch.setenv("ASTERISK_TTS_URL", "http://should-not-be-hit.example/synth")
+ tools["ari.play_tts"](channel_id="ch-1", text="ignored", media_uri="sound:custom")
+ assert len(mock_ari_post) == 1
+ assert mock_ari_post[0]["params"]["media"] == "sound:custom"
+
+ def test_no_tts_no_uri_returns_error(self, tools, mock_ari_post, monkeypatch):
+ """Without TTS_URL or media_uri we must return an error dict and
+ NOT hit ARI at all (the original draft would 404 on /playback).
+ """
+ monkeypatch.delenv("ASTERISK_TTS_URL", raising=False)
+ out = tools["ari.play_tts"](channel_id="ch-1", text="hello world")
+ assert mock_ari_post == [], (
+ f"REGRESSION: ari.play_tts made ARI calls without a media URI. "
+ f"Calls: {mock_ari_post}"
+ )
+ body = json.loads(out)
+ assert body["status"] == "error"
+
+ def test_tts_audio_response_base64_wrapped(self, tools, monkeypatch):
+ monkeypatch.setenv("ASTERISK_TTS_URL", "http://tts.local/synth")
+ calls = []
+
+ def router(self, method, url, params=None, **kwargs):
+ calls.append({"method": method, "url": url, "params": params or {}, "kwargs": kwargs})
+ if "tts.local" in url:
+ r = MagicMock()
+ r.status_code = 200
+ r.headers = {"content-type": "audio/wav"}
+ r.content = b"RIFFFAKE"
+ r.raise_for_status = MagicMock()
+ return r
+ return _ok_response({"id": "pb1"})
+
+ monkeypatch.setattr("httpx.Client.request", router)
+ tools["ari.play_tts"](channel_id="ch-1", text="hi")
+ ari_calls = [c for c in calls if c["url"].endswith("/play")]
+ assert len(ari_calls) == 1
+ # Raw audio body is base64-wrapped into a data: URI ARI can play.
+ assert ari_calls[0]["params"]["media"].startswith("data:audio/wav;base64,")
+
+ def test_tts_json_response_uri_passthrough(self, tools, monkeypatch):
+ monkeypatch.setenv("ASTERISK_TTS_URL", "http://tts.local/synth")
+ calls = []
+
+ def router(self, method, url, params=None, **kwargs):
+ calls.append({"method": method, "url": url, "params": params or {}, "kwargs": kwargs})
+ if "tts.local" in url:
+ r = MagicMock()
+ r.status_code = 200
+ r.headers = {"content-type": "application/json"}
+ r.content = b'{"media_uri": "http://tts.local/out/x.wav"}'
+ r.json.return_value = {"media_uri": "http://tts.local/out/x.wav"}
+ r.raise_for_status = MagicMock()
+ return r
+ return _ok_response({"id": "pb1"})
+
+ monkeypatch.setattr("httpx.Client.request", router)
+ tools["ari.play_tts"](channel_id="ch-1", text="hi")
+ ari_calls = [c for c in calls if c["url"].endswith("/play")]
+ assert ari_calls[0]["params"]["media"] == "http://tts.local/out/x.wav"
+
+
+# ---------------------------------------------------------------------------
+# 2. transfer must use /redirect, not /transfer
+# ---------------------------------------------------------------------------
+
+class TestTransfer:
+ def test_calls_redirect_not_transfer(self, tools, mock_ari_post):
+ tools["ari.transfer"](channel_id="ch-1", target="PJSIP/1001")
+ assert len(mock_ari_post) == 1
+ call = mock_ari_post[0]
+ assert call["url"].endswith("/ari/channels/ch-1/redirect"), (
+ "ARI redirect must POST to /channels/{id}/redirect. "
+ f"/channels/{{id}}/transfer does not exist. Got: {call['url']}"
+ )
+ assert call["params"]["endpoint"] == "PJSIP/1001"
+
+
+# ---------------------------------------------------------------------------
+# 3. dial: app, endpoint, callerId on query; variables as variable_* prefix
+# ---------------------------------------------------------------------------
+
+class TestDial:
+ def test_app_on_query(self, tools, mock_ari_post):
+ tools["ari.dial"](target="PJSIP/1001", app="stasis-bridge")
+ assert mock_ari_post[0]["params"]["app"] == "stasis-bridge"
+ assert mock_ari_post[0]["params"]["endpoint"] == "PJSIP/1001"
+ assert mock_ari_post[0]["url"].endswith("/ari/channels")
+
+ def test_variables_become_variable_prefix(self, tools, mock_ari_post):
+ tools["ari.dial"](
+ target="PJSIP/1001",
+ variables={"CALLERID(name)": "Hermes", "CHANNEL(language)": "en"},
+ )
+ p = mock_ari_post[0]["params"]
+ # ARI's convention: each variable becomes its own query param.
+ assert p["variable_CALLERID(name)"] == "Hermes"
+ assert p["variable_CHANNEL(language)"] == "en"
+ # The raw key must NOT appear in the params (the original draft
+ # tried to send variables as a JSON body, which ARI rejects).
+ assert "CALLERID(name)" not in p
+ assert "variables" not in p
+ # Make sure we didn't smuggle a JSON body in either.
+ assert "json" not in mock_ari_post[0]["kwargs"]
+
+ def test_caller_id_passed_through(self, tools, mock_ari_post):
+ tools["ari.dial"](target="PJSIP/1001", caller_id="+31xxxxxxxxx")
+ assert mock_ari_post[0]["params"]["callerId"] == "+31xxxxxxxxx"
+
+
+# ---------------------------------------------------------------------------
+# 4. stop_playback: DELETE /playbacks/{id}
+# ---------------------------------------------------------------------------
+
+class TestStopPlayback:
+ def test_delete_playbacks(self, tools, mock_ari_post):
+ tools["ari.stop_playback"](playback_id="pb-abc")
+ assert len(mock_ari_post) == 1
+ call = mock_ari_post[0]
+ assert call["method"] == "DELETE"
+ assert call["url"].endswith("/ari/playbacks/pb-abc")
+
+
+# ---------------------------------------------------------------------------
+# 5. answer / hangup
+# ---------------------------------------------------------------------------
+
+class TestAnswer:
+ def test_answer_url(self, tools, mock_ari_post):
+ tools["ari.answer"](channel_id="ch-1")
+ assert mock_ari_post[0]["method"] == "POST"
+ assert mock_ari_post[0]["url"].endswith("/ari/channels/ch-1/answer")
+
+
+class TestHangup:
+ def test_hangup_with_reason_code(self, tools, mock_ari_post):
+ tools["ari.hangup"](channel_id="ch-1")
+ call = mock_ari_post[0]
+ assert call["method"] == "POST"
+ assert call["url"].endswith("/ari/channels/ch-1/hangup")
+ assert call["params"]["reason_code"] == 16
+
+
+# ---------------------------------------------------------------------------
+# 6. list_channels
+# ---------------------------------------------------------------------------
+
+class TestListChannels:
+ def test_get_channels(self, tools, mock_ari_post):
+ tools["ari.list_channels"]()
+ assert mock_ari_post[0]["method"] == "GET"
+ assert mock_ari_post[0]["url"].endswith("/ari/channels")
+
+
+# ---------------------------------------------------------------------------
+# 7. Error handling: ARI returns 404 → tool reports error, not crash
+# ---------------------------------------------------------------------------
+
+class TestErrorHandling:
+ def test_ari_404_returns_error_dict(self, tools, monkeypatch):
+ def boom(self, method, url, params=None, **kwargs):
+ return _err_response(404, "Not Found")
+ monkeypatch.setattr("httpx.Client.request", boom)
+ out = tools["ari.play_tts"](channel_id="ch-1", text="hi", media_uri="sound:foo")
+ body = json.loads(out)
+ assert body["status"] == "error"
+ assert "404" in body["message"]
diff --git a/tests/tools/test_skill_manager_tool.py b/tests/tools/test_skill_manager_tool.py
index e7e5e4a78b27..add1f85f995c 100644
--- a/tests/tools/test_skill_manager_tool.py
+++ b/tests/tools/test_skill_manager_tool.py
@@ -619,10 +619,14 @@ def test_scan_runs_when_flag_on(self, tmp_path):
mock_scan.assert_called_once()
def test_scan_blocks_dangerous_when_flag_on(self, tmp_path):
- """Dangerous verdict + flag on → returns an error string for the agent."""
+ """Dangerous verdict + flag on → writes a pending approval record."""
from tools.skill_manager_tool import _security_scan_skill
from tools.skills_guard import ScanResult, Finding
+ home = tmp_path / "home"
+ skill_dir = tmp_path / "danger-skill"
+ skill_dir.mkdir()
+ (skill_dir / "SKILL.md").write_text(VALID_SKILL_CONTENT, encoding="utf-8")
finding = Finding(
pattern_id="test", severity="critical", category="exfiltration",
file="SKILL.md", line=1, match="curl $TOKEN", description="test",
@@ -636,11 +640,88 @@ def test_scan_blocks_dangerous_when_flag_on(self, tmp_path):
summary="dangerous",
)
with patch("tools.skill_manager_tool._guard_agent_created_enabled", return_value=True), \
+ patch("tools.skill_manager_tool.get_hermes_home", return_value=home), \
patch("tools.skill_manager_tool.scan_skill", return_value=fake_result):
- result = _security_scan_skill(tmp_path)
+ result = _security_scan_skill(skill_dir)
assert result is not None
- assert "Security scan blocked" in result
+ assert result.startswith("USER_APPROVAL_REQUIRED:sk")
+ record_path = next((home / "state" / "skill-approvals").glob("*.json"))
+ record = json.loads(record_path.read_text(encoding="utf-8"))
+ assert record["status"] == "pending"
+ assert record["skill_name"] == "test"
+ assert Path(record["pending_skill_dir"], "SKILL.md").is_file()
+
+ def test_approve_pending_skill_installs_snapshot_after_create_rollback(self, tmp_path):
+ """Approval installs the saved snapshot after _create_skill removes the candidate."""
+ from tools.skill_approval_records import approve_skill_approval
+ from tools.skills_guard import ScanResult, Finding
+
+ home = tmp_path / "home"
+ skills_dir = tmp_path / "candidate-skills"
+ install_dir = tmp_path / "installed-skills"
+ finding = Finding(
+ pattern_id="test", severity="critical", category="exfiltration",
+ file="SKILL.md", line=1, match="curl $TOKEN", description="test",
+ )
+ fake_result = ScanResult(
+ skill_name="danger-skill",
+ source="agent-created",
+ trust_level="agent-created",
+ verdict="dangerous",
+ findings=[finding],
+ summary="dangerous",
+ )
+
+ with patch("tools.skill_manager_tool.SKILLS_DIR", skills_dir), \
+ patch("tools.skill_manager_tool._guard_agent_created_enabled", return_value=True), \
+ patch("tools.skill_manager_tool.get_hermes_home", return_value=home), \
+ patch("tools.skill_manager_tool.scan_skill", return_value=fake_result):
+ result = _create_skill("danger-skill", VALID_SKILL_CONTENT)
+
+ assert result["success"] is False
+ assert result["error"].startswith("USER_APPROVAL_REQUIRED:")
+ approval_id = result["error"].splitlines()[0].split(":", 1)[1]
+ assert not (skills_dir / "danger-skill").exists()
+
+ record = approve_skill_approval(approval_id, home=home, skills_root=install_dir)
+ assert record["status"] == "approved"
+ assert (install_dir / "danger-skill" / "SKILL.md").read_text(encoding="utf-8") == VALID_SKILL_CONTENT
+
+ def test_pending_skill_approval_list_and_deny(self, tmp_path):
+ from tools.skill_approval_records import (
+ deny_skill_approval,
+ format_pending_skill_approvals,
+ list_pending_skill_approvals,
+ write_pending_skill_approval,
+ )
+ from tools.skills_guard import ScanResult, Finding
+
+ home = tmp_path / "home"
+ skill_dir = tmp_path / "deny-skill"
+ skill_dir.mkdir()
+ (skill_dir / "SKILL.md").write_text(VALID_SKILL_CONTENT, encoding="utf-8")
+ finding = Finding(
+ pattern_id="test", severity="critical", category="exfiltration",
+ file="SKILL.md", line=1, match="curl $TOKEN", description="test",
+ )
+ scan_result = ScanResult(
+ skill_name="deny-skill",
+ source="agent-created",
+ trust_level="agent-created",
+ verdict="dangerous",
+ findings=[finding],
+ summary="dangerous",
+ )
+
+ approval_id = write_pending_skill_approval(home, skill_dir, scan_result, "needs approval")
+ pending = list_pending_skill_approvals(home)
+ assert [record["id"] for record in pending] == [approval_id]
+ assert approval_id in format_pending_skill_approvals(pending)
+
+ denied = deny_skill_approval(approval_id, home=home)
+ assert denied["status"] == "denied"
+ assert list_pending_skill_approvals(home) == []
def test_guard_flag_reads_config_default_false(self):
"""_guard_agent_created_enabled returns False when config doesn't set it."""
@@ -939,3 +1020,234 @@ def test_broken_sidecar_fails_open(self, tmp_path):
side_effect=RuntimeError("sidecar broken")):
result = _delete_skill("my-skill")
assert result["success"] is True
+
+
+class TestSkillApprovalFailureModes:
+ """Failure-mode coverage for approve/deny/load guards.
+
+ Happy-path coverage for the approval flow lives in the
+ ``test_pending_skill_approval_list_and_deny`` and
+ ``test_approve_pending_skill_installs_snapshot_after_create_rollback``
+ tests above. This class covers the guardrails around bad state
+ and filesystem issues — the cases the agent will most often hit
+ when a user pastes the wrong id, or when a snapshot was removed
+ out from under us.
+ """
+
+ def test_load_skill_approval_nonexistent_id_raises(self, tmp_path):
+ from tools.skill_approval_records import (
+ SkillApprovalError,
+ load_skill_approval,
+ )
+
+ home = tmp_path / "home"
+ with pytest.raises(SkillApprovalError):
+ load_skill_approval("does-not-exist", home=home)
+
+ def test_load_skill_approval_corrupted_record_raises(self, tmp_path):
+ from tools.skill_approval_records import (
+ SkillApprovalError,
+ load_skill_approval,
+ )
+
+ home = tmp_path / "home"
+ records_dir = home / "state" / "skill-approvals"
+ records_dir.mkdir(parents=True, exist_ok=True)
+ (records_dir / "bad.json").write_text("this is not valid json", encoding="utf-8")
+
+ with pytest.raises(SkillApprovalError):
+ load_skill_approval("bad", home=home)
+
+ def test_load_skill_approval_rejects_path_traversal_in_id(self, tmp_path):
+ """Unsafe approval ids must be rejected before any path arithmetic."""
+ from tools.skill_approval_records import (
+ SkillApprovalError,
+ load_skill_approval,
+ )
+
+ home = tmp_path / "home"
+ with pytest.raises(SkillApprovalError):
+ load_skill_approval("../escape", home=home)
+
+ def test_source_dir_missing_raises_for_pending_record(self, tmp_path):
+ from tools.skill_approval_records import (
+ SkillApprovalError,
+ _source_dir_for_record,
+ )
+ from tools.skill_approval_records import approval_records_dir
+
+ home = tmp_path / "home"
+ records_dir = approval_records_dir(home)
+ missing_dir = tmp_path / "pending" / "missing-skill"
+ record = {
+ "id": "missing-source",
+ "skill_name": "missing-skill",
+ "status": "pending",
+ "pending_skill_dir": str(missing_dir),
+ }
+ records_dir.mkdir(parents=True, exist_ok=True)
+ (records_dir / "missing-source.json").write_text(json.dumps(record), encoding="utf-8")
+
+ with pytest.raises(SkillApprovalError):
+ _source_dir_for_record(record)
+
+ def test_safe_skill_name_rejects_path_traversal(self):
+ from tools.skill_approval_records import SkillApprovalError, _safe_skill_name
+
+ # Empty skill_name falls back to ``Path(skill_dir).name`` (see
+ # _safe_skill_name source), so we don't test that case here — the
+ # path-traversal/absolute values are the load-bearing rejection.
+ for unsafe in ("../evil", "/tmp/evil", "..", "."):
+ record = {"skill_name": unsafe, "skill_dir": "/tmp/x"}
+ with pytest.raises(SkillApprovalError):
+ _safe_skill_name(record)
+
+ def test_safe_skill_name_round_trip(self):
+ from tools.skill_approval_records import _safe_skill_name
+
+ record = {"skill_name": "my-safe-skill", "skill_dir": "/tmp/x"}
+ assert _safe_skill_name(record) == "my-safe-skill"
+ record_a = {"skill_name": "a", "skill_dir": "/tmp/x"}
+ assert _safe_skill_name(record_a) == "a"
+ # Dots and dashes are allowed by the regex (e.g. ``my.skill-v1``).
+ record_dotted = {"skill_name": "my.skill-v1", "skill_dir": "/tmp/x"}
+ assert _safe_skill_name(record_dotted) == "my.skill-v1"
+
+ def test_approve_with_missing_snapshot_dir_raises(self, tmp_path):
+ """If the snapshot dir is gone, approve must error cleanly."""
+ from tools.skill_approval_records import (
+ SkillApprovalError,
+ approve_skill_approval,
+ )
+ from tools.skill_approval_records import approval_records_dir
+
+ home = tmp_path / "home"
+ records_dir = approval_records_dir(home)
+ records_dir.mkdir(parents=True, exist_ok=True)
+ record = {
+ "id": "missing-snapshot",
+ "skill_name": "ghost-skill",
+ "status": "pending",
+ "pending_skill_dir": str(tmp_path / "no-such-dir" / "ghost-skill"),
+ }
+ (records_dir / "missing-snapshot.json").write_text(json.dumps(record), encoding="utf-8")
+
+ with pytest.raises(SkillApprovalError):
+ approve_skill_approval("missing-snapshot", home=home)
+
+ def test_approve_with_path_traversal_skill_name_raises(self, tmp_path):
+ """Unsafe skill_name in the record must error before any file move."""
+ from tools.skill_approval_records import (
+ SkillApprovalError,
+ approval_records_dir,
+ approve_skill_approval,
+ )
+
+ home = tmp_path / "home"
+ records_dir = approval_records_dir(home)
+ records_dir.mkdir(parents=True, exist_ok=True)
+ # Snapshot dir points at a real dir with SKILL.md, but the
+ # skill_name embedded in the record is path-traversal — the
+ # install path would escape the skills root and must be refused.
+ snapshot = tmp_path / "real-snapshot"
+ snapshot.mkdir()
+ (snapshot / "SKILL.md").write_text(VALID_SKILL_CONTENT, encoding="utf-8")
+ record = {
+ "id": "evil-name",
+ "skill_name": "../evil",
+ "status": "pending",
+ "pending_skill_dir": str(snapshot),
+ }
+ (records_dir / "evil-name.json").write_text(json.dumps(record), encoding="utf-8")
+
+ with pytest.raises(SkillApprovalError):
+ approve_skill_approval("evil-name", home=home)
+
+ def _create_dangerous_pending(self, home, skill_name, content):
+ """Drive ``_create_skill`` to write a pending approval record.
+
+ Mirrors the setup in
+ ``test_approve_pending_skill_installs_snapshot_after_create_rollback``
+ so the failure-mode tests can focus on the post-creation guards
+ without re-implementing the scan / guard plumbing.
+ """
+ from tools.skills_guard import Finding, ScanResult
+
+ finding = Finding(
+ pattern_id="test",
+ severity="critical",
+ category="exfiltration",
+ file="SKILL.md",
+ line=1,
+ match="curl $TOKEN",
+ description="test",
+ )
+ fake_result = ScanResult(
+ skill_name=skill_name,
+ source="agent-created",
+ trust_level="agent-created",
+ verdict="dangerous",
+ findings=[finding],
+ summary="dangerous",
+ )
+ with patch("tools.skill_manager_tool.SKILLS_DIR", home / "candidate-skills"), \
+ patch("tools.skill_manager_tool._guard_agent_created_enabled", return_value=True), \
+ patch("tools.skill_manager_tool.get_hermes_home", return_value=home), \
+ patch("tools.skill_manager_tool.scan_skill", return_value=fake_result):
+ result = _create_skill(skill_name, content)
+ assert result["success"] is False
+ assert result["error"].startswith("USER_APPROVAL_REQUIRED:")
+ return result["error"].splitlines()[0].split(":", 1)[1]
+
+ def test_approve_already_approved_id_raises(self, tmp_path):
+ from tools.skill_approval_records import (
+ SkillApprovalError,
+ approve_skill_approval,
+ )
+
+ home = tmp_path / "home"
+ install_dir = tmp_path / "installed-skills"
+
+ approval_id = self._create_dangerous_pending(
+ home, "double-approve-skill", VALID_SKILL_CONTENT
+ )
+ approve_skill_approval(approval_id, home=home, skills_root=install_dir)
+
+ with pytest.raises(SkillApprovalError):
+ approve_skill_approval(approval_id, home=home, skills_root=install_dir)
+
+ def test_deny_already_denied_id_raises(self, tmp_path):
+ from tools.skill_approval_records import (
+ SkillApprovalError,
+ deny_skill_approval,
+ )
+
+ home = tmp_path / "home"
+
+ approval_id = self._create_dangerous_pending(
+ home, "double-deny-skill", VALID_SKILL_CONTENT
+ )
+ deny_skill_approval(approval_id, home=home)
+
+ with pytest.raises(SkillApprovalError):
+ deny_skill_approval(approval_id, home=home)
+
+ def test_approve_already_denied_id_raises(self, tmp_path):
+ """Approving a record that was already denied must error."""
+ from tools.skill_approval_records import (
+ SkillApprovalError,
+ approve_skill_approval,
+ deny_skill_approval,
+ )
+
+ home = tmp_path / "home"
+ install_dir = tmp_path / "installed-skills"
+
+ approval_id = self._create_dangerous_pending(
+ home, "deny-then-approve-skill", VALID_SKILL_CONTENT
+ )
+ deny_skill_approval(approval_id, home=home)
+
+ with pytest.raises(SkillApprovalError):
+ approve_skill_approval(approval_id, home=home, skills_root=install_dir)
+
diff --git a/tools/skill_approval_records.py b/tools/skill_approval_records.py
new file mode 100644
index 000000000000..226306ae48d4
--- /dev/null
+++ b/tools/skill_approval_records.py
@@ -0,0 +1,208 @@
+"""Pending approval records for guarded agent-created skills."""
+
+from __future__ import annotations
+
+import json
+import re
+import shutil
+import time
+from pathlib import Path
+from typing import Any
+
+from hermes_constants import get_hermes_home
+
+
+_APPROVAL_ID_RE = re.compile(r"^[A-Za-z0-9_]+$")
+_SAFE_SKILL_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_.-]{0,63}$")
+
+
+class SkillApprovalError(ValueError):
+ """Raised when a skill approval record cannot be used."""
+
+
+def approval_records_dir(home: Path | None = None) -> Path:
+ return (home or get_hermes_home()) / "state" / "skill-approvals"
+
+
+def _approval_record_path(approval_id: str, home: Path | None = None) -> Path:
+ approval_id = (approval_id or "").strip()
+ if not _APPROVAL_ID_RE.match(approval_id):
+ raise SkillApprovalError("Invalid approval id.")
+ return approval_records_dir(home) / f"{approval_id}.json"
+
+
+def _atomic_write_json(path: Path, data: dict[str, Any]) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ tmp = path.with_suffix(path.suffix + ".tmp")
+ tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+ tmp.replace(path)
+
+
+def write_pending_skill_approval(
+ home: Path,
+ skill_dir: Path,
+ scan_result: Any,
+ reason: str,
+) -> str:
+ """Persist a pending approval record and a snapshot of the scanned skill."""
+ approval_dir = approval_records_dir(home)
+ approval_dir.mkdir(parents=True, exist_ok=True)
+
+ safe_fragment = re.sub(r"[^A-Za-z0-9_]", "_", skill_dir.name[:24]).strip("_")
+ base_id = f"sk{int(time.time())}_{safe_fragment or 'skill'}"
+ approval_id = base_id
+ counter = 1
+ while (approval_dir / f"{approval_id}.json").exists():
+ counter += 1
+ approval_id = f"{base_id}_{counter}"
+
+ snapshot_dir = approval_dir / approval_id / "skill"
+ if snapshot_dir.exists():
+ shutil.rmtree(snapshot_dir)
+ shutil.copytree(skill_dir, snapshot_dir, symlinks=False)
+
+ record = {
+ "id": approval_id,
+ "skill_dir": str(skill_dir),
+ "pending_skill_dir": str(snapshot_dir),
+ "skill_name": scan_result.skill_name,
+ "verdict": scan_result.verdict,
+ "reason": reason,
+ "findings_count": len(scan_result.findings),
+ "findings": [
+ {
+ "pattern_id": f.pattern_id,
+ "severity": f.severity,
+ "category": f.category,
+ "file": f.file,
+ "line": f.line,
+ "description": f.description,
+ }
+ for f in scan_result.findings
+ ],
+ "created_at": time.time(),
+ "status": "pending",
+ }
+ _atomic_write_json(approval_dir / f"{approval_id}.json", record)
+ return approval_id
+
+
+def load_skill_approval(approval_id: str, home: Path | None = None) -> dict[str, Any]:
+ path = _approval_record_path(approval_id, home)
+ if not path.exists():
+ raise SkillApprovalError(f"No approval record found for {approval_id}.")
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except json.JSONDecodeError as exc:
+ raise SkillApprovalError(f"Approval record {approval_id} is not valid JSON.") from exc
+ if not isinstance(data, dict):
+ raise SkillApprovalError(f"Approval record {approval_id} is malformed.")
+ return data
+
+
+def list_pending_skill_approvals(home: Path | None = None) -> list[dict[str, Any]]:
+ records: list[dict[str, Any]] = []
+ approval_dir = approval_records_dir(home)
+ if not approval_dir.exists():
+ return records
+ for path in sorted(approval_dir.glob("*.json")):
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except Exception:
+ continue
+ if isinstance(data, dict) and data.get("status") == "pending":
+ records.append(data)
+ return records
+
+
+def _source_dir_for_record(record: dict[str, Any]) -> Path:
+ for key in ("pending_skill_dir", "skill_dir"):
+ value = record.get(key)
+ if not value:
+ continue
+ path = Path(str(value)).expanduser()
+ if path.is_dir() and (path / "SKILL.md").is_file():
+ return path
+ raise SkillApprovalError("The pending skill files are missing.")
+
+
+def _safe_skill_name(record: dict[str, Any]) -> str:
+ name = str(record.get("skill_name") or Path(str(record.get("skill_dir", ""))).name).strip()
+ if not _SAFE_SKILL_NAME_RE.match(name):
+ raise SkillApprovalError(f"Unsafe skill name in approval record: {name!r}.")
+ return name
+
+
+def approve_skill_approval(
+ approval_id: str,
+ *,
+ home: Path | None = None,
+ skills_root: Path | None = None,
+) -> dict[str, Any]:
+ """Install the pending skill snapshot and mark the record approved."""
+ record = load_skill_approval(approval_id, home)
+ if record.get("status") != "pending":
+ raise SkillApprovalError(
+ f"Approval {approval_id} is already {record.get('status', 'resolved')}."
+ )
+
+ source_dir = _source_dir_for_record(record)
+ skill_name = _safe_skill_name(record)
+ root = skills_root or ((home or get_hermes_home()) / "skills")
+ root.mkdir(parents=True, exist_ok=True)
+ dest = (root / skill_name).resolve()
+ root_resolved = root.resolve()
+ try:
+ dest.relative_to(root_resolved)
+ except ValueError as exc:
+ raise SkillApprovalError("Resolved install path escapes the skills directory.") from exc
+
+ tmp_dest = root / f".{skill_name}.approval-tmp"
+ if tmp_dest.exists():
+ shutil.rmtree(tmp_dest)
+ shutil.copytree(source_dir, tmp_dest, symlinks=False)
+ if dest.exists():
+ shutil.rmtree(dest)
+ tmp_dest.replace(dest)
+
+ record["status"] = "approved"
+ record["approved_at"] = time.time()
+ record["installed_path"] = str(dest)
+ _atomic_write_json(_approval_record_path(approval_id, home), record)
+
+ try:
+ from agent.prompt_builder import clear_skills_system_prompt_cache
+
+ clear_skills_system_prompt_cache(clear_snapshot=True)
+ except Exception:
+ pass
+
+ return record
+
+
+def deny_skill_approval(approval_id: str, *, home: Path | None = None) -> dict[str, Any]:
+ """Mark a pending skill approval as denied."""
+ record = load_skill_approval(approval_id, home)
+ if record.get("status") != "pending":
+ raise SkillApprovalError(
+ f"Approval {approval_id} is already {record.get('status', 'resolved')}."
+ )
+ record["status"] = "denied"
+ record["denied_at"] = time.time()
+ _atomic_write_json(_approval_record_path(approval_id, home), record)
+ return record
+
+
+def format_pending_skill_approvals(records: list[dict[str, Any]]) -> str:
+ if not records:
+ return "No pending skill approvals."
+ lines = ["Pending skill approvals:"]
+ for record in records:
+ approval_id = record.get("id", "?")
+ skill_name = record.get("skill_name", "?")
+ findings = record.get("findings_count", 0)
+ reason = record.get("reason", "")
+ lines.append(f"- {approval_id}: {skill_name} ({findings} findings) {reason}")
+ lines.append("")
+ lines.append("Approve with `/skill-approve ` or deny with `/skill-deny `.")
+ return "\n".join(lines)
diff --git a/tools/skill_manager_tool.py b/tools/skill_manager_tool.py
index 4ce5f06e4c93..894787e7a894 100644
--- a/tools/skill_manager_tool.py
+++ b/tools/skill_manager_tool.py
@@ -79,6 +79,10 @@ def _security_scan_skill(skill_dir: Path) -> Optional[str]:
"""Scan a skill directory after write. Returns error string if blocked, else None.
No-op when skills.guard_agent_created is disabled (the default).
+
+ For "ask" verdicts (agent-created skills with dangerous findings), writes a
+ pending approval record and returns a sentinel-prefixed message the gateway
+ can strip before displaying the user-facing approval prompt.
"""
if not _GUARD_AVAILABLE:
return None
@@ -92,11 +96,29 @@ def _security_scan_skill(skill_dir: Path) -> Optional[str]:
return f"Security scan blocked this skill ({reason}):\n{report}"
if allowed is None:
# "ask" verdict — for agent-created skills this means dangerous
- # findings were detected. Surface as an error so the agent can
- # retry with the flagged content removed.
+ # findings were detected. Persist a snapshot before callers roll
+ # back the candidate directory, then surface a sentinel-prefixed
+ # prompt so chat users can approve or deny the install.
+ from tools.skill_approval_records import write_pending_skill_approval
+
+ approval_id = write_pending_skill_approval(
+ get_hermes_home(),
+ skill_dir,
+ result,
+ reason,
+ )
report = format_scan_report(result)
- logger.warning("Agent-created skill blocked (dangerous findings): %s", reason)
- return f"Security scan blocked this skill ({reason}):\n{report}"
+ logger.warning(
+ "Agent-created skill needs user approval (id=%s): %s",
+ approval_id,
+ reason,
+ )
+ return (
+ f"USER_APPROVAL_REQUIRED:{approval_id}\n"
+ f"{reason}\n\n{report}\n\n"
+ f"Reply `!approve {approval_id}` in any chat to install, "
+ f"or `!deny {approval_id}` to discard."
+ )
except Exception as e:
logger.warning("Security scan failed for %s: %s", skill_dir, e, exc_info=True)
return None