Middleware & Guardrails

What this is

This page is about the layers that wrap an agent's work without being the agent's work. Middleware is the machinery that runs around every model call, every tool call, and every whole turn — to cache, retry, log, truncate, and measure. Guardrails are a specialized branch of that same machinery whose job is not to help the call but to judge it, and to stop the run when a line is crossed.

This is the contract-and-code page — read the stories first

Middleware (concept) tells the why of wrapping cross-cutting concerns; Guardrails (concept) tells the why of safety checks that can halt. This page is the precise L1 implementation — the real classes, the one context type, the exact halt mechanism. We cross-link heavily and try not to repeat the stories.

Two analogies to hold the whole page

Middleware is airport security. Your bag travels inward through a stack of stations — ID check, X-ray, metal detector — reaches the gate, then you walk back outward past the same stations. Each station can inspect you on the way in and on the way out.

A guardrail is a checkpoint that can deny boarding. Most stations just look and wave you through. A guardrail is the one that, if it doesn't like what it sees, says "you're not getting on this flight" — the run stops dead.

This is the agents/ layer (L1). It owns the Middleware Protocol and MiddlewareContext dataclass itself (agents/middleware/_contracts.py) — kernel (L0) only contributes the dependency-free MiddlewareStage enum (kernel/agent/middleware.py); a kernel-minimal duplicate of the Protocol would have zero real consumers, so it isn't kept there. Middleware never reaches up into capabilities or fabric.

One middleware concept, not three

Earlier designs in this framework gave "agent middleware," "chat middleware," and "function middleware" separate context types and separate pipelines. That's gone: there is exactly one Middleware Protocol, one MiddlewareContext dataclass, and one MiddlewarePipeline per agent. MiddlewareStage (TURN/CHAT/TOOL) is a value on the context saying which moment a given instance represents — not a different kind of middleware. A middleware that only cares about one moment declares that via a stages class attribute; the pipeline itself skips calling process() for any stage a middleware didn't declare.

The onion model

A middleware is anything with one method, process(context, call_next). The pipeline threads many of them into nested layers around the real work. Each layer runs code before it calls inward, hands control to the next layer with call_next(), then runs code after as control unwinds back out.

diagram

Rendering diagram…

The request travels inward to the core, the response travels outward in reverse. A layer has three moves:

Wrap — do work, await call_next(), do more work. The normal onion.
Short-circuit — skip call_next() entirely. Inner layers never run. This is exactly how the cache returns a stored hit.
Abort — raise MiddlewareTermination. Inner layers never run, and the run stops. This is exactly how a guardrail blocks.

The contract

Middleware (agents/middleware/_contracts.py) is the whole interface — one shape, used identically by every middleware regardless of what it wraps:

python

class Middleware(Protocol):
    async def process(
        self, context: MiddlewareContext, call_next: Callable[[], Awaitable[None]]
    ) -> None: ...

Every layer has the same skeleton — before, inward, after:

python

class TimingMiddleware:
    async def process(self, context, call_next):
        start = time.monotonic()                                  # ── before
        await call_next()                                         # ── go inward
        context.metadata["elapsed"] = time.monotonic() - start    # ── after

Middleware returns nothing — the context IS the channel

process returns None. It never returns a value. Instead it reads and writes a shared context object that flows through the whole chain: read the inputs before call_next(), read or mutate the relevant result field (turn_result/chat_result/tool_result) / context.metadata after.

How the pipeline builds the chain

MiddlewarePipeline.execute(context, final) does two things: it filters out any middleware that didn't declare interest in context.stage, then folds the remaining list into a recursive call_next() chain where the innermost call_next() is final — the actual work being wrapped.

python

class MiddlewarePipeline:
    def __init__(self, middlewares=None) -> None:
        self._middlewares = list(middlewares or [])

    def add(self, middleware) -> None:
        self._middlewares.append(middleware)

    async def execute(self, context, final) -> None:
        active = [mw for mw in self._middlewares
                  if context.stage in getattr(mw, "stages", _ALL_STAGES)]

        async def build_chain(idx: int) -> None:
            if idx >= len(active):
                await final(context)              # innermost: do the real work
                return
            # this middleware's call_next() is "run the next index"
            await active[idx].process(context, lambda: build_chain(idx + 1))

        await build_chain(0)

So MiddlewarePipeline([A, B, C]) produces the call tree A( B( C( final ) ) ) for whichever of A/B/C declared interest in the current stage. The first middleware in the list is the outermost layer of the onion — for that stage.

Watching the chain unwind

diagram

Rendering diagram…

The one context type

Middleware wraps three different moments, but there is one context dataclass, MiddlewareContext (agents/middleware/_contracts.py). A stage field says which moment a given instance represents; fields that don't apply to that stage are simply None. It's a plain dataclass — no I/O, no surprises.

Field	Populated for	Meaning
`stage`, `agent_name`, `run_id`, `metadata`	always	shared by every stage
`session_id`, `turn_result`	`TURN`	one inbox message
`messages`	`TURN`, `CHAT`	the turn's full history (TURN) or this call's window (CHAT)
`system_instructions`, `tools`, `chat_result`	`CHAT`	one model call
`function_name`, `arguments`, `tool_result`	`TOOL`	one tool call

Three result fields, not one loose `result`

turn_result: AgentRunResult, chat_result: LLMResponse, and tool_result: InvocationResult are three separate, precisely-typed fields rather than one result: Any — the three result shapes are genuinely different classes, and this way both a middleware author and a type checker know exactly which one is meaningful for a given stage.

`InvocationResult` is frozen

InvocationResult (kernel/tools/chain.py) is the real wire-form RunContext.tool() returns — status: "ok" | "error" | "denied", text, structured. It's frozen (pydantic model_config = {"frozen": True}), so a middleware that wants to change it (like ContentTruncatorMiddleware) must reassign via context.tool_result = context.tool_result.model_copy(update=), not mutate in place.

TURN fires once per inbox message, not once per agent.run()

A single agent.run() call can process several inbox messages in a batch; each gets its own MiddlewareContext(stage=MiddlewareStage.TURN, ...) and its own AgentRunResult.

python

@dataclass(kw_only=True)
class MiddlewareContext:
    stage: MiddlewareStage
    agent_name: str
    run_id: str
    metadata: dict[str, Any] = field(default_factory=dict)

    session_id: str | None = None
    messages: list[ChatMessage] | None = None
    turn_result: AgentRunResult | None = None

    system_instructions: str | None = None
    tools: list[Tool] | None = None
    chat_result: LLMResponse | None = None

    function_name: str | None = None
    arguments: dict[str, Any] | None = None
    tool_result: InvocationResult | None = None

AgentRunResult is the terminal output of a whole turn, also defined here (in _contracts.py, to avoid a circular import with react.py):

python

@dataclass
class AgentRunResult:
    output: str
    status: str  # "success" | "error" | "max_iterations" | "paused"
    tool_calls: list[ToolCallRecord] = field(default_factory=list)
    run_id: str = ""
    error: str | None = None

The built-in middlewares

These ship in agents/middleware/*.py. Each is a plain class with a process method and a stages class attribute — no base class, no registration.

Middleware	Stage	What it does
`CacheMiddleware`	`TOOL`	Returns a stored `InvocationResult` for an identical `(function_name, args)` — short-circuits by skipping `call_next()`
`RetryMiddleware`	`CHAT`	Re-runs the inner call on transient errors with exponential backoff + jitter
`RateLimiterMiddleware`	`TURN`	Token-bucket limiter — raises `MiddlewareTermination` when the bucket is empty
`SchemaValidatorMiddleware`	`CHAT`	Parses model output against a Pydantic schema, stashes the parsed object in `context.metadata`
`AuditLoggerMiddleware`	`TURN`	Logs RUN START / END / ERROR with timing for compliance
`AgentTracingMiddleware` / `ChatTracingMiddleware` / `FunctionTracingMiddleware`	`TURN` / `CHAT` / `TOOL`	OpenTelemetry spans (falls back to DEBUG logs if OTel is absent); attached by default to every agent built via `create_assistant_agent()`
`ContentTruncatorMiddleware`	`TOOL`	Trims an over-long tool result to `max_chars`
`HistoryTruncatorMiddleware`	`CHAT`	Drops oldest non-system messages to keep the window under `max_messages`
`FileValidatorMiddleware`	`TOOL`	Vets file-path arguments (existence, extension, size) — raises `MiddlewareTermination` on a bad file

A short-circuit example: the cache

CacheMiddleware is the canonical short-circuit. On a hit it sets the result and returns without calling inward — the tool never actually runs:

python

class CacheMiddleware:
    stages = frozenset({MiddlewareStage.TOOL})

    async def process(self, context: MiddlewareContext, call_next) -> None:
        key = self._make_key(context)               # sha256 of {function, args}
        if key in self._cache:
            context.metadata["_cache_hit"] = True
            context.tool_result = self._cache[key]
            return                                  # ← skip call_next() — short-circuit

        context.metadata["_cache_hit"] = False
        await call_next()                           # miss: run the tool for real
        self._cache[key] = context.tool_result      # then remember the result

A wrap-and-retry example

RetryMiddleware wraps call_next() in a loop, sleeping with backoff between attempts and re-raising once max_retries is exhausted:

python

class RetryMiddleware:
    stages = frozenset({MiddlewareStage.CHAT})

    async def process(self, context: MiddlewareContext, call_next) -> None:
        attempt = 0
        while True:
            try:
                await call_next()
                return
            except self.retryable_exceptions as exc:
                if attempt >= self.max_retries:
                    raise                                   # give up — propagate
                delay = _backoff(attempt, self.base_delay, self.max_delay, self.jitter)
                await asyncio.sleep(delay)
                attempt += 1

Why `SchemaValidator` writes to `metadata`

SchemaValidatorMiddleware runs after the model answers, validates the text against a Pydantic schema you put in context.metadata["response_schema"], and stores the parsed object back in context.metadata["parsed"].

`LLMResponse` is frozen — the parsed object cannot live on it

LLMResponse is a frozen, slotted dataclass; you cannot attach a parsed attribute to it. So the validator deliberately puts its output in context.metadata, not on context.chat_result. Downstream code reads context.metadata["parsed"] and context.metadata["schema_valid"].

python

class SchemaValidatorMiddleware:
    stages = frozenset({MiddlewareStage.CHAT})

    async def process(self, context: MiddlewareContext, call_next) -> None:
        await call_next()                                    # let the model answer first
        schema = context.metadata.get("response_schema")
        if schema is None or context.chat_result is None:
            return
        text = context.chat_result.content[0].text           # first TextBlock
        try:
            obj = schema.model_validate_json(text)
            context.metadata["parsed"] = obj                # ← parsed object lives here
            context.metadata["schema_valid"] = True
        except Exception as exc:
            context.metadata["schema_valid"] = False        # fail open — don't halt

The three dispatch points

One MiddlewarePipeline is dispatched at three different moments in an agent's execution. Each moment builds its own MiddlewareContext (with the matching stage and only the relevant fields populated) and calls the same pipeline object.

diagram

Rendering diagram…

Stage	Fires	Good home for
`MiddlewareStage.TURN`	once per inbox message	input/prompt checks, rate limiting, audit logging
`MiddlewareStage.CHAT`	around every model call	token caps, retry, history truncation, output judging
`MiddlewareStage.TOOL`	around every tool call	caching, PII checks, tool-arg validation, file checks

How wiring works today

ReActAgent accepts a single middleware: MiddlewarePipeline (agents/middleware/pipeline.py). The identical pipeline object is dispatched at three real call sites, not by the Worker:

python

# agents/core/react.py — _handle_message(), once per inbox message
call_ctx = MiddlewareContext(stage=MiddlewareStage.TURN, ...)
await self.middleware.execute(call_ctx, _final)   # _final runs _react_loop()

# agents/runtime/context.py — RunContext.llm(), once per model call
chat_ctx = MiddlewareContext(stage=MiddlewareStage.CHAT, ...)
await middleware.execute(chat_ctx, _final)        # _final does the real generate call

# agents/runtime/context.py — RunContext.tool(), once per tool call
func_ctx = MiddlewareContext(stage=MiddlewareStage.TOOL, ...)
await middleware.execute(func_ctx, _final)        # _final does the real tool invoke

The Worker itself (agents/runtime/worker.py) doesn't reference middleware at all — it just calls await agent.run(ctx, inbox_msgs). A middleware that only cares about one stage declares that via its stages class attribute; the pipeline skips calling process() for any stage it didn't declare, so a TOOL-only middleware never fires — not even a no-op pass-through — during a TURN or CHAT dispatch.

Composing a pipeline

Order the list outermost-first — index 0 is the outer onion layer. A TURN-stage middleware and a CHAT-stage middleware sit in the exact same list; there's no separate slot to route them into:

python

from substrate.agents.middleware import (
    MiddlewarePipeline, RateLimiterMiddleware,
    CacheMiddleware, RetryMiddleware, SchemaValidatorMiddleware,
)

agent = ReActAgent(
    "bot", model=model,
    middleware=MiddlewarePipeline([
        RateLimiterMiddleware(max_rate=60),      # TURN
        RetryMiddleware(max_retries=3),          # CHAT — outermost of the two CHAT ones
        SchemaValidatorMiddleware(),             # CHAT — innermost, closest to the real call
        CacheMiddleware(),                       # TOOL
    ]),
)

Or use create_assistant_agent(...) (agents/factory.py), which already attaches the default tracing middlewares and appends yours to the same list:

python

from substrate.agents.factory import create_assistant_agent

agent = create_assistant_agent(
    model_client=model,
    middleware=[
        RateLimiterMiddleware(max_rate=60),
        RetryMiddleware(max_retries=3),
        SchemaValidatorMiddleware(),
        CacheMiddleware(),
    ],
)

Guardrails: middleware that can deny boarding

Guardrails are an ordinary middleware family with one defining behaviour: when their policy fires, they halt the run by raising MiddlewareTermination. They don't transform the call — they judge it and either let it pass (call_next) or block it (raise).

MiddlewareTermination is a kernel error (kernel/core/errors.py) and carries a human-readable .message. Its whole purpose is to be an intentional stop, not a crash.

Guardrail	Stage	What it catches
`PromptInjectionMiddleware`	`TURN`	"ignore previous instructions", "jailbreak", "developer mode", and similar override attempts in the last message
`ContentFilterMiddleware`	`TURN`	Banned keywords or regex patterns in the last message
`MaxTokenMiddleware`	`CHAT`	Input over a token budget (uses `tiktoken` if available, else a chars/token estimate)
`LLMJudgeMiddleware`	`CHAT`	Unsafe model output, classified by a secondary LLM
`PIIDetectionMiddleware`	`TOOL`	Email / phone / SSN / credit-card / IP patterns leaking into tool arguments
`ToolCallValidationMiddleware`	`TOOL`	Calls to blocked tools, calls outside an allow-list, or blocked argument patterns

How a guardrail halts

The shape is uniform: do the check around call_next(); if the policy fires, raise. MaxTokenMiddleware checks before the call (it's gating the input):

python

class MaxTokenMiddleware:
    stages = frozenset({MiddlewareStage.CHAT})

    async def process(self, context: MiddlewareContext, call_next) -> None:
        total_text = "".join(
            " ".join(b.text for b in m.content if isinstance(b, TextBlock)) + " "
            for m in context.messages
        )
        token_count = self._count_tokens(total_text.strip())
        if token_count > self.max_tokens:
            raise MiddlewareTermination(          # ← hard stop, before the model runs
                f"MaxToken: Input too long: {token_count} tokens — limit is {self.max_tokens}"
            )
        await call_next()

LLMJudgeMiddleware checks after the call (it's judging the output), and is the one guardrail that uses a second model — its own LLMClient — to classify:

python

class LLMJudgeMiddleware:
    stages = frozenset({MiddlewareStage.CHAT})

    async def process(self, context: MiddlewareContext, call_next) -> None:
        await call_next()                                   # let the primary model answer
        if not context.chat_result:
            return
        text = " ".join(b.text for b in context.chat_result.content if isinstance(b, TextBlock))
        try:
            judgment = await self._classify(text)           # ask the JUDGE model
            if not judgment.get("safe", True):
                raise MiddlewareTermination(
                    f"LLMJudge flagged as unsafe: {judgment.get('reason')}"
                )
        except MiddlewareTermination:
            raise                                           # a real block — propagate it
        except Exception as e:
            logger.warning("[LLMJudge] error — failing open: %s", e)   # judge itself broke

diagram

Rendering diagram…

The Worker treats a trip as a block, not a crash

When MiddlewareTermination propagates all the way out of agent.run, the Worker catches it in its exception handler and recognizes it as a guardrail trip — distinct from a budget exhaustion and, crucially, from an unexpected crash. This handling is entirely dispatch-agnostic: it doesn't matter whether the exception came from a TURN, CHAT, or TOOL-stage middleware.

python

# agents/runtime/worker.py — terminal exception handling
is_guardrail = isinstance(exc, MiddlewareTermination)
is_budget    = isinstance(exc, BudgetExhaustedError)
is_crash     = not is_guardrail and not is_budget

if is_guardrail or is_budget:
    for msg in inbox_msgs:
        await self._inbox.ack(agent.id, msg.id)      # ack — do NOT retry
else:
    for msg in inbox_msgs:
        await self._inbox.nack(agent.id, msg.id, error=str(exc))  # crash — retry

if is_guardrail:
    payload = {"error": f"Request blocked: {exc.message}", "status": "guardrail_tripped"}
# ... then append run.failed with this payload, release the lease as FAILED

The three outcomes diverge exactly here:

diagram

Rendering diagram…

Why ack, not nack?

A guardrail trip is the correct outcome — the request should be blocked. Re-running it would just trip the guardrail again. So the Worker acks the message (consumes it, no redelivery), writes run.failed with status: "guardrail_tripped", and returns a clean blocked response. A genuine crash instead nacks, so the durable runtime can retry it. See Durability for what retry-on-crash looks like.

Fail-open vs. fail-closed

Not every guardrail should halt on every problem. There is one convention worth internalizing:

The fail-open / fail-closed convention

Policy violation → fail CLOSED. The check worked and found something bad. Raise MiddlewareTermination and block the action. (Prompt injection matched, PII detected, token limit exceeded.)
The guardrail's OWN machinery errored → fail OPEN. The check itself broke (the judge model timed out, a parser blew up). Log a warning and let the run continue — don't take the agent down over your own hiccup.

LLMJudgeMiddleware is the textbook case: it re-raises a real MiddlewareTermination (fail closed) but swallows any other exception with a warning (fail open). SchemaValidatorMiddleware also fails open — a parse failure sets schema_valid = False rather than halting.

Choose per guardrail based on which mistake is worse for your use case: letting bad content through (a false negative) or blocking good content on an infrastructure blip (a false positive).

Composing guardrails

Guardrails are middleware, so they compose in the one MiddlewarePipeline like anything else — PromptInjectionMiddleware (TURN), MaxTokenMiddleware (CHAT), and PIIDetectionMiddleware (TOOL) all go in the same list, in the order you want them to wrap:

python

from substrate.agents.middleware import MiddlewarePipeline
from substrate.agents.middleware.guardrails import (
    PromptInjectionMiddleware, MaxTokenMiddleware, PIIDetectionMiddleware,
)

agent = ReActAgent(
    "bot", model=model,
    middleware=MiddlewarePipeline([
        PromptInjectionMiddleware(),           # TURN — check input first
        MaxTokenMiddleware(max_tokens=8000),   # CHAT — bound cost
        PIIDetectionMiddleware(),              # TOOL — vet tool arguments
    ]),
)

Where this lives

Piece	Location
`MiddlewarePipeline`	`agents/middleware/pipeline.py`
`MiddlewareStage` enum	`kernel/agent/middleware.py`
`Middleware` Protocol, `MiddlewareContext`, `AgentRunResult`, `ToolCallRecord`	`agents/middleware/_contracts.py`
Built-in middlewares (`Cache`, `Retry`, `RateLimiter`, …)	`agents/middleware/*.py`
Observability (`AgentTracing`, `ChatTracing`, `FunctionTracing`)	`agents/middleware/observability.py`
Guardrail middlewares	`agents/middleware/guardrails/`
`MiddlewareTermination`	`kernel/core/errors.py`
Pipeline dispatched once per turn	`agents/core/react.py` (`_handle_message`)
Pipeline dispatched per LLM/tool call	`agents/runtime/context.py` (`RunContext.llm()`/`.tool()`)
Trip handling (dispatch-agnostic)	`agents/runtime/worker.py` (exception handling only — no middleware reference)
Default tracing + `middleware=` wiring	`agents/factory.py`
Public exports	`agents/middleware/__init__.py`

Next: Tools: Toolbox & Invoker — how agents act on the world, and the dispatch path that tool-stage middleware and guardrails wrap.

Previous4 · The LLM Stack Next6 · Tools — Toolbox & Invoker

Middleware & Guardrails

What this is

The onion model

The contract

How the pipeline builds the chain

Watching the chain unwind

The one context type

The built-in middlewares

A short-circuit example: the cache

A wrap-and-retry example

Why SchemaValidator writes to metadata

The three dispatch points

Composing a pipeline

Guardrails: middleware that can deny boarding

How a guardrail halts

The Worker treats a trip as a block, not a crash

Fail-open vs. fail-closed

Composing guardrails

Where this lives

Why `SchemaValidator` writes to `metadata`