ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

refactor(hermes_cli): middleware/mcp_picker/mcp_security/mcp_startup/memory_oauth — hoist _DownstreamExecutionError, compact layout and docstrings

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:33:09 -07:00
parent 890029b5d4
commit ff5d0c4312
5 changed files with 102 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/mcp_picker.py
+12 -57
View File
@@ -10,23 +10,11 @@ from hermes_cli.colors import Colors, color
from hermes_cli.cli_output import prompt_yes_no
from hermes_cli.curses_ui import curses_single_select
from hermes_cli.mcp_catalog import (
CatalogEntry,
CatalogError,
catalog_diagnostics,
install_entry,
is_enabled,
is_installed,
list_catalog,
installed_servers,
remove_server,
server_enabled,
uninstall_entry,
CatalogEntry, CatalogError, catalog_diagnostics, install_entry, is_enabled, is_installed,
list_catalog, installed_servers, remove_server, server_enabled, uninstall_entry,
)
from hermes_cli.config import load_config, save_config
# ─── Status badges ────────────────────────────────────────────────────────────
_STATUS_NOT_INSTALLED = "available"
_STATUS_DISABLED = "installed (disabled)"
_STATUS_ENABLED = "enabled"
@@ -34,13 +22,9 @@ _STATUS_CUSTOM_ENABLED = "custom — enabled"
_STATUS_CUSTOM_DISABLED = "custom — disabled"
# ─── Row model — unifies catalog and custom entries ──────────────────────────
@dataclass
class _Row:
"""A row in the picker. ``entry`` is set for catalog rows; for custom
user-added MCPs only ``name`` + ``description`` + status are populated."""
"""A picker row. ``entry`` is set for catalog rows; custom MCPs carry only name/description/status."""
name: str
description: str
@@ -68,16 +52,12 @@ def _build_rows() -> List[_Row]:
else:
status = _STATUS_DISABLED
rows.append(_Row(entry.name, entry.description, status, entry))
# Custom MCPs the user added directly (not in the catalog)
# Custom (non-catalog) MCPs: the transport URL/command doubles as the description.
for name, cfg in sorted(servers.items()):
if name in catalog_names:
continue
status = _STATUS_CUSTOM_ENABLED if server_enabled(cfg) else _STATUS_CUSTOM_DISABLED
# Use the transport URL/command as the "description" for custom rows
desc = cfg.get("url") or cfg.get("command") or "(no transport)"
rows.append(_Row(name, str(desc), status))
rows.append(_Row(name, str(cfg.get("url") or cfg.get("command") or "(no transport)"), status))
return rows
@@ -85,8 +65,6 @@ def _format_row(row: _Row) -> str:
return f"{row.name:<18} {row.status:<24} {row.description}"
# ─── Actions ──────────────────────────────────────────────────────────────────
def _enable_disable(name: str, *, enable: bool) -> None:
cfg = load_config()
@@ -156,19 +134,13 @@ def _run_submenu(title: str, actions: list) -> None:
def _handle_row(row: _Row) -> None:
"""Act on the picked row based on its current status."""
# === Catalog row, not yet installed ===
if row.entry and not is_installed(row.name):
_install(row.entry, "install")
return
# === Catalog row, installed but disabled ===
if row.entry and not is_enabled(row.name):
_enable_disable(row.name, enable=True)
return
# === Catalog row, installed + enabled OR custom row ===
if row.is_custom:
# Custom (non-catalog) row submenu
enabled = is_enabled(row.name)
_run_submenu(f"Action for '{row.name}' (custom)", [
("Configure tools (probe server + re-pick)", lambda: _configure_tools(row.name)),
@@ -177,7 +149,6 @@ def _handle_row(row: _Row) -> None:
("Remove from config", lambda: _remove_custom(row.name)),
])
return
# Catalog row, installed + enabled
print()
print(color(f" '{row.name}' is already enabled.", Colors.DIM))
@@ -191,12 +162,8 @@ def _handle_row(row: _Row) -> None:
])
# ─── Output / entry points ────────────────────────────────────────────────────
def _print_rows_text(rows: List[_Row]) -> None:
"""Plain-text catalog dump used as a fallback when curses can't run, and
as the default output of `hermes mcp catalog`."""
"""Plain-text catalog dump: `hermes mcp catalog` output and the non-curses fallback."""
print()
if not rows:
print(color(" No MCPs in the catalog or configured.", Colors.DIM))
@@ -210,13 +177,8 @@ def _print_rows_text(rows: List[_Row]) -> None:
for row in rows:
print(f" {_format_row(row)}")
print()
print(color(
" Install: hermes mcp install <name> Picker: hermes mcp",
Colors.DIM,
))
# Surface manifest-version warnings so users know when their Hermes is
# too old to install everything in the catalog.
print(color(" Install: hermes mcp install <name> Picker: hermes mcp", Colors.DIM))
# Manifest-version warnings: the user's Hermes is too old to install everything listed.
future = [d for d in catalog_diagnostics() if d[1] == "future_manifest"]
if future:
print()
@@ -236,22 +198,15 @@ def show_catalog() -> None:
def run_picker() -> None:
"""`hermes mcp picker` (and default `hermes mcp`) — interactive selector.
Loops until the user hits ESC/q. After each action the picker re-renders so the user can manage
several entries in one session.
"""
"""`hermes mcp picker` (and default `hermes mcp`) — interactive selector; re-renders after each
action until ESC/q."""
while True:
rows = _build_rows()
if not rows or not sys.stdin.isatty():
# Non-interactive shell: degrade to the text dump rather than failing.
_print_rows_text(rows)
_print_rows_text(rows) # non-interactive: degrade to the text dump
return
labels = [_format_row(r) for r in rows]
idx = curses_single_select(
"MCP Catalog — ↑↓ navigate ENTER act on entry ESC/q quit",
labels,
"MCP Catalog — ↑↓ navigate ENTER act on entry ESC/q quit", [_format_row(r) for r in rows],
)
if idx is None:
return
hermes_cli/mcp_security.py
+17 -42
View File
@@ -1,12 +1,9 @@
"""Security checks for user-configured MCP server entries.
3. A hardcoded indicator-of-compromise (IOC) blocklist for that campaign — the attacker's
``hermes-0day`` SSH public key and source IPs. Any entry whose command/args/env carry an IOC is
refused outright, regardless of shape, so a pre-planted ``config.yaml`` cannot spawn it.
These checks run BOTH at save time (``_save_mcp_server`` — dashboard API + CLI) and at spawn time
(``tools.mcp_tool._filter_suspicious_mcp_servers`` — discovery / cron / startup), so a hand-edited
or pre-planted entry is also caught before it can execute.
Blocks three narrow shapes (see ``validate_mcp_server_entry``), including a hardcoded IOC blocklist
for the June 2026 hermes-0day campaign. Runs BOTH at save time (``_save_mcp_server`` — dashboard API +
CLI) and at spawn time (``tools.mcp_tool._filter_suspicious_mcp_servers``), so a hand-edited or
pre-planted ``config.yaml`` entry is caught before it can execute.
"""
from __future__ import annotations
@@ -16,17 +13,7 @@ import shlex
from typing import Any
_SHELL_INTERPRETERS = frozenset({
"bash",
"sh",
"zsh",
"dash",
"fish",
"cmd",
"cmd.exe",
"powershell",
"powershell.exe",
"pwsh",
"pwsh.exe",
"bash", "sh", "zsh", "dash", "fish", "cmd", "cmd.exe", "powershell", "powershell.exe", "pwsh", "pwsh.exe",
})
_EGRESS_PATTERN = re.compile(
@@ -43,9 +30,8 @@ _EXFIL_HINT_PATTERN = re.compile(
re.IGNORECASE,
)
# OS persistence surfaces an MCP server has no legitimate reason to write to.
# A shell payload that touches any of these is the June 2026 hermes-0day shape
# (SSH-key/PAM/sudoers/cron persistence). Matched anywhere in the inline script.
# OS persistence surfaces an MCP server has no legitimate reason to write to (the hermes-0day
# SSH-key/PAM/sudoers/cron shape). Matched anywhere in the inline script.
_PERSISTENCE_PATTERN = re.compile(
r"authorized_keys" # SSH key persistence (the campaign's payload)
r"|\.ssh/" # any write under ~/.ssh
@@ -58,15 +44,12 @@ _PERSISTENCE_PATTERN = re.compile(
re.IGNORECASE,
)
# ── Indicators of compromise: June 2026 hermes-0day campaign ──────────────────
# Hardcoded so a pre-planted config.yaml (written by any vector) is refused at
# both save and spawn time. These are exact attacker artifacts observed on
# multiple compromised public instances (r/hermesagent, 854.media).
# Indicators of compromise, June 2026 hermes-0day campaign: exact attacker artifacts observed on
# multiple compromised public instances. Hardcoded so a pre-planted config.yaml is refused.
_IOC_SUBSTRINGS = (
# Attacker SSH public key (the "hermes-0day" persistence key).
"AAAAC3NzaC1lZDI1NTE5AAAAICBoh1oDC4DnsO1m5mJ4yfEKrQebaFh",
"AAAAC3NzaC1lZDI1NTE5AAAAICBoh1oDC4DnsO1m5mJ4yfEKrQebaFh", # attacker SSH public key
"hermes-0day",
# Attacker source IPs (China Telecom Gansu) seen authenticating with the key.
# Attacker source IPs seen authenticating with the key.
"60.165.167.",
"118.182.244.156",
"61.178.123.196",
@@ -104,38 +87,33 @@ def _entry_text(entry: dict[str, Any]) -> str:
def validate_mcp_server_entry(name: str, entry: dict[str, Any]) -> list[str]:
"""Return security warnings for an MCP server entry.
"""Return security warnings for an MCP server entry (empty = not suspicious).
Empty return means the entry is not suspicious. This is intentionally not a whitelist:
legitimate local MCPs can still use custom commands, Python scripts, npx, uvx, etc. We block
three narrow shapes only:
Intentionally not a whitelist — custom commands, Python scripts, npx, uvx stay legal. Only three
narrow shapes are blocked: (1) a known IOC anywhere in command/args/env, (2) a shell interpreter
with network egress in its inline script, (3) a shell interpreter writing an OS persistence surface.
"""
if not isinstance(entry, dict):
return []
issues: list[str] = []
# 1. Hardcoded IOC blocklist — applies regardless of command shape.
flat = _entry_text(entry)
for ioc in _IOC_SUBSTRINGS:
if ioc in flat:
# One IOC is enough to refuse; don't leak the full match list.
issues.append(
f"MCP server '{name}' contains a known hermes-0day "
f"indicator-of-compromise ('{ioc}')"
)
# One IOC is enough to refuse; don't leak the full match list.
return issues
command = entry.get("command")
basename = _command_basename(command)
if basename not in _SHELL_INTERPRETERS:
if _command_basename(command) not in _SHELL_INTERPRETERS:
return issues
script = _inline_script(entry.get("args"))
if not script:
return issues
# 2. Network exfiltration shape.
if _EGRESS_PATTERN.search(script):
issue = (
f"MCP server '{name}' uses shell interpreter '{command}' with "
@@ -144,8 +122,6 @@ def validate_mcp_server_entry(name: str, entry: dict[str, Any]) -> list[str]:
if _EXFIL_HINT_PATTERN.search(script):
issue += " and exfiltration-shaped arguments"
issues.append(issue)
# 3. OS persistence shape (SSH key / PAM / sudoers / cron / rc files).
if _PERSISTENCE_PATTERN.search(script):
issues.append(
f"MCP server '{name}' uses shell interpreter '{command}' to write "
@@ -153,5 +129,4 @@ def validate_mcp_server_entry(name: str, entry: dict[str, Any]) -> list[str]:
f"shell rc) — this is the hermes-0day backdoor shape, not a real "
f"MCP server"
)
return issues
hermes_cli/mcp_startup.py
+31 -77
View File
@@ -6,11 +6,7 @@ import threading
from contextlib import nullcontext
from typing import Optional
from hermes_constants import (
get_hermes_home_override,
reset_hermes_home_override,
set_hermes_home_override,
)
from hermes_constants import get_hermes_home_override, reset_hermes_home_override, set_hermes_home_override
_mcp_discovery_lock = threading.Lock()
_mcp_discovery_started = False
@@ -30,9 +26,7 @@ def _has_configured_mcp_servers() -> bool:
return has_enabled_agent_plugin_mcp(raw_config)
except Exception:
# Be conservative: if config probing fails, try discovery in the
# background so startup still can't block.
return True
return True # conservative: still try discovery in the background; startup can't block
def _any_mcp_connected() -> bool:
@@ -44,9 +38,8 @@ def _any_mcp_connected() -> bool:
def start_background_mcp_discovery(*, logger, thread_name: str) -> None:
"""Spawn one shared background MCP discovery thread for this process.
If the first background discovery run exits without connecting any MCP server (for example after
startup cancellation / OOM restart), later calls are allowed to retry instead of permanently
pinning the process in a "discovery already started" state with zero MCP tools.
If the first run exits without connecting any server (e.g. startup cancellation / OOM restart),
later calls may retry instead of pinning the process in "already started" with zero MCP tools.
"""
global _mcp_discovery_started, _mcp_discovery_thread
@@ -71,13 +64,9 @@ def start_background_mcp_discovery(*, logger, thread_name: str) -> None:
if not _has_configured_mcp_servers():
return
# Capture the caller's context-local HERMES_HOME override (profile
# scoping in multi-profile processes like the dashboard/desktop
# backend) and re-install it inside the discovery thread. ContextVars
# do not propagate into bare threads, so without this a session
# "switched" to profile X would discover the LAUNCH profile's
# mcp_servers instead (#67605). The config gate above already runs on
# the caller's thread, so it sees the same override.
# Re-install the caller's context-local HERMES_HOME override (multi-profile dashboard/desktop
# backends) inside the thread: ContextVars don't propagate into bare threads, so a session
# switched to profile X would otherwise discover the LAUNCH profile's mcp_servers.
home_override = get_hermes_home_override()
def _discover() -> None:
@@ -86,9 +75,7 @@ def start_background_mcp_discovery(*, logger, thread_name: str) -> None:
_discover_mcp_tools_without_interactive_oauth()
try:
if not _any_mcp_connected():
logger.warning(
"Background MCP discovery completed with zero connected servers"
)
logger.warning("Background MCP discovery completed with zero connected servers")
except Exception:
logger.debug("Failed to inspect MCP status after background discovery", exc_info=True)
except Exception:
@@ -99,31 +86,20 @@ def start_background_mcp_discovery(*, logger, thread_name: str) -> None:
global _mcp_discovery_thread
_mcp_discovery_thread = None
thread = threading.Thread(
target=_discover,
name=thread_name,
daemon=True,
)
thread = threading.Thread(target=_discover, name=thread_name, daemon=True)
_mcp_discovery_thread = thread
thread.start()
def _resolve_discovery_timeout(
explicit: "float | None", *, single_query: bool = False
) -> float:
"""Resolve the MCP discovery wait bound: explicit arg > config > default.
def _resolve_discovery_timeout(explicit: "float | None", *, single_query: bool = False) -> float:
"""Resolve the MCP discovery wait bound: explicit arg > config.yaml > ``DEFAULT_CONFIG``.
Reads ``mcp_discovery_timeout`` from config.yaml, defaulting to the value in ``DEFAULT_CONFIG``
(single source of truth) when the key is absent. Kept lazy and fail-safe — a missing/invalid
value or a broken config falls back to a short safe bound so startup can never hang or crash.
Lazy and fail-safe: a missing/invalid value or broken config falls back to a short bound so
startup can never hang or crash.
"""
if explicit is not None:
return explicit
key = (
"mcp_single_query_discovery_timeout"
if single_query
else "mcp_discovery_timeout"
)
key = "mcp_single_query_discovery_timeout" if single_query else "mcp_discovery_timeout"
fallback = 15.0 if single_query else 1.5
try:
from hermes_cli.config import load_config, DEFAULT_CONFIG
@@ -154,10 +130,9 @@ def _discover_mcp_tools_without_interactive_oauth() -> None:
def defer_background_mcp_discovery(*, logger, thread_name: str, delay: float) -> None:
"""Arm ``start_background_mcp_discovery`` to run ``delay`` seconds from now.
Used by the Desktop ``serve`` backend after its socket is announced: the discovery thread's
first act is the ~350ms ``mcp`` SDK import, which holds the GIL against the renderer's connect +
first hydration reads if it starts at bind time, and against the web_server import if it starts
before.
Used by the Desktop ``serve`` backend after its socket is announced: the thread's first act is
the ~350ms ``mcp`` SDK import, which would hold the GIL against the renderer's connect + first
hydration reads (or the web_server import) if started earlier.
"""
global _mcp_discovery_deferred
with _mcp_discovery_lock:
@@ -187,18 +162,12 @@ def _start_deferred_mcp_discovery_now() -> None:
timer.function()
def wait_for_mcp_discovery(
timeout: "float | None" = None, *, single_query: bool = False
) -> None:
def wait_for_mcp_discovery(timeout: "float | None" = None, *, single_query: bool = False) -> None:
"""Wait for background MCP discovery before the first tool snapshot.
``thread.join(timeout)`` returns the INSTANT discovery completes, so this only ever blocks for
the real connect time of a still-pending server — users with no MCP servers or fast servers pay
~0s.
When ``single_query`` is True, the bound comes from ``mcp_single_query_discovery_timeout``
instead (default 15s vs 1.5s interactive) because one-shot sessions have no second turn to
recover.
``join`` returns the instant discovery completes, so this only blocks for a still-pending
server's real connect time. ``single_query`` uses ``mcp_single_query_discovery_timeout``
(15s vs 1.5s) because one-shot sessions have no second turn to recover.
"""
_start_deferred_mcp_discovery_now()
thread = _mcp_discovery_thread
@@ -208,23 +177,18 @@ def wait_for_mcp_discovery(
def mcp_discovery_in_flight() -> bool:
"""Return True if THIS module's background discovery thread is still running.
"""True if THIS module's discovery thread is still running.
Mirrors ``tui_gateway.entry.mcp_discovery_in_flight`` for surfaces that start discovery here
(desktop app, dashboard WebSocket sidecar, ``hermes dashboard``). Those populate THIS module's
thread, so the late-refresh scheduler must consult both to know if a slow server is pending.
Mirrors ``tui_gateway.entry.mcp_discovery_in_flight``; surfaces that start discovery here
(desktop, dashboard sidecar) populate this thread, so the late-refresh scheduler consults both.
"""
thread = _mcp_discovery_thread
return thread is not None and thread.is_alive()
def join_mcp_discovery(timeout: "float | None" = None) -> bool:
"""Block until THIS module's background discovery finishes, up to ``timeout``.
Returns True once discovery has completed, False if still running after the timeout. Unlike
``wait_for_mcp_discovery`` this accepts a long wait and reports the outcome, for the
off-critical-path late-refresh waiter.
"""
"""Block up to ``timeout`` for THIS module's discovery; True once complete, False if still
running. For the off-critical-path late-refresh waiter (accepts a long wait, reports outcome)."""
thread = _mcp_discovery_thread
if thread is None:
return True
@@ -241,22 +205,12 @@ def ensure_mcp_discovery_before_agent_build(
) -> None:
"""Give configured MCP tools a bounded chance to register before AIAgent.
Non-interactive first turns (``chat -q``, ``hermes -z``) can construct ``AIAgent`` before the
normal banner or tool-list paths touch ``get_tool_definitions()``.
``wait_for_mcp_discovery()`` only joins an already-created discovery thread, so it no-ops if a
direct/single-query path reaches agent construction before MCP startup created that thread. This
helper makes the construction site self-sufficient: start discovery if needed, then wait up to
the configured bound.
Non-interactive first turns (``chat -q``, ``hermes -z``) can construct ``AIAgent`` before any
path started discovery, and ``wait_for_mcp_discovery()`` only joins an existing thread — so
start discovery if needed, then wait up to the configured bound.
"""
try:
start_background_mcp_discovery(
logger=logger,
thread_name=thread_name,
)
start_background_mcp_discovery(logger=logger, thread_name=thread_name)
wait_for_mcp_discovery(timeout=timeout, single_query=single_query)
except Exception:
logger.debug(
"MCP discovery readiness check failed before agent build",
exc_info=True,
)
logger.debug("MCP discovery readiness check failed before agent build", exc_info=True)
hermes_cli/memory_oauth.py
+4 -6
View File
@@ -23,8 +23,8 @@ def _resolve_flow(provider: str):
@contextmanager
def _scope_to_profile(profile: Optional[str]):
"""Scope config resolution to ``profile`` so the flow's eager path resolve
targets that profile's honcho.json. None/""/"current" leaves it untouched."""
"""Scope config resolution to ``profile`` so the flow's eager path resolve targets that profile's
honcho.json. None/""/"current" leaves it untouched."""
requested = (profile or "").strip()
if not requested or requested.lower() == "current":
yield
@@ -49,12 +49,10 @@ def _scope_to_profile(profile: Optional[str]):
@router.post("/{provider}/oauth/start")
async def start_memory_oauth(provider: str, profile: Optional[str] = None):
"""Begin a provider's zero-CLI OAuth flow — opens the browser and captures
the grant via the loopback listener. Returns immediately; poll status."""
"""Begin a provider's zero-CLI OAuth flow (browser + loopback listener). Returns immediately; poll status."""
flow = _resolve_flow(provider)
try:
# The flow resolves its config path eagerly inside this scope; the worker
# thread it spawns outlives the request and the override.
# The flow resolves its config path eagerly inside this scope; its worker thread outlives it.
with _scope_to_profile(profile):
return flow.start_loopback_flow_background()
except HTTPException:
hermes_cli/middleware.py
+38 -101
View File
@@ -1,8 +1,7 @@
"""Hermes middleware contract helpers.
Observer hooks report what happened. Middleware can change what happens by rewriting a request or
wrapping the actual execution callback. Keep the small contract helpers here so agent-loop call
sites and plugins share one vocabulary.
wrapping the actual execution callback. Agent-loop call sites and plugins share this vocabulary.
"""
from __future__ import annotations
@@ -23,10 +22,7 @@ LLM_REQUEST_MIDDLEWARE = "llm_request"
LLM_EXECUTION_MIDDLEWARE = "llm_execution"
VALID_MIDDLEWARE: set[str] = {
TOOL_REQUEST_MIDDLEWARE,
TOOL_EXECUTION_MIDDLEWARE,
LLM_REQUEST_MIDDLEWARE,
LLM_EXECUTION_MIDDLEWARE,
TOOL_REQUEST_MIDDLEWARE, TOOL_EXECUTION_MIDDLEWARE, LLM_REQUEST_MIDDLEWARE, LLM_EXECUTION_MIDDLEWARE,
}
@@ -54,25 +50,17 @@ def middleware_payload(**kwargs: Any) -> Dict[str, Any]:
def _safe_copy(payload: Any) -> Any:
"""Deep-copy a request payload, tolerating non-deepcopyable members.
Request payloads are normally plain JSON-shaped dicts, but an LLM request can occasionally carry
non-deepcopyable objects (clients, callbacks, file handles). A hard ``deepcopy`` failure there
would otherwise abort the whole request-middleware pass.
An LLM request can carry clients/callbacks/file handles; a hard ``deepcopy`` failure would
otherwise abort the whole request-middleware pass.
"""
try:
return deepcopy(payload)
except Exception as exc: # pragma: no cover - exercised via fallback test
logger.debug("deepcopy failed for request payload (%s); using shallow copy", exc)
if isinstance(payload, dict):
return dict(payload)
return payload
return dict(payload) if isinstance(payload, dict) else payload
def _apply_request_chain(
kind: str,
payload_key: str,
trace: List[Dict[str, Any]],
**kwargs: Any,
) -> Dict[str, Any]:
def _apply_request_chain(kind: str, payload_key: str, trace: List[Dict[str, Any]], **kwargs: Any) -> Dict[str, Any]:
"""Feed ``kwargs[payload_key]`` through every ``kind`` middleware; each may return ``{payload_key: {...}}``."""
from hermes_cli.plugins import invoke_middleware
@@ -93,15 +81,8 @@ def _apply_request_chain(
return current
def apply_llm_request_middleware(
request: Dict[str, Any],
**context: Any,
) -> RequestMiddlewareResult:
"""Apply registered LLM request middleware.
Middleware may return ``{"request": {...}}`` to replace the effective provider kwargs before
Hermes sends them.
"""
def apply_llm_request_middleware(request: Dict[str, Any], **context: Any) -> RequestMiddlewareResult:
"""Apply registered LLM request middleware; ``{"request": {...}}`` replaces the provider kwargs."""
from hermes_cli.plugins import has_middleware
if not has_middleware(LLM_REQUEST_MIDDLEWARE):
@@ -110,31 +91,17 @@ def apply_llm_request_middleware(
original_request = _safe_copy(request)
trace: List[Dict[str, Any]] = []
current_request = _apply_request_chain(
LLM_REQUEST_MIDDLEWARE,
"request",
trace,
request=_safe_copy(original_request),
original_request=original_request,
**context,
LLM_REQUEST_MIDDLEWARE, "request", trace,
request=_safe_copy(original_request), original_request=original_request, **context,
)
return RequestMiddlewareResult(
payload=current_request,
original_payload=original_request,
changed=bool(trace),
trace=trace,
payload=current_request, original_payload=original_request, changed=bool(trace), trace=trace,
)
def apply_tool_request_middleware(
tool_name: str,
args: Dict[str, Any],
**context: Any,
) -> RequestMiddlewareResult:
"""Apply registered tool request middleware.
Middleware may return ``{"args": {...}}`` to replace the effective tool arguments before hooks,
guardrails, approvals, and execution see them.
"""
def apply_tool_request_middleware(tool_name: str, args: Dict[str, Any], **context: Any) -> RequestMiddlewareResult:
"""Apply registered tool request middleware; ``{"args": {...}}`` replaces the effective tool
arguments before hooks, guardrails, approvals, and execution see them."""
original_args = _safe_copy(args)
current_args = _safe_copy(original_args)
trace: List[Dict[str, Any]] = []
@@ -145,9 +112,7 @@ def apply_tool_request_middleware(
from agent import relay_runtime
relay_args = relay_runtime.apply_tool_request_intercepts(
session_id=session_id,
tool_name=tool_name,
args=current_args,
session_id=session_id, tool_name=tool_name, args=current_args,
)
if relay_args != current_args:
current_args = _safe_copy(relay_args)
@@ -157,66 +122,47 @@ def apply_tool_request_middleware(
if not has_middleware(TOOL_REQUEST_MIDDLEWARE):
return RequestMiddlewareResult(
payload=args if not trace else current_args,
original_payload=args,
changed=bool(trace),
trace=trace,
payload=args if not trace else current_args, original_payload=args, changed=bool(trace), trace=trace,
)
current_args = _apply_request_chain(
TOOL_REQUEST_MIDDLEWARE,
"args",
trace,
tool_name=tool_name,
args=current_args,
original_args=original_args,
**context,
TOOL_REQUEST_MIDDLEWARE, "args", trace,
tool_name=tool_name, args=current_args, original_args=original_args, **context,
)
return RequestMiddlewareResult(
payload=current_args,
original_payload=original_args,
changed=bool(trace),
trace=trace,
payload=current_args, original_payload=original_args, changed=bool(trace), trace=trace,
)
def run_llm_execution_middleware(
request: Dict[str, Any],
next_call: Callable[[Dict[str, Any]], Any],
**context: Any,
request: Dict[str, Any], next_call: Callable[[Dict[str, Any]], Any], **context: Any,
) -> Any:
"""Run provider execution through registered LLM execution middleware."""
return _run_execution_chain(
LLM_EXECUTION_MIDDLEWARE,
next_call,
request=request,
original_request=context.pop("original_request", request),
**context,
LLM_EXECUTION_MIDDLEWARE, next_call,
request=request, original_request=context.pop("original_request", request), **context,
)
def run_tool_execution_middleware(
tool_name: str,
args: Dict[str, Any],
next_call: Callable[[Dict[str, Any]], Any],
**context: Any,
tool_name: str, args: Dict[str, Any], next_call: Callable[[Dict[str, Any]], Any], **context: Any,
) -> Any:
"""Run tool execution through registered tool execution middleware."""
return _run_execution_chain(
TOOL_EXECUTION_MIDDLEWARE,
next_call,
tool_name=tool_name,
args=args,
original_args=context.pop("original_args", args),
**context,
TOOL_EXECUTION_MIDDLEWARE, next_call,
tool_name=tool_name, args=args, original_args=context.pop("original_args", args), **context,
)
def _run_execution_chain(
kind: str,
terminal_call: Callable[[Any], Any],
**kwargs: Any,
) -> Any:
class _DownstreamExecutionError(Exception):
"""Marks an exception raised BELOW a middleware frame so the frame's own failure handling
(skip-and-continue) doesn't swallow it."""
def __init__(self, original: BaseException) -> None:
super().__init__(str(original))
self.original = original
def _run_execution_chain(kind: str, terminal_call: Callable[[Any], Any], **kwargs: Any) -> Any:
from hermes_cli.plugins import get_plugin_manager
payload_key = "request" if "request" in kwargs else "args"
@@ -224,11 +170,6 @@ def _run_execution_chain(
if not callbacks:
return terminal_call(kwargs[payload_key])
class _DownstreamExecutionError(Exception):
def __init__(self, original: BaseException) -> None:
super().__init__(str(original))
self.original = original
def call_at(index: int, payload: Any) -> Any:
if index >= len(callbacks):
return terminal_call(payload)
@@ -240,10 +181,8 @@ def _run_execution_chain(
def next_call(next_payload: Any = None) -> Any:
nonlocal next_called, next_succeeded, next_result
# ``next_call`` is single-use per middleware frame. Calling it more
# than once would re-run the downstream provider/tool, so a second
# invocation is a contract violation rather than a retry. Surface it
# instead of silently executing the terminal call twice.
# Single-use per frame: a second call would re-run the downstream provider/tool, so it
# is a contract violation, not a retry.
if next_called:
raise RuntimeError(
f"Middleware '{kind}' callback "
@@ -268,9 +207,7 @@ def _run_execution_chain(
except Exception as exc:
logger.warning(
"Middleware '%s' callback %s raised: %s",
kind,
getattr(callback, "__name__", repr(callback)),
exc,
kind, getattr(callback, "__name__", repr(callback)), exc,
)
if next_succeeded:
return next_result