guardrails

guardrails

GuardrailsEngine — security-aware inference engine wrapper.

Classes

SecurityBlockError

Bases: Exception

Raised when mode is BLOCK and security findings are detected.

GuardrailsEngine

GuardrailsEngine(
    engine: InferenceEngine,
    *,
    scanners: Optional[List[BaseScanner]] = None,
    mode: RedactionMode = REDACT,
    scan_input: bool = True,
    scan_output: bool = True,
    bus: Optional[EventBus] = None,
)

Bases: InferenceEngine

Wraps an existing InferenceEngine with security scanning.

Not registered in EngineRegistry — instantiated dynamically to wrap any engine at runtime.

PARAMETER	DESCRIPTION
engine	The wrapped inference engine. TYPE: InferenceEngine
scanners	List of scanners to run. Defaults to SecretScanner + PIIScanner. TYPE: Optional[List[BaseScanner]] DEFAULT: None
mode	Action taken on findings: WARN, REDACT, or BLOCK. TYPE: RedactionMode DEFAULT: REDACT
scan_input	Whether to scan input messages. TYPE: bool DEFAULT: True
scan_output	Whether to scan output content. TYPE: bool DEFAULT: True
bus	Optional event bus for publishing security events. TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/security/guardrails.py

def __init__(
    self,
    engine: InferenceEngine,
    *,
    scanners: Optional[List[BaseScanner]] = None,
    mode: RedactionMode = RedactionMode.REDACT,
    scan_input: bool = True,
    scan_output: bool = True,
    bus: Optional[EventBus] = None,
) -> None:
    self._engine = engine
    self._scanners: List[BaseScanner] = (
        scanners
        if scanners is not None
        else [
            SecretScanner(),
            PIIScanner(),
        ]
    )
    self._mode = mode
    self._scan_input = scan_input
    self._scan_output = scan_output
    self._bus = bus

Attributes

engine_id property

engine_id: str

Delegate to the wrapped engine.

Methods:

generate

generate(
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> Dict[str, Any]

Scan input, call wrapped engine, scan output.

Source code in src/diapason/security/guardrails.py

def generate(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> Dict[str, Any]:
    """Scan input, call wrapped engine, scan output."""
    messages = self._process_input_messages(messages)

    # Call wrapped engine
    response = self._engine.generate(
        messages,
        model=model,
        temperature=temperature,
        max_tokens=max_tokens,
        **kwargs,
    )

    # Scan output
    if self._scan_output:
        content = response.get("content", "")
        if content:
            result = self._scan_text(content)
            if not result.clean:
                response["content"] = self._handle_findings(
                    content, result, "output"
                )

    return response

stream async

stream(
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[str]

Use the same safe boundaries as rich output, including long secrets.

Source code in src/diapason/security/guardrails.py

async def stream(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[str]:
    """Use the same safe boundaries as rich output, including long secrets."""
    messages = self._process_input_messages(messages)

    async def chunks() -> AsyncIterator[StreamChunk]:
        async with aclosing(
            self._engine.stream(
                messages,
                model=model,
                temperature=temperature,
                max_tokens=max_tokens,
                **kwargs,
            )
        ) as source:
            async for token in source:
                yield StreamChunk(content=token)

    async with aclosing(self._checked_stream(chunks())) as checked:
        async for chunk in checked:
            if chunk.content is not None:
                yield chunk.content

stream_full async

stream_full(
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[StreamChunk]

Stream verified text without releasing tools or terminal data early.

Source code in src/diapason/security/guardrails.py

async def stream_full(
    self,
    messages: Sequence[Message],
    *,
    model: str,
    temperature: float = 0.7,
    max_tokens: int = 1024,
    **kwargs: Any,
) -> AsyncIterator[StreamChunk]:
    """Stream verified text without releasing tools or terminal data early."""
    messages = self._process_input_messages(messages)
    source = self._engine.stream_full(
        messages,
        model=model,
        temperature=temperature,
        max_tokens=max_tokens,
        **kwargs,
    )
    async with aclosing(self._checked_stream(source)) as checked:
        async for chunk in checked:
            yield chunk

list_models

list_models() -> List[str]

Delegate to wrapped engine.

Source code in src/diapason/security/guardrails.py

def list_models(self) -> List[str]:
    """Delegate to wrapped engine."""
    return self._engine.list_models()

health

health() -> bool

Delegate to wrapped engine.

Source code in src/diapason/security/guardrails.py

def health(self) -> bool:
    """Delegate to wrapped engine."""
    return self._engine.health()