Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,11 @@ plugin.

## Working Rules

- Keep `@tool` and `register_all` backward compatible. Use `@command`,
`@middleware`, `@hook`, `plugin_skill`, and `register_plugin` for full plugin
lifecycle registration.
- Keep `@tool`, slash-default `@command`, and `register_all` backward
compatible. Use `@command(type="cli")` for terminal subcommands and
`@command(type="slash")` for explicit in-session commands; use
`@middleware`, `@hook`, `plugin_skill`, and `register_plugin` for full
plugin lifecycle registration.
- Use `load_plugin_config` for effective `plugins.<name>` runtime settings;
current Hermes `PluginManifest` objects do not expose profile config. Use
`configure_stderr_logging` for operator-gated registration receipts instead
Expand Down
46 changes: 35 additions & 11 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,12 @@
![python](https://img.shields.io/badge/python-3.11%2B-blue)

`hermes-plugin-kit` is a tiny, dependency-free helper for authoring plugins for
[hermes-agent](https://github.com/NousResearch/hermes-agent). Decorate a slash
command with `@command`, a tool with `@tool`, or a lifecycle callback with
`@middleware` or `@hook`, then use `register_plugin` to register commands,
tools, middleware, hooks, and plugin-owned skills together. Existing tool-only
plugins can keep using `register_all`; the LLM-facing schema,
[hermes-agent](https://github.com/NousResearch/hermes-agent). Decorate an
in-session slash command or terminal CLI subcommand with `@command`, a tool
with `@tool`, or a lifecycle callback with `@middleware` or `@hook`, then use
`register_plugin` to register commands, tools, middleware, hooks, and
plugin-owned skills together. Existing tool-only plugins can keep using
`register_all`; the LLM-facing schema,
argument validation, structured logging, and the JSON result envelope are all
generated for you — correctly, every time.

Expand Down Expand Up @@ -140,6 +141,7 @@ Use the lifecycle entrypoint when a plugin provides more than tools:
import time
from pathlib import Path
from hermes_plugin_kit import (
CommandType,
MiddlewareKind,
command,
hook,
Expand All @@ -153,6 +155,19 @@ def valdris_status(raw_args):
"""Show the current Valdris plugin status."""
return build_status(raw_args)

def configure_valdris_cli(parser):
parser.add_argument("--scope", default="all")

@command(
"valdris",
type=CommandType.CLI,
help="Manage Valdris",
setup_fn=configure_valdris_cli,
)
def valdris_cli(args):
"""Manage Valdris from the terminal."""
return run_valdris_cli(scope=args.scope)

@middleware(MiddlewareKind.TOOL_REQUEST)
def normalize_tool_request(**kwargs):
args = {**kwargs["args"]}
Expand Down Expand Up @@ -187,12 +202,21 @@ def register(ctx):
return register_plugin(ctx, __name__, skills=SKILLS)
```

`@command` requires a bare lowercase kebab-case name without the leading slash.
Its handler receives the trailing command text unchanged and may return
`str | None` synchronously or asynchronously. The optional `args_hint` is
forwarded to Hermes for native command pickers. Command logs include only the
command name, elapsed time, result type, and argument character count, never
the raw arguments.
`@command` requires a bare lowercase kebab-case name. Slash commands are the
backward-compatible default: the handler receives trailing command text
unchanged and may return `str | None` synchronously or asynchronously. The
optional `args_hint` is forwarded to Hermes for native command pickers.

Use `type=CommandType.CLI` (or `type="cli"`) for a terminal command such as
`hermes valdris`. Its synchronous handler receives the parsed
`argparse.Namespace`. `setup_fn` configures that command's argparse subparser;
omit it for a command with no command-specific arguments. `help` defaults to
the first line of the resolved description. CLI commands cannot use
`args_hint`, and slash commands cannot use `help` or `setup_fn`.

Command logs include only the command name, elapsed time, result type, and,
for slash commands, the argument character count. Raw slash text and parsed
CLI argument values are never logged.

`@middleware` changes runtime behavior rather than merely observing it. Request
middleware rewrites the effective payload before Hermes continues; execution
Expand Down
157 changes: 135 additions & 22 deletions hermes_plugin_kit/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ def register(ctx):

from __future__ import annotations

import argparse
import copy
import functools
import importlib
Expand Down Expand Up @@ -82,6 +83,7 @@ def register(ctx):
"MediaPayload",
"ResolvedDeliveryTarget",
"MediaDeliveryResult",
"CommandType",
"MiddlewareKind",
"PluginSkill",
"RegistrationSummary",
Expand Down Expand Up @@ -144,6 +146,14 @@ class RegistrationSummary:
skipped_optional_skills: tuple[str, ...] = ()
commands: tuple[str, ...] = ()
middlewares: tuple[str, ...] = ()
cli_commands: tuple[str, ...] = ()


class CommandType(str, Enum):
"""Command surfaces currently supported by hermes-agent."""

SLASH = "slash"
CLI = "cli"


def log_registration_summary(
Expand All @@ -159,10 +169,11 @@ def log_registration_summary(
raise TypeError("summary must be a RegistrationSummary")
logger.info(
"hermes_plugin_kit: registered plugin lifecycle; plugin=%s; "
"commands=%s; tools=%s; middlewares=%s; hooks=%s; skills=%s; "
"skipped_optional_skills=%s",
"commands=%s; cli_commands=%s; tools=%s; middlewares=%s; hooks=%s; "
"skills=%s; skipped_optional_skills=%s",
clean_plugin_name,
",".join(summary.commands) or "<none>",
",".join(summary.cli_commands) or "<none>",
",".join(summary.tools) or "<none>",
",".join(summary.middlewares) or "<none>",
",".join(summary.hooks) or "<none>",
Expand Down Expand Up @@ -524,17 +535,38 @@ def _safe_context(kwargs: dict[str, Any]) -> dict[str, Any]:
# Decorators
# ---------------------------------------------------------------------------

def _noop_cli_setup(_parser: argparse.ArgumentParser) -> None:
"""Configure a CLI command that accepts no command-specific arguments."""


def command(
name: str,
description: str | None = None,
args_hint: str = "",
*,
type: CommandType | str = CommandType.SLASH,
help: str | None = None,
setup_fn: Callable[[argparse.ArgumentParser], None] | None = None,
) -> Callable:
"""Mark and instrument a Hermes in-session slash command.
"""Mark and instrument a Hermes slash or terminal CLI command.

Slash commands are the backward-compatible default. Their handler receives
the original ``raw_args`` string and may be synchronous or asynchronous.

``name`` is the bare command name without a leading slash. The wrapped
handler receives the original ``raw_args`` string and returns ``str | None``.
Synchronous and asynchronous handlers preserve their native callable shape.
CLI commands use ``type="cli"``. Their synchronous handler receives an
``argparse.Namespace``. ``setup_fn`` may configure the argparse subparser;
when omitted, the command accepts no command-specific arguments. ``help``
defaults to the first line of the resolved description.
"""
if isinstance(type, CommandType):
command_type = type
elif isinstance(type, str):
try:
command_type = CommandType(type.strip().lower())
except ValueError:
raise ValueError("command type must be 'slash' or 'cli'") from None
else:
raise TypeError("command type must be a CommandType or string")
if not isinstance(name, str) or not _COMMAND_NAME_RE.fullmatch(name):
raise ValueError(
"command name must be a bare lowercase kebab-case name matching "
Expand All @@ -544,7 +576,24 @@ def command(
raise TypeError("command description must be a string or None")
if not isinstance(args_hint, str):
raise TypeError("command args_hint must be a string")
if help is not None and not isinstance(help, str):
raise TypeError("command help must be a string or None")
if setup_fn is not None and not callable(setup_fn):
raise TypeError("command setup_fn must be callable or None")
clean_args_hint = args_hint.strip()
clean_help = (help or "").strip()
if command_type is CommandType.CLI and clean_args_hint:
raise ValueError("command args_hint is only valid for slash commands")
if command_type is CommandType.SLASH and help is not None:
raise ValueError("command help is only valid for CLI commands")
if command_type is CommandType.SLASH and setup_fn is not None:
raise ValueError("command setup_fn is only valid for CLI commands")
if (
command_type is CommandType.CLI
and setup_fn is not None
and inspect.iscoroutinefunction(setup_fn)
):
raise TypeError("CLI command setup_fn must be synchronous")

def decorate(fn: Callable) -> Callable:
explicit_description = (description or "").strip()
Expand All @@ -554,9 +603,11 @@ def decorate(fn: Callable) -> Callable:
f"@command {name!r}: a description is required "
"(docstring or description=)."
)
if command_type is CommandType.CLI and inspect.iscoroutinefunction(fn):
raise TypeError("CLI command handler must be synchronous")
log = logging.getLogger(fn.__module__ or "hermes_plugin_kit")

def log_invocation(raw_args: str) -> float:
def log_slash_invocation(raw_args: str) -> float:
started = time.perf_counter()
try:
args_chars = len(raw_args)
Expand All @@ -565,27 +616,46 @@ def log_invocation(raw_args: str) -> float:
log.debug("%s: invoked; args_chars=%d", name, args_chars)
return started

def log_cli_invocation() -> float:
started = time.perf_counter()
log.info("%s: invoked; type=cli", name)
return started

def log_failure(started: float, exc: Exception) -> None:
log.warning(
"%s: handler raised; elapsed_ms=%.2f; error_type=%s",
name,
(time.perf_counter() - started) * 1000,
type(exc).__name__,
exc.__class__.__name__,
)

def log_success(started: float, result: Any) -> None:
log.info(
"%s: ok; elapsed_ms=%.2f; result=%s",
name,
(time.perf_counter() - started) * 1000,
type(result).__name__,
result.__class__.__name__,
)

if inspect.iscoroutinefunction(fn):
if command_type is CommandType.CLI:

@functools.wraps(fn)
def cli_wrapper(args: argparse.Namespace) -> Any:
started = log_cli_invocation()
try:
result = fn(args)
except Exception as exc:
log_failure(started, exc)
raise
log_success(started, result)
return result

wrapper = cli_wrapper
elif inspect.iscoroutinefunction(fn):

@functools.wraps(fn)
async def async_wrapper(raw_args: str) -> str | None:
started = log_invocation(raw_args)
started = log_slash_invocation(raw_args)
try:
result = await fn(raw_args)
except Exception as exc:
Expand All @@ -599,7 +669,7 @@ async def async_wrapper(raw_args: str) -> str | None:

@functools.wraps(fn)
def sync_wrapper(raw_args: str) -> str | None:
started = log_invocation(raw_args)
started = log_slash_invocation(raw_args)
try:
result = fn(raw_args)
except Exception as exc:
Expand All @@ -610,13 +680,26 @@ def sync_wrapper(raw_args: str) -> str | None:

wrapper = sync_wrapper

command_help: str | None = None
command_setup_fn: Callable[[argparse.ArgumentParser], None] | None = None
if command_type is CommandType.CLI:
command_help = clean_help or next(
line.strip()
for line in doc.splitlines()
if line.strip()
)
command_setup_fn = setup_fn or _noop_cli_setup

setattr(
wrapper,
_COMMAND_SPEC_ATTR,
{
"name": name,
"type": command_type.value,
"description": doc,
"args_hint": clean_args_hint,
"help": command_help,
"setup_fn": command_setup_fn,
},
)
return wrapper
Expand Down Expand Up @@ -1423,7 +1506,7 @@ def register_plugin(
module: Any,
skills: tuple[PluginSkill, ...] | list[PluginSkill] = (),
) -> RegistrationSummary:
"""Register decorated commands, tools, middleware, hooks, and skills.
"""Register decorated slash/CLI commands, tools, middleware, hooks, and skills.

Unlike the backward-compatible :func:`register_all`, this lifecycle-level
entrypoint rejects distinct declarations that share a public name. Missing
Expand All @@ -1433,17 +1516,33 @@ def register_plugin(
module = sys.modules[module]
log = logging.getLogger(getattr(module, "__name__", "hermes_plugin_kit"))

commands: dict[str, Callable] = {}
slash_commands: dict[str, Callable] = {}
cli_commands: dict[str, Callable] = {}
tools: dict[str, Callable] = {}
middlewares: dict[str, Callable] = {}
hooks: dict[str, Callable] = {}
for _, obj in inspect.getmembers(module):
command_spec = getattr(obj, _COMMAND_SPEC_ATTR, None)
if command_spec:
existing = commands.get(command_spec["name"])
raw_command_type = command_spec.get("type", CommandType.SLASH.value)
try:
command_type = CommandType(raw_command_type)
except (TypeError, ValueError):
raise ValueError(
f"unsupported command type: {raw_command_type!r}"
) from None
command_registry = (
cli_commands
if command_type is CommandType.CLI
else slash_commands
)
existing = command_registry.get(command_spec["name"])
if existing is not None and existing is not obj:
raise ValueError(f"duplicate command name: {command_spec['name']}")
commands[command_spec["name"]] = obj
label = "CLI command" if command_type is CommandType.CLI else "command"
raise ValueError(
f"duplicate {label} name: {command_spec['name']}"
)
command_registry[command_spec["name"]] = obj

tool_spec = getattr(obj, _SPEC_ATTR, None)
if tool_spec:
Expand Down Expand Up @@ -1492,17 +1591,30 @@ def register_plugin(
)
skipped_skills.append(name)

registered_commands: list[str] = []
for name in sorted(commands):
obj = commands[name]
registered_slash_commands: list[str] = []
for name in sorted(slash_commands):
obj = slash_commands[name]
spec = getattr(obj, _COMMAND_SPEC_ATTR)
ctx.register_command(
name=spec["name"],
handler=obj,
description=spec["description"],
args_hint=spec["args_hint"],
)
registered_commands.append(name)
registered_slash_commands.append(name)

registered_cli_commands: list[str] = []
for name in sorted(cli_commands):
obj = cli_commands[name]
spec = getattr(obj, _COMMAND_SPEC_ATTR)
ctx.register_cli_command(
name=spec["name"],
help=spec["help"],
setup_fn=spec["setup_fn"],
handler_fn=obj,
description=spec["description"],
)
registered_cli_commands.append(name)

registered_tools: list[str] = []
for name in sorted(tools):
Expand Down Expand Up @@ -1531,7 +1643,8 @@ def register_plugin(
registered_skills.append(skill.name)

summary = RegistrationSummary(
commands=tuple(registered_commands),
commands=tuple(registered_slash_commands),
cli_commands=tuple(registered_cli_commands),
tools=tuple(registered_tools),
middlewares=tuple(registered_middlewares),
hooks=tuple(registered_hooks),
Expand Down
Loading