Guardrails

Guardrails are the agent's nervous system: hooks that run around execution to validate, clean, redact, or observe, all without coupling that logic into the agent's handler.

Three stages

Stage	Signature	Runs	Purpose
pre	Payload -> Payload	before handle	clean / validate / reject input
post	Payload -> Payload	after output validation	redact / validate output
on_error	Exception -> None	on any failure	observe / alert (cannot suppress)

pre and post hooks fold over the payload (each receives the previous one's output), so they compose. They run in registration order.

Building a bundle

from weaveflow import Guardrails, agent, DataType

def strip_input(payload):
    return payload.with_value(payload.value.strip())

def redact_emails(payload):
    import re
    cleaned = re.sub(r"[\w.]+@[\w.]+", "[redacted]", payload.value)
    return payload.with_value(cleaned)

def alert(error):
    print("guardrail saw:", error)

rails = (
    Guardrails()
    .pre(strip_input)
    .post(redact_emails)
    .on_error(alert)
)

@agent(name="safe", input=DataType.TEXT, output=DataType.TEXT, guardrails=rails)
async def safe(ctx):
    return ctx.input.value

Rejecting input

To reject a payload, raise from a hook. Use GuardrailRejectionError for an actionable, typed failure:

from weaveflow.errors import GuardrailRejectionError

def max_len(payload):
    if len(payload.value) > 10_000:
        raise GuardrailRejectionError("Input too large", detail="max 10000 chars")
    return payload

rails = Guardrails().pre(max_len)

Error handling contract

When handle (or validation) raises:

1. every on_error hook is called with the exception,
2. the failure is logged with the agent name and error code,
3. a typed error is re-raised. A WeaveflowError passes through; anything else is wrapped in AgentExecutionError.

Guardrails observe errors; they never silently swallow them.