From 134249ff27525e5d669011278057e932d820f034 Mon Sep 17 00:00:00 2001 From: Offending Commit Date: Mon, 13 Jul 2026 21:13:30 -0500 Subject: [PATCH] feat: add guarded host tool invocation Some Hermes runtime capabilities, including send_message, are not registry-backed. Plugin handlers that dispatch them through tools.registry receive unknown-tool errors even when guard policy allows the operation.\n\nProvide one contract-tested invocation seam that resolves supported host handlers while preserving pre_tool_call blocking and post_tool_call observation. --- AGENTS.md | 3 + README.md | 30 +++++++ hermes_plugin_kit/__init__.py | 151 ++++++++++++++++++++++++++++++++++ tests/test_hermes_contract.py | 20 +++++ tests/test_kit.py | 80 ++++++++++++++++++ 5 files changed, 284 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index c12603b..e1cb869 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -9,6 +9,9 @@ plugin. - Keep `@tool` and `register_all` backward compatible. Use `@hook`, `plugin_skill`, and `register_plugin` for full plugin lifecycle registration. +- Use `invoke_host_tool` for host-managed capabilities such as `send_message`; + do not assume every Hermes capability is registered in `tools.registry`. + Nested host calls must remain visible to `pre_tool_call` and `post_tool_call`. - Use `tool_name(namespace, verb, noun)` for new tools and prefer explicit verbs such as `read`, `write`, and `patch`. Do not use Hermes agent-loop names (`memory`, `todo`, `session_search`, `delegate_task`) as plugin tools. diff --git a/README.md b/README.md index 5a95ba1..b4ce703 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,8 @@ the same boilerplate. `hermes-plugin-kit` makes them structurally impossible: - **Envelope + safety** — return a plain `dict` (or raise); the kit encodes the JSON string, catches exceptions, and always returns `str` from an `(args, **kwargs)` handler. +- **Host invocation** — call non-registry Hermes capabilities such as + `send_message` without bypassing plugin guard and audit hooks. ## Who it's for @@ -194,6 +196,34 @@ A handler returns a `dict` (becomes the success `data`), or raises (becomes a to error), or returns a `str` as an escape hatch (treated as already-encoded JSON). It must accept `(args, **kwargs)` — runtime keys like `task_id`/`session_id` arrive as kwargs. +## Calling host-managed capabilities + +Not every Hermes capability lives in `tools.registry`. In particular, +`send_message` is a host-managed runtime service, so calling +`registry.dispatch("send_message", ...)` from inside a plugin returns an unknown-tool +error. Use the kit's host invocation seam instead: + +```python +from hermes_plugin_kit import invoke_host_tool + +def deliver_generated_image(path: str, target: str, **runtime_context): + return invoke_host_tool( + "send_message", + { + "action": "send", + "target": target, + "message": f"MEDIA:{path}", + }, + **runtime_context, + ) +``` + +`invoke_host_tool` resolves the supported direct host handler and wraps the nested +operation with Hermes `pre_tool_call` and `post_tool_call` hooks. A blocking hook +prevents the handler from running. If the guard API is unavailable, invocation is +refused rather than sending without policy checks. `send_message` is the currently +supported host tool; unknown names fail explicitly. + ## Logging contract The kit logs under the decorated handler's module logger, so each plugin can diff --git a/hermes_plugin_kit/__init__.py b/hermes_plugin_kit/__init__.py index 48c1342..777f93d 100644 --- a/hermes_plugin_kit/__init__.py +++ b/hermes_plugin_kit/__init__.py @@ -17,6 +17,8 @@ - **Envelope + safety** — a handler returns a plain ``dict`` (or raises); the kit encodes the JSON string, wraps exceptions, and always returns ``str`` from an ``(args, **kwargs)`` signature, exactly as the registry requires. +- **Host invocation** — ``invoke_host_tool`` reaches supported Hermes runtime + services that are not registry-backed while preserving tool lifecycle hooks. Usage:: @@ -44,6 +46,7 @@ def register(ctx): from __future__ import annotations import functools +import importlib import inspect import json import logging @@ -59,6 +62,7 @@ def register(ctx): "hook", "plugin_skill", "register_plugin", + "invoke_host_tool", "PluginSkill", "RegistrationSummary", "register_all", @@ -79,6 +83,16 @@ def register(ctx): _AGENT_LOOP_TOOL_NAMES = frozenset({"todo", "memory", "session_search", "delegate_task"}) _RESERVED_NAMESPACE_PREFIXES = ("memory_",) _SKILL_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]+$") +_HOST_TOOL_IMPORTS = { + "send_message": ("tools.send_message_tool", "send_message_tool"), +} +_HOOK_CONTEXT_KEYS = ( + "task_id", + "session_id", + "tool_call_id", + "turn_id", + "api_request_id", +) @dataclass(frozen=True) @@ -347,6 +361,143 @@ def plugin_skill( raise ValueError("skill description is required") return PluginSkill(name, skill_path, description.strip(), bool(optional)) + +def _load_host_tool(name: str) -> Callable: + target = _HOST_TOOL_IMPORTS.get(name) + if target is None: + raise ValueError( + f"unsupported host tool {name!r}; supported: " + f"{', '.join(sorted(_HOST_TOOL_IMPORTS))}" + ) + module_name, handler_name = target + module = importlib.import_module(module_name) + handler = getattr(module, handler_name, None) + if not callable(handler): + raise RuntimeError( + f"Hermes host tool {name!r} has no callable {module_name}.{handler_name}" + ) + return handler + + +def _host_result_fields(result: str) -> tuple[str, str | None, str | None]: + try: + payload = json.loads(result) + except (TypeError, json.JSONDecodeError): + return "success", None, None + if not isinstance(payload, dict): + return "success", None, None + error = payload.get("error") + failed = payload.get("success") is False or payload.get("ok") is False + if error is not None or failed: + message = str(error or "host tool returned an unsuccessful result") + return "error", "host_tool_error", message + return "success", None, None + + +def _emit_host_post_tool_call( + name: str, + args: dict, + result: str, + *, + duration_ms: int, + status: str, + error_type: str | None, + error_message: str | None, + context: dict[str, Any], +) -> None: + try: + from hermes_cli.plugins import has_hook, invoke_hook + + if not has_hook("post_tool_call"): + return + hook_context = {key: context.get(key) or "" for key in _HOOK_CONTEXT_KEYS} + invoke_hook( + "post_tool_call", + tool_name=name, + args=args, + result=result, + duration_ms=duration_ms, + status=status, + error_type=error_type, + error_message=error_message, + **hook_context, + ) + except Exception: + logging.getLogger("hermes_plugin_kit").warning( + "%s: post_tool_call hook failed after host invocation", + name, + exc_info=True, + ) + + +def invoke_host_tool(name: str, args: dict | None = None, **context: Any) -> str: + """Invoke a Hermes host capability that is not registry-backed. + + Some Hermes capabilities, notably ``send_message``, are runtime services + rather than entries in ``tools.registry``. Plugin handlers must use this + seam instead of ``registry.dispatch`` so the direct host handler is found + while ``pre_tool_call`` and ``post_tool_call`` hooks still observe the + nested operation. + """ + handler = _load_host_tool(name) + tool_args = {} if args is None else args + if not isinstance(tool_args, dict): + raise TypeError("host tool args must be a dict") + + try: + from hermes_cli.plugins import resolve_pre_tool_block + except (ImportError, AttributeError) as exc: + raise RuntimeError( + "Hermes pre_tool_call guard API is unavailable; host invocation refused" + ) from exc + + hook_context = {key: context.get(key) or "" for key in _HOOK_CONTEXT_KEYS} + block_message = resolve_pre_tool_block(name, tool_args, **hook_context) + if block_message is not None: + result = json.dumps({"error": str(block_message)}, ensure_ascii=False) + _emit_host_post_tool_call( + name, + tool_args, + result, + duration_ms=0, + status="blocked", + error_type="plugin_block", + error_message=str(block_message), + context=context, + ) + return result + + started = time.perf_counter() + try: + result = handler(tool_args, **context) + except Exception as exc: + logging.getLogger("hermes_plugin_kit").exception( + "%s: host handler raised; error_type=%s", + name, + type(exc).__name__, + ) + result = json.dumps( + {"error": f"{name} host handler failed: {type(exc).__name__}"}, + ensure_ascii=False, + ) + if not isinstance(result, str): + result = json.dumps(result, ensure_ascii=False) + + duration_ms = int((time.perf_counter() - started) * 1000) + status, error_type, error_message = _host_result_fields(result) + _emit_host_post_tool_call( + name, + tool_args, + result, + duration_ms=duration_ms, + status=status, + error_type=error_type, + error_message=error_message, + context=context, + ) + return result + + def tool( *, toolset: str, diff --git a/tests/test_hermes_contract.py b/tests/test_hermes_contract.py index 9ba7a79..283df39 100644 --- a/tests/test_hermes_contract.py +++ b/tests/test_hermes_contract.py @@ -22,6 +22,7 @@ import types import unittest from pathlib import Path +from unittest.mock import patch import hermes_plugin_kit as hpk @@ -208,6 +209,25 @@ def test_lifecycle_calls_bind_to_real_plugincontext_signatures(self) -> None: description="Probe", ) + def test_host_tool_invocation_uses_real_non_registry_handler(self) -> None: + from tools import send_message_tool # type: ignore + + expected = '{"success": true, "message_id": "contract-probe"}' + args = { + "action": "send", + "target": "telegram:8670382527", + "message": "MEDIA:/opt/data/avatars/generated/contract-probe.png", + } + with patch.object( + send_message_tool, + "send_message_tool", + return_value=expected, + ) as handler: + result = hpk.invoke_host_tool("send_message", args) + + self.assertEqual(result, expected) + handler.assert_called_once_with(args) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_kit.py b/tests/test_kit.py index ad444af..8447be0 100644 --- a/tests/test_kit.py +++ b/tests/test_kit.py @@ -1,10 +1,12 @@ from __future__ import annotations import json +import sys import tempfile import types import unittest from pathlib import Path +from unittest.mock import Mock, patch import hermes_plugin_kit as hpk @@ -239,6 +241,84 @@ def test_hook_name_is_required(self) -> None: hpk.hook("") +class HostToolInvocationTests(unittest.TestCase): + def _runtime_modules(self, *, block_message=None, result='{"success": true}'): + plugins = types.ModuleType("hermes_cli.plugins") + plugins.resolve_pre_tool_block = Mock(return_value=block_message) + plugins.has_hook = Mock(return_value=True) + plugins.invoke_hook = Mock(return_value=[]) + + hermes_cli = types.ModuleType("hermes_cli") + hermes_cli.plugins = plugins + + send_message = types.ModuleType("tools.send_message_tool") + send_message.send_message_tool = Mock(return_value=result) + tools_package = types.ModuleType("tools") + tools_package.__path__ = [] + tools_package.send_message_tool = send_message + + modules = { + "hermes_cli": hermes_cli, + "hermes_cli.plugins": plugins, + "tools": tools_package, + "tools.send_message_tool": send_message, + } + return modules, plugins, send_message.send_message_tool + + def test_invokes_non_registry_host_tool_through_plugin_hooks(self) -> None: + modules, plugins, handler = self._runtime_modules() + args = { + "action": "send", + "target": "telegram:8670382527", + "message": "MEDIA:/opt/data/avatars/generated/portrait.png", + } + + with patch.dict(sys.modules, modules): + result = hpk.invoke_host_tool( + "send_message", + args, + session_id="session-1", + task_id="task-1", + ) + + self.assertEqual(json.loads(result), {"success": True}) + plugins.resolve_pre_tool_block.assert_called_once_with( + "send_message", + args, + task_id="task-1", + session_id="session-1", + tool_call_id="", + turn_id="", + api_request_id="", + ) + handler.assert_called_once_with(args, session_id="session-1", task_id="task-1") + post_call = plugins.invoke_hook.call_args + self.assertEqual(post_call.args, ("post_tool_call",)) + self.assertEqual(post_call.kwargs["tool_name"], "send_message") + self.assertEqual(post_call.kwargs["status"], "success") + + def test_blocked_host_tool_does_not_reach_handler(self) -> None: + modules, _plugins, handler = self._runtime_modules( + block_message="Outbound messaging is guarded" + ) + + with patch.dict(sys.modules, modules): + result = hpk.invoke_host_tool( + "send_message", + {"action": "send", "target": "telegram", "message": "hello"}, + ) + + self.assertEqual( + json.loads(result), + {"error": "Outbound messaging is guarded"}, + ) + handler.assert_not_called() + + def test_rejects_unknown_host_tool(self) -> None: + with self.assertRaisesRegex(ValueError, "unsupported host tool"): + hpk.invoke_host_tool("not_a_host_tool", {}) + + class RegisterPluginTests(unittest.TestCase): def _module(self, **attrs): module = types.ModuleType("sample_plugin")