deepset-ai / deepset-ai/haystack

Feature: GuardrailProvider interface for automated tool-call policy enforcement

Open
#10,821 9 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

P3
Dominant language
Python
Stars
26.6k
Forks
3.2k
Avg merge
1d 3h
Merged PRs (30d)
194

Description

Is your feature request related to a problem? Please describe.

Haystack's Agent component supports ConfirmationStrategy for human-in-the-loop tool approval (see hitl-hayhooks-redis-openwebui), but there is no programmatic way to enforce tool-call policies - things like rate limits, argument validation, scope restrictions, or audit logging - without requiring human interaction.

In practice, teams building production agents are stuck choosing between:

  1. Writing custom wrapper components around ToolInvoker to enforce policies
  2. Relying on ConfirmationStrategy alone, which blocks on human input and does not scale to automated guardrails
  3. Putting ad-hoc validation inside individual tool functions, which scatters policy logic across the codebase

This gap has come up in several existing discussions:

  • #8865 - Pipeline validation strictness options (error/warning/disable) showed demand for configurable enforcement levels
  • #9685 - Runtime tool injection into ToolInvoker highlighted the need for dynamic tool-level control at execution time
  • #7674 - The Tools abstraction epic established unified tool handling but did not cover pre-invocation policy enforcement
  • haystack-experimental#98 - Tool support discussion focused on invocation mechanics without touching on governance or safety constraints
  • haystack-experimental#209 - Agent component discussion with no mention of hooks, middleware, or guardrails

Describe the solution you'd like

A GuardrailProvider protocol that plugs into the Agent's tool invocation path alongside the existing ConfirmationStrategy, enabling automated policy enforcement without human interaction.

from typing import Protocol, runtime_checkable
from dataclasses import dataclass
from haystack.dataclasses import ChatMessage, Tool

@dataclass
class GuardrailResult:
    """Result of a guardrail evaluation."""
    allowed: bool
    reason: str | None = None
    modified_args: dict | None = None  # Optional: sanitized arguments

@runtime_checkable
class GuardrailProvider(Protocol):
    """Protocol for automated tool-call policy enforcement."""

    def evaluate_tool_call(
        self,
        tool: Tool,
        tool_call_args: dict,
        messages: list[ChatMessage],
        agent_state: dict | None = None,
    ) -> GuardrailResult:
        """
        Evaluate whether a tool call should proceed.

        Called after the LLM produces a tool call but before ToolInvoker
        executes it. Runs before ConfirmationStrategy (if any).

        Args:
            tool: The Tool about to be invoked.
            tool_call_args: Arguments the LLM generated for the tool call.
            messages: Conversation history up to this point.
            agent_state: Current agent state dict (if using state_schema).

        Returns:
            GuardrailResult indicating whether to proceed, deny, or
            proceed with modified arguments.
        """
        ...
Integration point in Agent

The Agent.__init__ would accept an optional guardrail_providers parameter:

@component
class Agent:
    def __init__(
        self,
        *,
        chat_generator: ChatGenerator,
        tools: ToolsType | None = None,
        guardrail_providers: list[GuardrailProvider] | None = None,  # NEW
        confirmation_strategies: dict[str | tuple[str, ...], ConfirmationStrategy] | None = None,
        # ... existing params
    ) -> None:

Execution order in the tool invocation path:

  1. LLM generates tool call
  2. GuardrailProviders evaluate (all must return allowed=True)
  3. ConfirmationStrategy runs (if configured for that tool)
  4. ToolInvoker executes the tool

This preserves backward compatibility - omitting guardrail_providers changes nothing.

Example usage
from haystack.components.agents import Agent

class RateLimitGuardrail:
    """Deny tool calls that exceed a per-tool rate limit."""

    def __init__(self, max_calls_per_tool: int = 10):
        self._counts: dict[str, int] = {}
        self._max = max_calls_per_tool

    def evaluate_tool_call(self, tool, tool_call_args, messages, agent_state=None):
        self._counts[tool.name] = self._counts.get(tool.name, 0) + 1
        if self._counts[tool.name] > self._max:
            return GuardrailResult(allowed=False, reason=f"Rate limit exceeded for {tool.name}")
        return GuardrailResult(allowed=True)

class ArgumentSanitizer:
    """Strip disallowed arguments before tool execution."""

    def __init__(self, blocked_keys: set[str]):
        self._blocked = blocked_keys

    def evaluate_tool_call(self, tool, tool_call_args, messages, agent_state=None):
        sanitized = {k: v for k, v in tool_call_args.items() if k not in self._blocked}
        return GuardrailResult(allowed=True, modified_args=sanitized)

agent = Agent(
    chat_generator=my_chat_generator,
    tools=[my_tool],
    guardrail_providers=[RateLimitGuardrail(max_calls_per_tool=5), ArgumentSanitizer({"password"})],
)

Describe alternatives you've considered

  1. Extending ConfirmationStrategy - This mixes up human approval with automated policy. ConfirmationStrategy is built for blocking human interaction; guardrails should be non-blocking and composable.

  2. Custom @component wrappers around ToolInvoker - This works but fragments the ecosystem. Every team ends up building their own, and policies are not portable across projects.

  3. Validation inside tool functions - This tangles business logic with policy enforcement, making policies invisible to the pipeline and impossible to audit in one place.

Additional context

  • The GuardrailProvider protocol follows Haystack's existing patterns: runtime-checkable Protocol (like ChatGenerator), dataclass results, and optional integration that preserves backward compatibility.
  • The interface is intentionally minimal - a single method - to keep the contract simple and implementations portable.
  • An async variant (async def evaluate_tool_call_async) could mirror the run/run_async pattern already used by Agent.
  • APort provides a reference implementation of this pattern for cross-framework agent guardrails.
  • This proposal complements rather than replaces ConfirmationStrategy - it addresses the automated policy enforcement gap while preserving human-in-the-loop workflows.

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start by reading the Agent tool-invocation path, ToolInvoker integration, and existing ConfirmationStrategy behavior. Review the related issues and discussions before deciding the interface and execution semantics; done would require an agreed design, implementation, and tests covering provider evaluation, denial, argument modification, ordering, and backward compatibility.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
ai, backend-api-design
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Needs clarification
Newbie friendliness
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.