`agentkit.middlewares`¶

The standard middleware set: tracing, meter, retry, fallback, memoize, output_coerce, compaction, guard, security.

See the Middlewares concept for the mental model.

Middlewares — the batteries. The app assembles two explicit, ordered chains (chat + tool) and hands them to the Invoker. There are two kinds:

BaseMiddleware (transform / guard / observe) — phase classes that don't control invoking next: MeterMiddleware, Compaction, Egress, Audit, SecurityMiddleware. Lowercase factories (meter(), compaction(), …) return instances. MeterMiddleware (the recording middleware) is named to disambiguate from runtime.Meter (the Protocol it records into).
Raw (call, next) (resilience / caching / instrumentation) — must re-invoke, skip, or wrap-with-a- context-manager next, which phase methods can't: retry/fallback (re-invoke), memoize/idempotent/ semantic_memoize (skip on hit), tracing (hold a span open across the call).

Typical chat chain: [tracing(), compaction(…), meter(), fallback([...]), retry(breaker=…)] Typical tool chain: [tracing(), meter(), egress(guardrail), idempotent(), audit(), retry(breaker=…)]

(compaction sits ahead of meter so the meter estimates tokens on the already-compacted transcript.)

SecurityMiddleware ¶

SecurityMiddleware(patterns: list[str] | None = None)

Bases: BaseMiddleware

Blocks malicious input before it reaches the model.

Scans the latest prompt against injection/unsafe-content patterns; on a match it emits an error observation and raises Blocked (refusing the call). Clean input passes through untouched. Drop it in the chat chain ahead of the model: Invoker(llm=…, chat_middleware=[SecurityMiddleware(), tracing(), …]).

Egress ¶

Egress(guardrail: Any)

Bases: BaseMiddleware

A tool with request.url_arg set has that URL checked (SSRF + allowlist) before it can run.

idempotent ¶

idempotent(*, store: Any = None) -> Middleware

Dedupe SIDE-EFFECTING tool calls on (run, scope, tool, args); a failure is never stored.

semantic_memoize ¶

semantic_memoize(
    *,
    vector: Any = None,
    threshold: float = 0.85,
    text: Callable[[Call], str] | None = None,
    when: Callable[[Call], bool] | None = None,
) -> Middleware

Near-duplicate reuse for READ-ONLY calls over a VectorPort (scored search), scope-isolated. NEVER attach to a side-effecting call — masking a real action behind a hit would be a correctness bug, so guard with when= (defaults to chat calls / explicitly read-only).

fallback ¶

fallback(
    models: list[str],
    *,
    classify_fn: Callable[
        [BaseException], ErrorClass
    ] = classify,
    breakers: dict[str, Any] | None = None,
) -> Middleware

models is the ordered fallover chain; the served model is left on request.model. Falls over on a PRE-stream transient failure / OPEN breaker; commits to a model once its first delta flows.

retry ¶

retry(
    *,
    breaker: Any = None,
    max_attempts: int = 3,
    classify_fn: Callable[
        [BaseException], ErrorClass
    ] = classify,
    sleep: Callable[[float], Any] | None = None,
) -> Middleware

Re-invoke on a PRE-stream failure (before the first delta — the common case: rate-limit/connect/OPEN breaker); once the first delta is yielded, commit and stream through (a mid-stream failure propagates — you can't un-send tokens). PERMANENT errors fail fast.

agentkit.middlewares¶