Add guarded host tool invocation

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.

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

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,

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()

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")