hermes - 💡(How to fix) Fix Feature request: Runtime extension hooks (transform_user_input, before_llm_call, intercept_tool_call) [1 participants]

Official PRs (…)
ON THIS PAGE

Recommended Tools

×6

Utilities matched from this issue’s tags and category — try them while you read without losing context.

GitHub issue graph ai analysis

Paste a GitHub issue URL. We fetch that issue, discover linked issues from bodies/comments/timeline, collect linked pull requests, and produce a structured English report.

The report is written in English Markdown for sharing and archival.

Helpful · Quick feedback

Loading…
GitHub stats
NousResearch/hermes-agent#18148Fetched 2026-05-01 05:53:40
View on GitHub
Comments
0
Participants
1
Timeline
4
Reactions
0
Participants
Timeline (top)
labeled ×4

Add Python-level runtime extension hooks to Hermes Agent that allow extensions to intercept, rewrite, and block agent behavior at critical points in the conversation loop. This is distinct from the existing shell-hook system (which fires external scripts as observers) and the plugin system (which registers tools and lifecycle callbacks). The proposed hooks would operate at the Python level within the agent loop, returning control values that can alter the flow.

Error Message

  • Block tool calls and return a custom error message to the agent (e.g., prevent write/edit when no valid plan exists) value: Any # rewritten message (for "rewrite") or error message (for "block") value: Any # text to inject (for inject actions) or error message (for "block") value: Any # replacement args (for "rewrite") or error message (for "block") function_result = json.dumps({"error": tool_result.value, "blocked_by_extension": True})

Skip handle_function_call, inject error result directly

Root Cause

  1. Extend shell hooks to support blocking — Rejected because shell scripts are too slow (process spawn per tool call) and the JSON serialization overhead would add latency to every tool call.

Fix Action

Fix / Workaround

Per-extension config (optional)

workflow-guard: plan_path: plans/active/plan.md blocked_tools: [write, edit, patch] allowed_paths: [plans/, handoff/]

Code Example

def transform_user_input(user_message: str, session_id: str, platform: str, ...) -> HookResult

---

@dataclass
class HookResult:
    action: str  # "pass" | "rewrite" | "block"
    value: Any   # rewritten message (for "rewrite") or error message (for "block")
    metadata: dict  # optional context for downstream hooks

---

def before_llm_call(messages: list, system_prompt: str, session_id: str, ...) -> HookResult

---

@dataclass
class HookResult:
    action: str  # "pass" | "inject_system" | "inject_user" | "block"
    value: Any   # text to inject (for inject actions) or error message (for "block")
    metadata: dict

---

def intercept_tool_call(tool_name: str, tool_args: dict, session_id: str, task_id: str, ...) -> HookResult

---

@dataclass
class HookResult:
    action: str  # "pass" | "block" | "rewrite"
    value: Any   # replacement args (for "rewrite") or error message (for "block")
    metadata: dict

---

from agent.extension_hooks import register_hook

   @register_hook("transform_user_input", priority=10)
   def my_input_hook(user_message, **kwargs):
       if is_implementation_request(user_message) and not has_valid_plan():
           return HookResult("rewrite", f"/start-feature {user_message}")
       return HookResult("pass")

---

from agent.extension_hooks import get_turn_state

state = get_turn_state(session_id)
state["plan_required"] = True
state["authorized_paths"] = ["plans/", "handoff/"]

---

~/.hermes/extensions/
├── workflow-guard/
│   ├── __init__.py    # calls register_hook() for each handler
│   ├── plan_check.py
│   └── ...
└── module-router/
    ├── __init__.py
    ├── routing.py
    └── ...

---

# ~/.hermes/config.yaml
extensions:
  enabled: true
  global_dir: ~/.hermes/extensions/
  project_local: true  # load from .hermes/extensions/ in cwd

  # Per-extension config (optional)
  workflow-guard:
    plan_path: plans/active/plan.md
    blocked_tools: [write, edit, patch]
    allowed_paths: [plans/, handoff/]

---

# In run_conversation():

# 1. transform_user_input (after sanitization)
user_message_result = self._invoke_hook_chain("transform_user_input", {
    "user_message": user_message,
    "session_id": self.session_id,
    "platform": self.platform,
})
if user_message_result.action == "rewrite":
    user_message = user_message_result.value
elif user_message_result.action == "block":
    return {"final_response": user_message_result.value, "messages": messages, "blocked": True}

# 2. before_llm_call (before API call, in the main loop)
llm_result = self._invoke_hook_chain("before_llm_call", {
    "messages": messages,
    "system_prompt": self._cached_system_prompt,
    "session_id": self.session_id,
})
if llm_result.action == "inject_system":
    self._cached_system_prompt += "\n\n" + llm_result.value
elif llm_result.action == "inject_user":
    messages.append({"role": "user", "content": llm_result.value})
elif llm_result.action == "block":
    messages.append({"role": "assistant", "content": llm_result.value})
    return {"final_response": llm_result.value, "messages": messages}

# 3. intercept_tool_call (before handle_function_call)
tool_result = self._invoke_hook_chain("intercept_tool_call", {
    "tool_name": function_name,
    "tool_args": function_args,
    "session_id": self.session_id,
    "task_id": effective_task_id,
})
if tool_result.action == "block":
    function_result = json.dumps({"error": tool_result.value, "blocked_by_extension": True})
    # Skip handle_function_call, inject error result directly
elif tool_result.action == "rewrite":
    function_args = tool_result.value
    function_result = handle_function_call(function_name, function_args, task_id)
else:
    function_result = handle_function_call(function_name, function_args, task_id)

---

// Pi extension (JavaScript)
export default function workflowGuardExtension(pi) {
  // Input hook: rewrite user input before prompt processing
  pi.on("input", (event) => {
    if (isImplementationRequest(event.text) && !hasValidPlan()) {
      return { rewrite: `/start-feature ${event.text}` };
    }
  });

  // Before-agent-start: inject system prompt additions
  pi.on("before_agent_start", (event) => {
    if (hasValidPlan()) {
      return { systemPrompt: "Follow TDD. Be surgical. Run tests before claiming done." };
    } else {
      return { message: "No valid plan exists. Create one before implementing." };
    }
  });

  // Tool-call hook: block specific tools
  pi.on("tool_call", (event) => {
    if (!hasValidPlan() && (event.tool === "edit" || event.tool === "write")) {
      return { block: true, reason: "No valid plan. Plan first." };
    }
  });
}
RAW_BUFFERClick to expand / collapse

Summary

Add Python-level runtime extension hooks to Hermes Agent that allow extensions to intercept, rewrite, and block agent behavior at critical points in the conversation loop. This is distinct from the existing shell-hook system (which fires external scripts as observers) and the plugin system (which registers tools and lifecycle callbacks). The proposed hooks would operate at the Python level within the agent loop, returning control values that can alter the flow.

Problem

Hermes Agent currently has two extension mechanisms:

  1. Shell hooks (VALID_HOOKS in hermes_cli/plugins.py) — fire external scripts as observers/notifications. Scripts receive JSON payloads via stdin and can return JSON via stdout, but the agent loop does not block or rewrite based on their output.

  2. Plugins — register tools, hook callbacks, and provide standalone/backend functionality.

Neither mechanism allows an extension to:

  • Rewrite user input before prompt expansion (e.g., redirect "fix the bug" to "/start-feature fix the bug")
  • Inject system prompt additions before the LLM call (e.g., append "do not implement code yet" based on runtime state)
  • Block tool calls and return a custom error message to the agent (e.g., prevent write/edit when no valid plan exists)
  • Maintain per-turn state across tool calls within the same conversation turn

This limits Hermes' ability to enforce structured development workflows at the runtime level. Compare to Pi Coding Agent's extension system, which provides input, before_agent_start, and tool_call hooks that mechanically enforce planning-before-implementation, module-aware exploration, and unauthorized-change blocking.

Proposed Solution

Add three new hook points to the agent loop in run_agent.py, with a Python-level hook registry that supports return values with control flow semantics.

New Hook Points

1. transform_user_input (equivalent to Pi's input hook)

Where: In run_conversation(), after user message sanitization, before messages are constructed.

Signature:

def transform_user_input(user_message: str, session_id: str, platform: str, ...) -> HookResult

HookResult:

@dataclass
class HookResult:
    action: str  # "pass" | "rewrite" | "block"
    value: Any   # rewritten message (for "rewrite") or error message (for "block")
    metadata: dict  # optional context for downstream hooks

Use case: Detect implementation-style prompts without a valid plan and rewrite them to a planning command.

2. before_llm_call (equivalent to Pi's before_agent_start)

Where: In run_conversation(), after system prompt construction and message assembly, before client.chat.completions.create().

Signature:

def before_llm_call(messages: list, system_prompt: str, session_id: str, ...) -> HookResult

HookResult:

@dataclass
class HookResult:
    action: str  # "pass" | "inject_system" | "inject_user" | "block"
    value: Any   # text to inject (for inject actions) or error message (for "block")
    metadata: dict

Use case: Inject implementation guardrails when a valid plan exists, or inject a planning reminder when no plan is found.

3. intercept_tool_call (equivalent to Pi's tool_call hook)

Where: In run_conversation(), inside the tool-calling loop, before handle_function_call() is invoked.

Signature:

def intercept_tool_call(tool_name: str, tool_args: dict, session_id: str, task_id: str, ...) -> HookResult

HookResult:

@dataclass
class HookResult:
    action: str  # "pass" | "block" | "rewrite"
    value: Any   # replacement args (for "rewrite") or error message (for "block")
    metadata: dict

Use case: Block write/edit tool calls targeting project code when no valid plan exists. Allow writes to plans/ and handoff/ directories.

Hook Registry

A new registry module (agent/extension_hooks.py) that:

  1. Discovers extensions from ~/.hermes/extensions/ directory and project-local .hermes/extensions/ directory
  2. Registers hook handlers via a simple interface:
    from agent.extension_hooks import register_hook
    
    @register_hook("transform_user_input", priority=10)
    def my_input_hook(user_message, **kwargs):
        if is_implementation_request(user_message) and not has_valid_plan():
            return HookResult("rewrite", f"/start-feature {user_message}")
        return HookResult("pass")
  3. Chains multiple hooks by priority (lower = earlier)
  4. Stops on block — if any hook returns action="block", the chain short-circuits
  5. Passes rewritten values through the chain — if a hook returns action="rewrite", subsequent hooks see the rewritten value

Per-Turn State

Extensions need a way to maintain state across tool calls within the same turn:

from agent.extension_hooks import get_turn_state

state = get_turn_state(session_id)
state["plan_required"] = True
state["authorized_paths"] = ["plans/", "handoff/"]

State is cleared at the end of each run_conversation() call.

Extension Loading Model

Extensions are Python packages with an __init__.py that registers hooks:

~/.hermes/extensions/
├── workflow-guard/
│   ├── __init__.py    # calls register_hook() for each handler
│   ├── plan_check.py
│   └── ...
└── module-router/
    ├── __init__.py
    ├── routing.py
    └── ...

Project-local extensions at .hermes/extensions/ are loaded after global ones, allowing project-specific overrides.

Configuration

# ~/.hermes/config.yaml
extensions:
  enabled: true
  global_dir: ~/.hermes/extensions/
  project_local: true  # load from .hermes/extensions/ in cwd

  # Per-extension config (optional)
  workflow-guard:
    plan_path: plans/active/plan.md
    blocked_tools: [write, edit, patch]
    allowed_paths: [plans/, handoff/]

Implementation Guide

Files to Modify

  1. agent/extension_hooks.py (new) — Hook registry, HookResult dataclass, turn state management, extension discovery

  2. run_agent.py — Add hook invocation at three points:

    • transform_user_input after user message sanitization
    • before_llm_call after system prompt construction / before API call
    • intercept_tool_call before each handle_function_call() call
  3. hermes_cli/config.py — Add extensions: section to DEFAULT_CONFIG

  4. hermes_cli/extensions_cmd.py (new) — CLI commands: hermes extensions list, hermes extensions enable/disable, hermes extensions doctor

  5. hermes_cli/commands.py — Add /extensions slash command

API Call Integration Points

In run_agent.py, the hook calls would be placed as follows:

# In run_conversation():

# 1. transform_user_input (after sanitization)
user_message_result = self._invoke_hook_chain("transform_user_input", {
    "user_message": user_message,
    "session_id": self.session_id,
    "platform": self.platform,
})
if user_message_result.action == "rewrite":
    user_message = user_message_result.value
elif user_message_result.action == "block":
    return {"final_response": user_message_result.value, "messages": messages, "blocked": True}

# 2. before_llm_call (before API call, in the main loop)
llm_result = self._invoke_hook_chain("before_llm_call", {
    "messages": messages,
    "system_prompt": self._cached_system_prompt,
    "session_id": self.session_id,
})
if llm_result.action == "inject_system":
    self._cached_system_prompt += "\n\n" + llm_result.value
elif llm_result.action == "inject_user":
    messages.append({"role": "user", "content": llm_result.value})
elif llm_result.action == "block":
    messages.append({"role": "assistant", "content": llm_result.value})
    return {"final_response": llm_result.value, "messages": messages}

# 3. intercept_tool_call (before handle_function_call)
tool_result = self._invoke_hook_chain("intercept_tool_call", {
    "tool_name": function_name,
    "tool_args": function_args,
    "session_id": self.session_id,
    "task_id": effective_task_id,
})
if tool_result.action == "block":
    function_result = json.dumps({"error": tool_result.value, "blocked_by_extension": True})
    # Skip handle_function_call, inject error result directly
elif tool_result.action == "rewrite":
    function_args = tool_result.value
    function_result = handle_function_call(function_name, function_args, task_id)
else:
    function_result = handle_function_call(function_name, function_args, task_id)

Reference: Pi Coding Agent Extension API

For comparison, here's how Pi's extensions work:

// Pi extension (JavaScript)
export default function workflowGuardExtension(pi) {
  // Input hook: rewrite user input before prompt processing
  pi.on("input", (event) => {
    if (isImplementationRequest(event.text) && !hasValidPlan()) {
      return { rewrite: `/start-feature ${event.text}` };
    }
  });

  // Before-agent-start: inject system prompt additions
  pi.on("before_agent_start", (event) => {
    if (hasValidPlan()) {
      return { systemPrompt: "Follow TDD. Be surgical. Run tests before claiming done." };
    } else {
      return { message: "No valid plan exists. Create one before implementing." };
    }
  });

  // Tool-call hook: block specific tools
  pi.on("tool_call", (event) => {
    if (!hasValidPlan() && (event.tool === "edit" || event.tool === "write")) {
      return { block: true, reason: "No valid plan. Plan first." };
    }
  });
}

Benefits

  1. Structured workflow enforcement — Planning-before-implementation can be mechanically enforced, not just prompted
  2. Module-aware exploration — Broad grep/find/ls can be blocked until routing artifacts are consulted
  3. Unauthorized change prevention — New module creation, structural changes can require procedural approval
  4. Per-turn state — Extensions can track what's been done in the current turn (artifacts read, modules authorized)
  5. Composable — Multiple extensions can coexist and chain their effects
  6. Backward compatible — Existing shell hooks, plugins, and skills continue to work unchanged

Alternative Approaches Considered

  1. Extend shell hooks to support blocking — Rejected because shell scripts are too slow (process spawn per tool call) and the JSON serialization overhead would add latency to every tool call.

  2. Use MCP servers for extension logic — Rejected because MCP is designed for tool provision, not agent-loop interception. The round-trip latency would be prohibitive for tool_call hooks.

  3. Skill-level enforcement via pre-action checks — This is the current approach. It works but relies on model compliance rather than mechanical enforcement. This FR adds the mechanical layer that skills can optionally leverage.

extent analysis

TL;DR

To address the limitations in Hermes Agent's extension mechanisms, implement a Python-level runtime extension hook system with three new hook points: transform_user_input, before_llm_call, and intercept_tool_call, allowing extensions to intercept, rewrite, and block agent behavior.

Guidance

  • Introduce a hook registry in agent/extension_hooks.py to manage hook handlers and their priorities.
  • Modify run_agent.py to invoke hooks at the specified points: after user message sanitization, before the LLM call, and before each tool call.
  • Implement the HookResult dataclass to standardize the return values from hook handlers, including actions like "pass", "rewrite", "block", and associated values or metadata.
  • Develop a system for extensions to maintain per-turn state, such as using a dictionary stored in get_turn_state(session_id).

Example

# Example hook handler for transform_user_input
from agent.extension_hooks import register_hook, HookResult

@register_hook("transform_user_input", priority=10)
def my_input_hook(user_message, **kwargs):
    if is_implementation_request(user_message) and not has_valid_plan():
        return HookResult("rewrite", f"/start-feature {user_message}")
    return HookResult("pass")

Notes

The proposed solution requires careful integration with the existing agent loop and tool call handling to ensure seamless functionality and minimal performance impact. Thorough testing of the hook system and its interactions with various extensions will be crucial.

Recommendation

Apply the proposed workaround by implementing the Python-level runtime extension hook system as described, allowing for more flexible and mechanical enforcement of structured development workflows within Hermes Agent. This approach provides a robust foundation for extensions to modify agent behavior at critical points, enhancing the overall functionality and usability of the system.

Vote matrix · Quick signals

Works
Did the solution work? Tap to confirm.
Easy Fix
Was it a quick fix?
Time Saver
Did it save you time?
Blocking
Was it severely blocking?
Common Issue
Are others likely hitting this too?
Flaky / Intermittent
Is it intermittent?
Verified / Reproducible
Can you reproduce it reliably?
Loading…

Still need to ship something?

×6

Another batch ranked right after the header list — different links, same matching logic.

Back to top recommendations

TRENDING

hermes - 💡(How to fix) Fix Feature request: Runtime extension hooks (transform_user_input, before_llm_call, intercept_tool_call) [1 participants]