DEV Community: Antonio Berben

Route each prompt to the right model, with a decision model in your cluster and one agentgateway policy

Antonio Berben — Thu, 01 Oct 2026 06:30:00 +0000

A hands-on lab, every file included. A 170-line ExtProc asks a decision model which tier should answer, and agentgateway routes on the reply. Runs on kind, on a CPU, and the prompts stay home.

Every platform team I talk to lands in the same meeting sooner or later. The agents send everything to the biggest model. Finance wants the bill down. Somebody says "route by prompt", somebody else opens the vLLM Semantic Router docs and finds a full platform: classifiers, a semantic cache, tooling. It is a good project, and if you need what it does, it is the right answer. This team needed one decision, and that is a smaller problem.

The decision is small. Does this prompt need the strong model, or will the fast one do? Written down like that, it is a classification with two labels. And this September a new kind of model showed up that does exactly that, and nothing else.

So this lab is the small version. One ExtProc, 172 lines of Go, that reads the prompt, asks a decision model which tier should answer, and hands the answer to agentgateway as a header. agentgateway does the routing, the way it already routes everything else. Everything you need is in this article: the Go file, five manifests, and the commands in order. Nothing to clone. You will run it on kind, on your laptop, with the classifier on a CPU inside the cluster.

One thing to be clear about before the first command. The classifier is real: Laya runs in the cluster and every probability in this article came out of it. The two models it routes to are not. They are two nginx pods that answer any chat completion with a fixed JSON naming their tier. That is not to save money. It is so you can see, in the response body, which model the gateway actually sent the request to. A real model answers the question, which tells you nothing about the route it took; a stand-in that says [strong-model] answered does. Swap them for vLLM or a hosted provider at the end and nothing in the routing changes.

A model that does not write

On 15 September 2026 TypeSafe AI released Jev, which they call a System One model. You send it a block of state and one or more typed questions, and it answers all of them in one pass with a choice, a score or a yes-or-no probability. It never generates a sentence. Output tokens are free because there are none. The hosted API is a single endpoint, POST /v1/systemone, and the response carries probabilities and a confidence for every question you asked.

Within a week the open weights community had built the same shape on top of open models. The one I use here is Laya: a ModernBERT-large classifier of 421 million parameters, Apache 2.0, with a built-in server that exposes the same /v1/systemone contract. It runs on a CPU. On my MacBook, inside kind, a decision plus the whole gateway round trip takes about 200 ms.

Why does this matter for routing? Three properties, and they are the three things a router wants:

The answer is one of the labels you sent. There is no text to parse and no way to get a label you did not define.
You get a probability for every label, so "not sure" is a number you can compare against a threshold instead of a feeling.
It is small. The question is short, the answer is a few floats, and there is no reason for it to cost more than a lookup.

The request looks like this. The state is the last user message. The question is a choice with two criteria, and the criteria are plain English descriptions of what each tier is for:

{
  "model": "jev-latest",
  "state": "Design a multi-region rate limiter that stays consistent under network partition, compare token bucket with sliding window, then implement it in Go with unit tests.",
  "questions": {
    "tier": {
      "type": "choice",
      "instructions": "Which model tier should answer this user request?",
      "criteria": {
        "fast": "Short factual questions, greetings, translations, one-line definitions, rewording a sentence. Anything a small model answers well in a few lines.",
        "strong": "Multi-step reasoning, writing or debugging code, proofs, legal, financial or security analysis, long structured documents. Anything where a wrong answer is costly."
      }
    }
  }
}

And this is the shape of what comes back, from Laya, inside the cluster:

{
  "model": "laya-rl-agent",
  "answers": {
    "tier": {
      "type": "choice",
      "choice": "strong",
      "probabilities": {"fast": 0.259, "strong": 0.741},
      "confidence": 0.16
    }
  },
  "usage": {"input_tokens": 127, "output_tokens": 0}
}

The router reads choice and the probability of that choice. Jev also returns a calibrated confidence in the 0 to 1 range; Laya's confidence lives on its own scale, so the probability is the signal that works for both, and it is the one the router compares.

Fast or strong is one question, not the only one

I picked cost tiers for this lab because they are the easiest to see, and for a while that was all I had in mind. It was a conversation with my colleague Rinor Maloku that widened it. The router only knows labels and their descriptions; it has no idea what "fast" or "strong" mean. So the labels can be anything: effort, domain, task type. Write the descriptions, and the same router classifies on that instead. That is what makes "model": "auto" worth having. It no longer means "cheap or expensive". It means "the right model for this request". Some axes I have seen teams need:

Effort. trivial, moderate, deep: how much reasoning the answer needs, which is what decides whether a reasoning model earns its latency.
Task type. code, prose, data, conversation: send code to a coding model and a customer reply to a model that writes well, instead of asking one model to do both.
Domain. legal, finance, support, general: the tier is a fine-tuned or RAG-backed model per domain, and the router is the only place that needs to know the mapping.
Sensitivity. contains-personal-data, contains-credentials, clean: not a tier at all, but a label the route can use to keep some prompts on the model that never leaves the cluster.
Language. es, en, de: route to the model that is strong in the user's language, or add the label as a header your telemetry can group by.

One request to the decision model can carry several of these questions at once; they are evaluated in parallel and isolated from each other. This lab asks one, so the plumbing stays visible. Adding a second is another entry in the questions map and another header.

Where the decision has to happen

Gateway API routes on paths and headers. It has no body matcher, by design, because a route is supposed to be cheap. So a decision made from the prompt has to become a header before the route is chosen, or it changes nothing.

agentgateway, one of the open source projects under the Agentic AI Foundation, has a phase for exactly this. A traffic policy attached to the Gateway with phase: PreRouting runs its filters before route selection. Put an extProc there, let it read the buffered body and write a header, and the HTTPRoute that comes next sees the header as if the client had sent it.

Six steps, and the agent sees none of them:

The agent sends {"model": "auto", "messages": [...]} to /v1/chat/completions.
agentgateway buffers the body and streams it to the router over gRPC, the Envoy ext_proc protocol.
The router takes the last user message and asks the decision model the choice question above.
The router writes x-selected-model: strong-model on the request, plus three x-decision-* headers that explain itself.
The HTTPRoute matches on x-selected-model and picks the AgentgatewayBackend for that tier.
The backend pins the model name and rewrites it into the body, so the client's auto never reaches a model server.

Two fields in the policy carry all of this. phase: PreRouting, because at the default PostRouting the backend is already chosen when the header appears. And requestBodyMode: Buffered, because the router needs the whole prompt before it can answer.

One router, and the URL decides where your prompts go

The router speaks one contract, so where the decision model lives is a deployment choice, not a code change. DECISION_URL points at api.typesafe.ai and you are using Jev, hosted in the United States. It points at a Service in your namespace and you are using Laya, and nothing leaves the cluster.

That choice is worth a paragraph, because it is easy to make by accident. A router reads every prompt by construction. Not a sample, not the ones that fail: all of them, before any policy has looked at them. Whatever the router sends to its classifier, it sends for one hundred percent of your traffic. With Jev that is the last user message, without the system prompt and without the history, under TypeSafe's data processing addendum with EU standard contractual clauses. With Laya it is an HTTP call to the pod next door.

For me, in Europe, that is the whole reason the lab runs Laya. Not because the hosted option is worse at the job. Because the classifier is the one component that sees everything, and I want to decide where it sits with my eyes open. The same binary, the same policy and the same route serve both, and switching is one variable and one key.

What you need

Docker, kind, kubectl, helm, jq and curl. No Go toolchain: the router compiles inside its pod.
About 6 GB of RAM free for kind, most of it for Laya.
Twenty minutes. The router pod fetches two Go modules and compiles (about two minutes); the Laya pod installs its package and downloads the checkpoint from Hugging Face (about five).
No API key of any kind.

Everything below goes into the agentgateway-system namespace. Save each YAML block to the file name in its heading, or paste it into kubectl apply -f -.

Step 0: a cluster and a gateway

kind create cluster --name jev-router

kubectl apply --server-side --force-conflicts \
  -f https://github.com/kubernetes-sigs/gateway-api/releases/download/v1.6.1/standard-install.yaml

helm upgrade --install agentgateway-crds oci://cr.agentgateway.dev/charts/agentgateway-crds \
  --version 1.5.0 --create-namespace --namespace agentgateway-system
helm upgrade --install agentgateway oci://cr.agentgateway.dev/charts/agentgateway \
  --version 1.5.0 --namespace agentgateway-system --wait

One Gateway, one HTTP listener. Everything in the lab is reached through a port-forward, nothing is exposed. 00-gateway.yaml:

apiVersion: gateway.networking.k8s.io/v1
kind: Gateway
metadata:
  name: agentgateway-proxy
  namespace: agentgateway-system
spec:
  gatewayClassName: agentgateway
  listeners:
  - name: http
    port: 80
    protocol: HTTP
    allowedRoutes:
      namespaces:
        from: Same

kubectl apply -f 00-gateway.yaml

Step 1: two tiers

The two tiers are stand-ins on purpose: an nginx that answers every chat completion with a fixed OpenAI-shaped JSON naming its tier. llm-fast answers [fast-model] answered, llm-strong answers [strong-model] answered. The point is not to avoid a provider bill. It is that the answer itself becomes the evidence of the route: when the body says strong-model, the request went to the strong tier, and there is no other way it could have got there. A real model would answer the question, and the route would stay invisible. Replace them with vLLM, Ollama or a hosted provider once you trust the routing; the policy, the route and the router do not change.

The fast tier is a ConfigMap, a Deployment, a Service and one AgentgatewayBackend. 10-tier-fast.yaml:

# Two stand-in model servers. Each answers every chat completion with a fixed
# OpenAI-shaped JSON that names its tier, so you can see which one the gateway picked.
apiVersion: v1
kind: ConfigMap
metadata:
  name: llm-fast
  namespace: agentgateway-system
data:
  default.conf: |
    server {
      listen 8080;
      location / {
        default_type application/json;
        return 200 '{"id":"mock","object":"chat.completion","model":"fast-model","choices":[{"index":0,"message":{"role":"assistant","content":"[fast-model] answered"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}';
      }
    }
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-fast
  namespace: agentgateway-system
spec:
  selector: {matchLabels: {app: llm-fast}}
  template:
    metadata: {labels: {app: llm-fast}}
    spec:
      containers:
      - name: nginx
        image: nginx:1.27-alpine
        ports: [{containerPort: 8080}]
        volumeMounts: [{name: conf, mountPath: /etc/nginx/conf.d}]
      volumes: [{name: conf, configMap: {name: llm-fast}}]
---
apiVersion: v1
kind: Service
metadata:
  name: llm-fast
  namespace: agentgateway-system
spec:
  selector: {app: llm-fast}
  ports: [{port: 8080}]
---
# One gateway backend per tier. `model` is both the value the router writes into
# x-selected-model and the value the gateway rewrites into the body before forwarding.
apiVersion: agentgateway.dev/v1alpha1
kind: AgentgatewayBackend
metadata:
  name: llm-fast
  namespace: agentgateway-system
spec:
  ai:
    provider:
      openai:
        model: fast-model
      host: llm-fast.agentgateway-system.svc.cluster.local
      port: 8080

The strong tier is the same file with the names swapped, so let sed write it:

kubectl apply -f 10-tier-fast.yaml
sed 's/fast-model/strong-model/g; s/llm-fast/llm-strong/g' 10-tier-fast.yaml | kubectl apply -f -

The backend's model field does two jobs. It is the value the router will write into x-selected-model, and it is the value agentgateway substitutes into the body's model field before forwarding. Same name in three places: the router's table, the backend, and the HTTPRoute.

Step 2: Laya, in the cluster

A python:3.11-slim pod that installs laya[serve] with a CPU torch wheel at start and runs laya-serve. 20-laya.yaml:

# Laya (ModernBERT, Apache 2.0) behind its Jev-compatible server, on CPU.
# The pod installs the package at start and downloads the checkpoint from
# Hugging Face; the startup probe gives it up to 25 minutes.
apiVersion: apps/v1
kind: Deployment
metadata:
  name: laya
  namespace: agentgateway-system
spec:
  progressDeadlineSeconds: 1800
  selector: {matchLabels: {app: laya}}
  template:
    metadata: {labels: {app: laya}}
    spec:
      # The Service is called laya, so Kubernetes would inject LAYA_PORT=tcp://...
      # and laya-serve reads that as its port. Links off, port explicit.
      enableServiceLinks: false
      containers:
      - name: laya
        image: python:3.11-slim
        command: ["sh", "-c"]
        args:
        - pip install --quiet torch --index-url https://download.pytorch.org/whl/cpu &&
          pip install --quiet "laya[serve]==0.3.21" && exec laya-serve
        env:
        - {name: LAYA_HOST, value: "0.0.0.0"}
        - {name: LAYA_PORT, value: "8000"}
        - {name: LAYA_DEVICE, value: cpu}
        - {name: LAYA_MODELS, value: english}
        - {name: LAYA_PRELOAD, value: "1"}
        - {name: LAYA_THREADS, value: "4"}
        - {name: USE_TF, value: "0"}
        - {name: HF_HOME, value: /cache}
        ports: [{containerPort: 8000}]
        startupProbe: {httpGet: {path: /health, port: 8000}, periodSeconds: 15, failureThreshold: 100}
        readinessProbe: {httpGet: {path: /health, port: 8000}, periodSeconds: 10}
        resources: {requests: {cpu: "1", memory: 2Gi}, limits: {memory: 4Gi}}
        volumeMounts: [{name: cache, mountPath: /cache}]
      volumes: [{name: cache, emptyDir: {sizeLimit: 6Gi}}]
---
apiVersion: v1
kind: Service
metadata:
  name: laya
  namespace: agentgateway-system
spec:
  selector: {app: laya}
  ports: [{port: 8000}]

kubectl apply -f 20-laya.yaml

Three settings earn their place. enableServiceLinks: false, because the Service is called laya and Kubernetes would otherwise inject LAYA_PORT=tcp://10.96.x.x:8000 into the pod, which laya-serve reads as its listen port; the port is set explicitly instead. LAYA_PRELOAD loads the checkpoint before the server opens its port, so /health doubles as a readiness signal. And the startupProbe gives the install and the first download room. Start it now and let it work while you do the next two steps.

Step 3: the router

This is the whole ExtProc. Save it as main.go:

// decision-router: an agentgateway ExtProc that asks a decision model which
// model tier should answer, and writes the answer as a request header.
package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "io"
    "log"
    "net"
    "net/http"
    "os"
    "strconv"
    "time"

    corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
    extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
    "google.golang.org/grpc"
)

// The routing table: label -> description the decision model reads, and the
// model name the gateway backend pins. Edit these, restart, routing changes.
var tiers = map[string]struct{ Description, Model string }{
    "fast":   {"Short factual questions, greetings, translations, one-line definitions, rewording a sentence. Anything a small model answers well in a few lines.", "fast-model"},
    "strong": {"Multi-step reasoning, writing or debugging code, proofs, legal, financial or security analysis, long structured documents. Anything where a wrong answer is costly.", "strong-model"},
}

var (
    decisionURL = os.Getenv("DECISION_URL")             // http://laya:8000 or https://api.typesafe.ai
    decisionKey = os.Getenv("DECISION_API_KEY")         // empty = no Authorization header
    defaultTier = env("DEFAULT_TIER", "fast")
    threshold, _ = strconv.ParseFloat(env("CONFIDENCE_THRESHOLD", "0.55"), 64)
    client      = &http.Client{Timeout: 2 * time.Second}
)

func env(k, d string) string {
    if v := os.Getenv(k); v != "" {
        return v
    }
    return d
}

// decide returns the headers for one request. It never fails: anything odd
// lands on the default tier with the reason in x-decision-reason.
func decide(body []byte) map[string]string {
    var req struct {
        Model    string `json:"model"`
        Messages []struct {
            Role    string `json:"role"`
            Content string `json:"content"`
        } `json:"messages"`
    }
    if err := json.Unmarshal(body, &req); err == nil && req.Model != "" && req.Model != "auto" {
        return map[string]string{"x-selected-model": req.Model, "x-decision-reason": "passthrough"}
    }
    prompt := ""
    for _, m := range req.Messages {
        if m.Role == "user" {
            prompt = m.Content
        }
    }
    tier, conf, reason := defaultTier, 0.0, "fallback"
    if prompt != "" {
        if label, p, err := ask(prompt); err != nil {
            log.Printf("decision api: %v", err)
        } else if _, ok := tiers[label]; ok && p >= threshold {
            tier, conf, reason = label, p, "classified"
        } else {
            conf, reason = p, "low-confidence"
        }
    }
    return map[string]string{
        "x-selected-model":      tiers[tier].Model,
        "x-decision-tier":       tier,
        "x-decision-reason":     reason,
        "x-decision-confidence": fmt.Sprintf("%.3f", conf),
    }
}

// ask sends one choice question to the /v1/systemone API and returns the label
// and the probability of that label.
func ask(prompt string) (string, float64, error) {
    criteria := map[string]string{}
    for name, t := range tiers {
        criteria[name] = t.Description
    }
    payload, _ := json.Marshal(map[string]any{
        "model": "jev-latest", "state": prompt,
        "questions": map[string]any{"tier": map[string]any{
            "type": "choice", "instructions": "Which model tier should answer this user request?", "criteria": criteria}},
    })
    req, _ := http.NewRequest("POST", decisionURL+"/v1/systemone", bytes.NewReader(payload))
    req.Header.Set("Content-Type", "application/json")
    if decisionKey != "" {
        req.Header.Set("Authorization", "Bearer "+decisionKey)
    }
    resp, err := client.Do(req)
    if err != nil {
        return "", 0, err
    }
    defer resp.Body.Close()
    var out struct {
        Answers struct {
            Tier struct {
                Choice        string             `json:"choice"`
                Probabilities map[string]float64 `json:"probabilities"`
            } `json:"tier"`
        } `json:"answers"`
    }
    raw, _ := io.ReadAll(resp.Body)
    if resp.StatusCode != 200 {
        return "", 0, fmt.Errorf("%s: %s", resp.Status, raw)
    }
    if err := json.Unmarshal(raw, &out); err != nil {
        return "", 0, err
    }
    return out.Answers.Tier.Choice, out.Answers.Tier.Probabilities[out.Answers.Tier.Choice], nil
}

func mutation(h map[string]string) *extprocv3.HeaderMutation {
    m := &extprocv3.HeaderMutation{RemoveHeaders: []string{"x-selected-model", "x-decision-tier", "x-decision-reason", "x-decision-confidence"}}
    for k, v := range h {
        m.SetHeaders = append(m.SetHeaders, &corev3.HeaderValueOption{Header: &corev3.HeaderValue{Key: k, RawValue: []byte(v)}})
    }
    return m
}

type server struct{ extprocv3.UnimplementedExternalProcessorServer }

// Process handles one HTTP exchange: request headers, request body, response headers.
func (server) Process(stream extprocv3.ExternalProcessor_ProcessServer) error {
    var decision map[string]string
    for {
        req, err := stream.Recv()
        if err != nil {
            return nil
        }
        var resp *extprocv3.ProcessingResponse
        switch v := req.Request.(type) {
        case *extprocv3.ProcessingRequest_RequestHeaders:
            // Strip the router's headers so a client cannot pick its own tier.
            resp = &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_RequestHeaders{
                RequestHeaders: &extprocv3.HeadersResponse{Response: &extprocv3.CommonResponse{HeaderMutation: mutation(nil)}}}}
        case *extprocv3.ProcessingRequest_RequestBody:
            decision = decide(v.RequestBody.Body)
            log.Printf("decision %v", decision)
            resp = &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_RequestBody{
                RequestBody: &extprocv3.BodyResponse{Response: &extprocv3.CommonResponse{HeaderMutation: mutation(decision), ClearRouteCache: true}}}}
        case *extprocv3.ProcessingRequest_ResponseHeaders:
            // Echo the decision on the response so the client can see it.
            resp = &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_ResponseHeaders{
                ResponseHeaders: &extprocv3.HeadersResponse{Response: &extprocv3.CommonResponse{HeaderMutation: mutation(decision)}}}}
        default:
            resp = &extprocv3.ProcessingResponse{}
        }
        if err := stream.Send(resp); err != nil {
            return err
        }
    }
}

func main() {
    lis, err := net.Listen("tcp", ":9090")
    if err != nil {
        log.Fatal(err)
    }
    srv := grpc.NewServer()
    extprocv3.RegisterExternalProcessorServer(srv, server{})
    log.Printf("decision-router on :9090, decision API %s, threshold %.2f", decisionURL, threshold)
    log.Fatal(srv.Serve(lis))
}

Read it top to bottom once. The routing table is at the top: two labels, the description the decision model reads for each, and the model name the backend pins. The descriptions are the classifier's whole instruction set; the model was never trained on your tiers, so it needs to be told what they mean. Change them, restart the pod, and the routing changes. Swap them for code and prose, or legal and general, and nothing else in the file moves.

decide never fails the request. An explicit model other than auto is passed through untouched. A prompt goes to ask, which is the /v1/systemone call from the first section. A label at or above the threshold is classified; a label under it is low-confidence and goes to the default tier; a timeout or an error is fallback, also to the default tier. Every outcome is written to four headers so you can see it from the client.

Process is the ext_proc side. On the request headers phase the router strips every header it owns, so a client cannot pick its own tier by sending x-selected-model itself. On the request body phase it decides and returns a header mutation. On the response headers phase it echoes the same headers, which is how you read the decision without looking at logs.

The pod compiles this from a ConfigMap. 30-router.yaml:

# The router runs main.go straight from a ConfigMap: the pod fetches the two Go
# dependencies and compiles at start (about a minute). Build an image for production.
apiVersion: apps/v1
kind: Deployment
metadata:
  name: decision-router
  namespace: agentgateway-system
spec:
  selector: {matchLabels: {app: decision-router}}
  template:
    metadata: {labels: {app: decision-router}}
    spec:
      containers:
      - name: router
        image: golang:1.26-alpine
        workingDir: /src
        command: ["sh", "-c"]
        args:
        - cp /cm/main.go . && go mod init router >/dev/null 2>&1 &&
          go get github.com/envoyproxy/go-control-plane/envoy@v1.39.0 google.golang.org/grpc@v1.84.0 &&
          go mod tidy && exec go run .
        env:
        - {name: DECISION_URL, value: http://laya.agentgateway-system.svc.cluster.local:8000}
        - {name: DEFAULT_TIER, value: fast}
        - {name: CONFIDENCE_THRESHOLD, value: "0.55"}
        ports: [{name: grpc, containerPort: 9090}]
        startupProbe: {tcpSocket: {port: 9090}, periodSeconds: 10, failureThreshold: 30}
        readinessProbe: {tcpSocket: {port: 9090}, periodSeconds: 5}
        volumeMounts:
        - {name: src, mountPath: /cm}
        - {name: work, mountPath: /src}
      volumes:
      - {name: src, configMap: {name: decision-router-src}}
      - {name: work, emptyDir: {}}
---
# appProtocol h2c is what tells agentgateway to speak gRPC to this Service.
apiVersion: v1
kind: Service
metadata:
  name: decision-router
  namespace: agentgateway-system
spec:
  selector: {app: decision-router}
  ports: [{name: grpc, port: 9090, targetPort: grpc, appProtocol: kubernetes.io/h2c}]

kubectl -n agentgateway-system create configmap decision-router-src --from-file=main.go
kubectl apply -f 30-router.yaml

Two things to notice. appProtocol: kubernetes.io/h2c on the Service is what tells agentgateway to speak gRPC to it. And the Deployment does go get and go run at start, which suits a lab and not production; the last section says what to do instead.

Step 4: the policy and the route

This is the whole integration. 40-policy.yaml:

apiVersion: agentgateway.dev/v1alpha1
kind: AgentgatewayPolicy
metadata:
  name: decision-router
  namespace: agentgateway-system
spec:
  targetRefs:
  - group: gateway.networking.k8s.io
    kind: Gateway
    name: agentgateway-proxy
  traffic:
    phase: PreRouting
    extProc:
      backendRef:
        name: decision-router
        namespace: agentgateway-system
        port: 9090
      failureMode: FailOpen
      processingOptions:
        requestHeaderMode: Send
        requestBodyMode: Buffered
        responseHeaderMode: Send
        responseBodyMode: None
        requestTrailerMode: Skip
        responseTrailerMode: Skip

Three fields to read twice. phase: PreRouting is what makes the header count for routing; the CRD documents this phase as the one for policies that need to influence the routing decision. requestBodyMode: Buffered gives the router the whole prompt in one message. failureMode: FailOpen covers the router pod itself being unreachable: the request continues to the default route instead of failing. The decision API being unreachable is the router's own job, and it handles it by falling back, so a classifier outage never becomes a gateway outage. responseBodyMode: None is deliberate: the router only needs response headers to echo its decision, so the answer body never takes the detour through gRPC.

Then one HTTPRoute that matches on the header the router wrote. The last rule is the safety net: no header, or a model name nobody knows, and the request goes to the default tier. 40-route.yaml:

apiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
  name: llm-tiers
  namespace: agentgateway-system
spec:
  parentRefs:
  - name: agentgateway-proxy
    namespace: agentgateway-system
    sectionName: http
  rules:
  - matches:
    - path: {type: PathPrefix, value: /v1/chat/completions}
      headers:
      - {type: Exact, name: x-selected-model, value: fast-model}
    backendRefs:
    - {name: llm-fast, group: agentgateway.dev, kind: AgentgatewayBackend}
  - matches:
    - path: {type: PathPrefix, value: /v1/chat/completions}
      headers:
      - {type: Exact, name: x-selected-model, value: strong-model}
    backendRefs:
    - {name: llm-strong, group: agentgateway.dev, kind: AgentgatewayBackend}
  - matches:
    - path: {type: PathPrefix, value: /}
    backendRefs:
    - {name: llm-fast, group: agentgateway.dev, kind: AgentgatewayBackend}

kubectl apply -f 40-policy.yaml -f 40-route.yaml
kubectl -n agentgateway-system rollout status deploy/decision-router deploy/laya --timeout=1500s
kubectl -n agentgateway-system get agentgatewaypolicy decision-router

NAME              ACCEPTED   ATTACHED   AGE
decision-router   True       True       2m

Step 5: run it

Open a port-forward and ask the gateway three things.

kubectl -n agentgateway-system port-forward svc/agentgateway-proxy 8080:80 &

An explicit model bypasses the classifier:

curl -s -D - -o /dev/null localhost:8080/v1/chat/completions -H 'content-type: application/json' \
  -d '{"model":"fast-model","messages":[{"role":"user","content":"Say hi."}]}' | grep -i x-

x-decision-reason: passthrough
x-selected-model: fast-model

auto with a simple prompt:

curl -s -D - localhost:8080/v1/chat/completions -H 'content-type: application/json' \
  -d '{"model":"auto","messages":[{"role":"user","content":"What is the capital of France?"}]}'

x-decision-reason: classified
x-decision-confidence: 0.673
x-selected-model: fast-model
x-decision-tier: fast

{"id":"mock","object":"chat.completion","model":"fast-model","choices":[{"index":0,"message":{"role":"assistant","content":"[fast-model] answered"},"finish_reason":"stop"}],...}

And auto with the rate limiter prompt from the top of the article:

x-decision-tier: strong
x-decision-reason: classified
x-decision-confidence: 0.741
x-selected-model: strong-model

{"id":"mock","object":"chat.completion","model":"strong-model","choices":[{"index":0,"message":{"role":"assistant","content":"[strong-model] answered"},...}

The body is the point. The client sent auto and the model server received strong-model. That rewrite is the backend's model field at work.

One negative check, because it is the one I care about most. A client that tries to pick its own tier with a header gets the classifier's answer anyway, because the router strips its headers before deciding:

curl -s -D - -o /dev/null localhost:8080/v1/chat/completions -H 'content-type: application/json' \
  -H 'x-selected-model: strong-model' \
  -d '{"model":"auto","messages":[{"role":"user","content":"What is the capital of France?"}]}' | grep -i x-decision-tier

x-decision-tier: fast

Now sixteen labelled prompts through the classifier, with the tier it chose and the probability behind it:

while IFS='|' read -r expect prompt; do
  out=$(curl -s -D - -o /dev/null localhost:8080/v1/chat/completions -H 'content-type: application/json' \
    -d "$(jq -cn --arg p "$prompt" '{model:"auto",messages:[{role:"user",content:$p}]}')")
  got=$(grep -i '^x-decision-tier:' <<<"$out" | cut -d' ' -f2 | tr -d '\r')
  conf=$(grep -i '^x-decision-confidence:' <<<"$out" | cut -d' ' -f2 | tr -d '\r')
  [[ "$got" == "$expect" ]] && mark=' ' || mark='!'
  printf '%s %-7s %-7s %-6s %s\n' "$mark" "$expect" "$got" "$conf" "${prompt:0:60}"
done <<'PROMPTS'
fast|What is the capital of Portugal?
fast|Translate 'good morning' into German.
fast|Give me a one-line definition of a Kubernetes Service.
fast|Rewrite this sentence in a friendlier tone: Your ticket has been closed.
fast|How many minutes are there in three hours?
fast|Summarize in one sentence: Kind runs Kubernetes nodes as Docker containers.
fast|What does the acronym CRD stand for in Kubernetes?
fast|Say hello in Spanish.
strong|Design a multi-region rate limiter for an API that must stay consistent under partition, and explain the trade-offs between token bucket and sliding window, then write it in Go with unit tests.
strong|Here is a Go stack trace with a nil pointer dereference in a gRPC stream handler. Walk through the likely root causes step by step and propose a fix with a regression test.
strong|Compare the obligations of a provider versus a deployer of a high-risk AI system under the EU AI Act, with article references, and draft a compliance checklist for a bank.
strong|Write a 1500-word architecture document for migrating a monolith to event-driven microservices, covering data ownership, saga patterns, idempotency and observability.
strong|Prove that the sum of the first n odd numbers equals n squared, then generalize the argument to arithmetic progressions.
strong|Refactor this 400-line Python module into testable units, explain each SOLID violation you found, and produce the refactored code with type hints.
strong|Analyze the security implications of running an ext_proc filter before route selection, enumerate threat scenarios, and recommend mitigations with configuration examples.
strong|Given quarterly revenue, churn and CAC figures for three SaaS products, build a cohort model, identify which product to sunset and justify the decision quantitatively.
PROMPTS

This is the run behind this article, on a MacBook with Laya on four CPU threads, each line 200 to 245 ms end to end through the gateway:

  fast    fast    0.678  What is the capital of Portugal?
  fast    fast    0.542  Translate 'good morning' into German.
  fast    fast    0.808  Give me a one-line definition of a Kubernetes Service.
  fast    fast    0.726  Rewrite this sentence in a friendlier tone: Your ticket has
  fast    fast    0.824  How many minutes are there in three hours?
  fast    fast    0.757  Summarize in one sentence: Kind runs Kubernetes nodes as Doc
  fast    fast    0.729  What does the acronym CRD stand for in Kubernetes?
  fast    fast    0.616  Say hello in Spanish.
  strong  strong  0.742  Design a multi-region rate limiter for an API that must stay
  strong  strong  0.714  Here is a Go stack trace with a nil pointer dereference in a
! strong  fast    0.534  Compare the obligations of a provider versus a deployer of a
  strong  strong  0.626  Write a 1500-word architecture document for migrating a mono
  strong  strong  0.604  Prove that the sum of the first n odd numbers equals n squar
  strong  strong  0.557  Refactor this 400-line Python module into testable units, ex
  strong  strong  0.646  Analyze the security implications of running an ext_proc fil
  strong  strong  0.590  Given quarterly revenue, churn and CAC figures for three SaaS

Fifteen of sixteen. Read the marked line with the threshold in mind: Laya leaned strong at 0.534, under the 0.55 line, so the router sent it to the cheap tier and labelled the decision low-confidence. That is the threshold doing its job. When the classifier is not sure, the request goes to the tier you chose as the default, and the header tells you it happened. Lower the threshold and it moves; raise it and more do. The number is yours.

Switching to Jev

Same router, same policy, same route. Set DECISION_URL to https://api.typesafe.ai and give the pod DECISION_API_KEY from a Secret:

kubectl -n agentgateway-system create secret generic typesafe --from-literal=KEY="$TYPESAFE_API_KEY"
kubectl -n agentgateway-system set env deploy/decision-router \
  DECISION_URL=https://api.typesafe.ai --from=secret/typesafe --prefix=DECISION_API_

That last command maps the Secret's KEY to DECISION_API_KEY. The request body from the first section is the one the router sends; model: jev-latest resolves to the current version. Jev's probabilities are the same field the router already reads, and Jev adds a calibrated confidence you could compare instead.

What changes is not the code. It is where the last user message of every auto request goes: to TypeSafe's service, hosted in the United States, under their data processing addendum, with zero data retention as an enterprise option. If that fits your data, the hosted model gives you TypeSafe's calibration and no pod to feed. If it does not, Laya is already running.

Before this goes near a customer

The lab is deliberately the smallest thing that shows the mechanism. What I would add, roughly in order:

Build an image for the router instead of go run in the pod, and move the tier table out of the source into a ConfigMap, so the platform team changes descriptions without touching Go.

Give Laya a PersistentVolumeClaim so a rescheduled pod is not a re-install and a re-download, and pin the version you tested. The emptyDir is fine on a laptop and nowhere else.

Tell a timeout from an error from an unknown label in the headers, and add the decision latency and the model id the API reported. The lab folds them into fallback because the tier is the same; a dashboard wants the difference.

Write your own eval set. Sixteen prompts show the mechanism; your traffic decides the descriptions and the threshold. The loop above is the whole tool.

Start in observe mode. The router already explains every decision in headers, so run it for a week with the HTTPRoute pointing everything at one tier, read x-decision-tier in your logs, and only then split the route.

Ask a second question in the same call. Questions to a decision model are evaluated in parallel and isolated from each other, so a noul asking "does this message contain personal data?" costs nothing extra and gives you a second header to route or block on.

Put the tiers behind failover. AgentgatewayBackend supports priority groups, so strong can be two providers in order, and the router does not need to know.

Next level

Three directions I would take it. More than two tiers: one more entry in the table, one more rule in the HTTPRoute, one more backend. A score question for urgency, routed to a header your rate limit policy can read. And the same router in front of vLLM on a GPU node, where the tiers are two real models and the stand-in answers become real ones.

If you build this and land on a different threshold, a better pair of descriptions, or a reason to put the classifier somewhere else, I want to hear it. That is usually the conversation where I learn something.

agentgateway is an open source project under the Agentic AI Foundation: docs at agentgateway.dev, source at github.com/agentgateway/agentgateway. Jev is documented at docs.typesafe.ai. Laya is at github.com/NandhaKishorM/laya.

I'm with Uncle Bob. But only if nobody can skip the checks

Antonio Berben — Mon, 21 Sep 2026 14:08:41 +0000

A few weeks ago I watched Uncle Bob do an interview in a bathrobe.

The Clean Code guy, half a century of programming behind him, a Diet Coke in his hand. I expected a complaint about kids these days and their robots. Instead he described an assembly line: one agent cleans, hands off to a second that hardens the tests, then to a third that defends the architecture. Each one short-lived, each one holding a tool it is not allowed to disagree with.

Then the line that made me pause the video:

Deterministic tools don't disappear that way.

The "that way" is the part everybody in my inbox is currently getting wrong.

The thing he says does not work

The agent does something silly, so you explain yourself better. Another paragraph in AGENTS.md, another "never do this again". The file grows, and around sentence eighty its middle stops existing: the tokens are in the context, the behaviour is gone.

Same shape as the runbooks nobody followed at 2am. We did not fix those by writing a better wiki page. We fixed them by putting the rule in the pipeline.

His pipeline is three tools he has been collecting since the early 2000s:

The CRAP score, an acronym he is not apologising for, which mixes test coverage with the cyclomatic complexity of each function.
Mutation testing. Flip an operator in the source, run the suite, and if the tests still pass you have a surviving mutant. In his words, it must be killed.
A specification file that declares which module may import which, plus a checker at the end that the agents are not allowed to edit.

And a loop around them: change the code until the tool says it is okay. The agent does not get to declare victory. The tool does.

That is the whole trick. The instruction is an opinion that competes with thirty turns of the agent's own output. The tool returns a number.

What I would not ship to a customer

On one laptop this is beautiful. But I spend my week with platform teams about to hand the pattern to forty engineers, and at that scale the question stops being which tools and becomes who gets to call which one.

The mutation checker runs your test suite. The dependency checker reads the whole repo. I do not want every agent in the company reaching those because it guessed the endpoint, and "it has been told not to" is, as we just established, a wish.

So the straitjacket has to be buckled somewhere the agent does not control. That is a gateway problem, which happens to be what I do for a living, so I spent an evening rebuilding his gauntlet behind agentgateway: an AAIF project that speaks MCP and can filter what a caller is allowed to see.

Three stages, three role tokens, one endpoint. The cleaner measures complexity and nothing else. The hardener runs mutation testing and cannot even see the complexity tool. The architect checks the dependency graph. Same URL for all three.

The rest is the lab. One laptop, no Kubernetes, about twenty minutes.

A note on the tools I picked

The harness is goose, and I will be upfront: Claude Code and Codex are probably better at the raw coding. They are also licensed products from one vendor, and the point here is that the constraints outlive whichever tool you like this quarter. Every piece below is an Agentic AI Foundation project: AGENTS.md, MCP, goose, agentgateway. Swap the model and the vendor, the gauntlet still stands.

What you need

Python 3 and openssl. No pip install: the checkers are ~80 lines of standard library each.
The agentgateway binary, v1.5.0.
goose v1.51.0, plus a configured provider: run goose configure once, or export GOOSE_PROVIDER, GOOSE_MODEL and your API key.

Everything below lives in one directory, and every file in this article is in it:

gauntlet/
├── agentgateway          # the binary
├── config.yaml           # step 2
├── jwt/mint.py           # step 3
├── verify.py             # step 4
├── recipes/
│   ├── cleaner.yaml      # step 5
│   ├── hardener.yaml     # step 6
│   └── architect.yaml    # step 6
├── servers/
│   ├── mcplite.py        # minimal MCP server, shared
│   ├── quality.py        # complexity
│   ├── mutation.py       # mutation testing
│   ├── deps.py           # architecture rules
│   └── oneshot.py        # same checkers, called from a shell
└── sample-project/       # the code the agents work on

Step 0: something for the agents to break

The gauntlet needs a victim: a tiny shopping cart with a declared architecture and a test suite that looks reasonable and is not.

sample-project/dependency-rules.txt

# Declared architecture. The agents may not edit this file.
# module -> modules it is allowed to import
cart    -> pricing
pricing -> nothing
storage -> nothing

That file is the contract. cart may import pricing, and nothing else may import anything. Right now the code does not obey it, which is the architect stage's problem later.

the four files of sample-project

sample-project/cart.py

import pricing


def cart_total(lines, customer_tier, coupon, is_weekend):
    total = 0
    for item, quantity in lines:
        total += pricing.price_for(item, quantity, customer_tier, coupon, is_weekend)
    return round(total, 2)

sample-project/pricing.py

import storage  # violates the declared rules on purpose


def _apply_customer_tier(total, customer_tier, quantity):
    if customer_tier == "gold":
        return total * 0.8
    if customer_tier == "silver":
        return total * 0.9
    if customer_tier == "bronze" and quantity > 10:
        return total * 0.95
    return total


def _apply_coupon(total, customer_tier, coupon, quantity):
    if coupon == "WELCOME" and customer_tier != "gold":
        return total - 5
    if coupon == "BULK" and quantity >= 20:
        return total - 15
    return total


def price_for(item, quantity, customer_tier, coupon, is_weekend):
    """One function doing the work of four. Complexity is the point."""
    base = storage.unit_price(item)
    if base is None:
        return 0
    total = base * quantity
    total = _apply_customer_tier(total, customer_tier, quantity)
    total = _apply_coupon(total, customer_tier, coupon, quantity)
    if is_weekend and total > 100:
        total = total * 0.97
    if total < 0:
        total = 0
    return round(total, 2)

sample-project/storage.py

CATALOG = {"widget": 10.0, "gizmo": 25.0, "doohickey": 4.5}


def unit_price(item):
    return CATALOG.get(item)

sample-project/test_cart.py

import unittest

import cart


class TestCart(unittest.TestCase):
    def test_single_line_no_discount(self):
        self.assertEqual(cart.cart_total([("widget", 1)], "none", None, False), 10.0)

    def test_two_lines(self):
        self.assertEqual(cart.cart_total([("widget", 1), ("gizmo", 1)], "none", None, False), 35.0)


if __name__ == "__main__":
    unittest.main()

sample-project/AGENTS.md

# Shopping cart service

A pricing and cart module for an online shop.

Run the tests with `python3 -m unittest discover -q` from this directory.

Architecture is declared in `dependency-rules.txt`. That file is the contract,
not a suggestion: change the code to fit it, never the other way round.

One warning about this copy. It is the state of my project after the cleaner already ran, so complexity is under the limit at 6 and that stage will pass immediately. Run the cleaner with a limit of 4 if you want to watch it work. The other two stages have plenty to do, as you will see at the end.

Step 1: write checkers that cannot be talked to

Three servers, one shared transport. The first reports cyclomatic complexity per function by walking the AST: the half of the CRAP score that needs no test run and answers in milliseconds.

servers/quality.py

#!/usr/bin/env python3
"""Quality checker MCP server: cyclomatic complexity per function.

Uncle Bob gates his agents on a CRAP score. CRAP mixes coverage and
cyclomatic complexity; this server ships the complexity half, which is the
part that needs no test run and is deterministic in milliseconds.
"""

import ast
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mcplite import serve  # noqa: E402

BRANCHING = (ast.If, ast.For, ast.AsyncFor, ast.While, ast.ExceptHandler,
             ast.With, ast.AsyncWith, ast.Assert, ast.IfExp, ast.comprehension)


def complexity(node):
    score = 1
    for child in ast.walk(node):
        if isinstance(child, BRANCHING):
            score += 1
        elif isinstance(child, ast.BoolOp):
            score += len(child.values) - 1
        elif isinstance(child, ast.Match):
            score += len(child.cases)
    return score


def scan(root, limit):
    rows = []
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if not d.startswith((".", "__"))]
        for name in sorted(filenames):
            if not name.endswith(".py"):
                continue
            path = os.path.join(dirpath, name)
            try:
                tree = ast.parse(open(path, encoding="utf-8").read())
            except SyntaxError as exc:
                rows.append((path, "<syntax error>", -1, str(exc)))
                continue
            for node in ast.walk(tree):
                if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    rows.append((os.path.relpath(path, root), node.name,
                                 complexity(node), None))
    return [r for r in rows if r[2] > limit or r[2] < 0], rows


def complexity_report(args):
    root = args.get("path", ".")
    limit = int(args.get("limit", 6))
    if not os.path.isdir(root):
        return f"FAIL: {root} is not a directory"
    offenders, every = scan(root, limit)
    if not every:
        return f"FAIL: no Python files found under {root}"
    if not offenders:
        worst = max(every, key=lambda r: r[2])
        return (f"PASS: {len(every)} functions, all at or below limit {limit}. "
                f"Worst is {worst[1]} at {worst[2]}.")
    lines = [f"FAIL: {len(offenders)} function(s) above limit {limit}"]
    for path, func, score, note in sorted(offenders, key=lambda r: -r[2]):
        lines.append(f"  {path}:{func} complexity={score}" + (f" {note}" if note else ""))
    return "\n".join(lines)


serve("quality", [{
    "name": "complexity_report",
    "description": ("Report cyclomatic complexity of every Python function under a path. "
                    "Returns PASS only when every function is at or below the limit."),
    "inputSchema": {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "Directory to scan"},
            "limit": {"type": "integer", "description": "Maximum allowed complexity"},
        },
        "required": ["path"],
    },
    "handler": complexity_report,
}])

serve() lives in a 70 line module with no dependencies. Nothing here is clever, and that is the point: a checker you cannot argue with should be one you can read in a single sitting.

servers/mcplite.py, the minimal MCP stdio server the three checkers share

"""Minimal MCP stdio server. Standard library only, no dependencies.

Speaks just enough of the protocol for agentgateway to connect, list tools
and call them: initialize, notifications/initialized, tools/list, tools/call.
"""

import json
import sys


def _send(msg):
    sys.stdout.write(json.dumps(msg) + "\n")
    sys.stdout.flush()


def _result(req_id, payload):
    _send({"jsonrpc": "2.0", "id": req_id, "result": payload})


def _error(req_id, code, message):
    _send({"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}})


def serve(server_name, tools):
    """tools: list of dicts with name, description, inputSchema, handler."""
    by_name = {t["name"]: t for t in tools}
    listing = [
        {"name": t["name"], "description": t["description"], "inputSchema": t["inputSchema"]}
        for t in tools
    ]

    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
        except json.JSONDecodeError:
            continue

        method = req.get("method")
        req_id = req.get("id")

        if method == "initialize":
            _result(req_id, {
                "protocolVersion": req.get("params", {}).get("protocolVersion", "2025-06-18"),
                "capabilities": {"tools": {}},
                "serverInfo": {"name": server_name, "version": "0.1.0"},
            })
        elif method in ("notifications/initialized", "notifications/cancelled"):
            continue  # notifications carry no id and expect no reply
        elif method == "ping":
            _result(req_id, {})
        elif method == "tools/list":
            _result(req_id, {"tools": listing})
        elif method == "tools/call":
            params = req.get("params", {})
            tool = by_name.get(params.get("name"))
            if tool is None:
                _error(req_id, -32602, f"unknown tool: {params.get('name')}")
                continue
            try:
                text = tool["handler"](params.get("arguments") or {})
                _result(req_id, {"content": [{"type": "text", "text": text}]})
            except Exception as exc:  # surface failures as tool errors, not crashes
                _result(req_id, {
                    "content": [{"type": "text", "text": f"{type(exc).__name__}: {exc}"}],
                    "isError": True,
                })
        elif req_id is not None:
            _error(req_id, -32601, f"method not found: {method}")

The second flips arithmetic and comparison operators, runs the suite against each mutant in a temporary copy and reports the survivors. The third checks every local import against dependency-rules.txt.

servers/mutation.py, the mutation tester

#!/usr/bin/env python3
"""Mutation testing MCP server.

Bob's hardener. Flip an operator in the source, run the test suite, and
expect it to fail. A mutant that survives is a hole in the tests.
Standard library only: ast for the mutation, unittest for the run.
"""

import ast
import os
import shutil
import subprocess
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mcplite import serve  # noqa: E402

FLIPS = {
    ast.Add: ast.Sub, ast.Sub: ast.Add, ast.Mult: ast.FloorDiv,
    ast.Lt: ast.LtE, ast.LtE: ast.Lt, ast.Gt: ast.GtE, ast.GtE: ast.Gt,
    ast.Eq: ast.NotEq, ast.NotEq: ast.Eq,
}


class Mutator(ast.NodeTransformer):
    """Replaces the nth mutable operator in the tree, counting from zero."""

    def __init__(self, target):
        self.target = target
        self.seen = 0
        self.applied = None

    def _maybe(self, op):
        replacement = FLIPS.get(type(op))
        if replacement is None:
            return op
        index, self.seen = self.seen, self.seen + 1
        if index != self.target:
            return op
        self.applied = f"{type(op).__name__} -> {replacement.__name__}"
        return replacement()

    def visit_BinOp(self, node):
        self.generic_visit(node)
        node.op = self._maybe(node.op)
        return node

    def visit_Compare(self, node):
        self.generic_visit(node)
        node.ops = [self._maybe(op) for op in node.ops]
        return node


def count_targets(source):
    probe = Mutator(-1)
    probe.visit(ast.parse(source))
    return probe.seen


def tests_pass(workdir):
    proc = subprocess.run([sys.executable, "-m", "unittest", "discover", "-q"],
                          cwd=workdir, capture_output=True, text=True, timeout=120)
    return proc.returncode == 0


def mutation_run(args):
    project = os.path.abspath(args.get("path", "."))
    module = args.get("module")
    if not module:
        return "FAIL: 'module' is required, e.g. cart.py"
    source_path = os.path.join(project, module)
    if not os.path.isfile(source_path):
        return f"FAIL: {module} not found under {project}"

    source = open(source_path, encoding="utf-8").read()
    total = count_targets(source)
    if total == 0:
        return f"FAIL: no mutable operators found in {module}"

    with tempfile.TemporaryDirectory() as tmp:
        work = os.path.join(tmp, "work")
        shutil.copytree(project, work, ignore=shutil.ignore_patterns("__pycache__", ".git"))
        if not tests_pass(work):
            return "FAIL: the test suite does not pass before mutating. Fix the tests first."

        survivors = []
        for index in range(total):
            mutator = Mutator(index)
            mutated = ast.fix_missing_locations(mutator.visit(ast.parse(source)))
            if mutator.applied is None:
                continue
            with open(os.path.join(work, module), "w", encoding="utf-8") as handle:
                handle.write(ast.unparse(mutated))
            if tests_pass(work):
                survivors.append(f"#{index} {mutator.applied}")
        with open(os.path.join(work, module), "w", encoding="utf-8") as handle:
            handle.write(source)

    if not survivors:
        return f"PASS: {total} mutants generated in {module}, all killed by the tests."
    lines = [f"FAIL: {len(survivors)} of {total} mutants survived in {module}"]
    lines += [f"  {s}" for s in survivors]
    lines.append("Every surviving mutant is a behaviour your tests do not check.")
    return "\n".join(lines)


serve("mutation", [{
    "name": "mutation_run",
    "description": ("Mutate arithmetic and comparison operators in a module, run the "
                    "unittest suite against each mutant, and report the survivors."),
    "inputSchema": {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "Project root containing the tests"},
            "module": {"type": "string", "description": "File to mutate, relative to path"},
        },
        "required": ["path", "module"],
    },
    "handler": mutation_run,
}])

servers/deps.py, the architecture checker

#!/usr/bin/env python3
"""Dependency rule checker MCP server.

Bob keeps a file declaring which module may depend on which, and a checker
that the agents cannot argue with. This is that checker: it reads the
declared rules, walks the imports, and reports every edge nobody allowed.
"""

import ast
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mcplite import serve  # noqa: E402

RULES_FILE = "dependency-rules.txt"


def load_rules(project):
    """Each line is 'module -> allowed, allowed'. Lines starting with # are notes."""
    path = os.path.join(project, RULES_FILE)
    if not os.path.isfile(path):
        return None
    rules = {}
    for raw in open(path, encoding="utf-8"):
        line = raw.split("#", 1)[0].strip()
        if not line or "->" not in line:
            continue
        left, right = line.split("->", 1)
        allowed = {item.strip() for item in right.split(",") if item.strip()}
        rules[left.strip()] = allowed - {"nothing"}
    return rules


def local_imports(path, known):
    found = set()
    try:
        tree = ast.parse(open(path, encoding="utf-8").read())
    except SyntaxError:
        return found
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                head = alias.name.split(".")[0]
                if head in known:
                    found.add(head)
        elif isinstance(node, ast.ImportFrom) and node.module:
            head = node.module.split(".")[0]
            if head in known:
                found.add(head)
    return found


def deps_check(args):
    project = os.path.abspath(args.get("path", "."))
    rules = load_rules(project)
    if rules is None:
        return f"FAIL: no {RULES_FILE} in {project}. Declare the allowed edges first."

    known = set(rules)
    violations = []
    for module, allowed in sorted(rules.items()):
        path = os.path.join(project, f"{module}.py")
        if not os.path.isfile(path):
            violations.append(f"  {module}: declared in rules but {module}.py is missing")
            continue
        for target in sorted(local_imports(path, known) - {module}):
            if target not in allowed:
                violations.append(f"  {module} -> {target} is not allowed")

    if not violations:
        return f"PASS: {len(rules)} modules checked, every import matches {RULES_FILE}."
    return "\n".join([f"FAIL: {len(violations)} dependency violation(s)"] + violations +
                     ["Invert the dependency or split the module. Do not edit the rules file."])


serve("deps", [{
    "name": "deps_check",
    "description": ("Check every local import against the edges declared in "
                    "dependency-rules.txt. Returns PASS only when the graph obeys the rules."),
    "inputSchema": {
        "type": "object",
        "properties": {"path": {"type": "string", "description": "Project root"}},
        "required": ["path"],
    },
    "handler": deps_check,
}])

None of them asks a model anything. Same input, same verdict, every run. That is why a 12ms tool call beats a paragraph of prose.

Step 2: one endpoint in front of the three

My first config was wrong in a way worth showing, because I have seen it in three customer designs:

policies:
  mcpAuthorization:
    rules:
      - 'mcp.tool.target == "quality"'

Read it out loud: "the quality tool may be called". By whom? By anyone who can open a socket. That is an allow-list of tools mistaken for access control, which is the security posture of a door with a sign on it.

The rule has to be about the caller. Three lines, one per stage:

mcp:
  port: 3000
  policies:
    jwtAuth:
      issuer: gauntlet.local
      audiences: [agentgateway.local]
      jwks:
        file: ./jwt/jwks.json
    mcpAuthorization:
      rules:
        - '"cleaner" in jwt.roles && mcp.tool.target == "quality"'
        - '"hardener" in jwt.roles && mcp.tool.target == "mutation"'
        - '"architect" in jwt.roles && mcp.tool.target == "deps"'
  targets:
    - name: quality
      stdio:
        cmd: python3
        args: ["-u", "servers/quality.py"]
    - name: mutation
      stdio:
        cmd: python3
        args: ["-u", "servers/mutation.py"]
    - name: deps
      stdio:
        cmd: python3
        args: ["-u", "servers/deps.py"]

The gateway spawns the three checkers over stdio and multiplexes them behind one streamable HTTP endpoint. The agent sees a single URL and never learns there are three processes back there. What it sees is decided elsewhere.

Start it:

One ordering detail that cost me a 401 the first time: the gateway reads the JWKS file at
startup, so the keys have to exist before it boots. Step 3 explains the script, but you run
it now:

python3 jwt/mint.py bootstrap cleaner > /dev/null   # creates jwt/priv.pem and jwt/jwks.json
./agentgateway -f config.yaml

Step 3: the role is a signed claim, not a string in the prompt

Those rules lean on jwt.roles, so something has to mint tokens. In production that is your IdP. Here it is a 90 line script: EC P-256 keypair, JWKS, ES256 tokens signed with openssl.

python3 jwt/mint.py cleaner-bot cleaner 3600

The private key never leaves the directory. The gateway only ever reads the public half.

jwt/mint.py, the stand-in identity provider

#!/usr/bin/env python3
"""Mint ES256 JWTs for the gauntlet roles. openssl plus the standard library.

First run generates an EC P-256 keypair and the JWKS that agentgateway reads.
After that it just signs: mint.py <subject> <role> [ttl_seconds]

The key never leaves this directory, and the gateway only ever sees the public
half. This is a lab issuer: in a real setup it is your identity provider.
"""

import base64
import hashlib
import json
import os
import re
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
PRIV = os.path.join(HERE, "priv.pem")
JWKS = os.path.join(HERE, "jwks.json")
ISSUER = "gauntlet.local"
AUDIENCE = "agentgateway.local"


def b64u(raw):
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()


def run(cmd, stdin=None):
    return subprocess.run(cmd, input=stdin, capture_output=True, check=True).stdout


def public_point():
    """Return the raw X and Y coordinates of the public key."""
    text = run(["openssl", "ec", "-in", PRIV, "-pubout", "-text", "-noout"]).decode()
    block = text.split("pub:", 1)[1].split("ASN1 OID", 1)[0]
    raw = bytes.fromhex("".join(re.findall(r"[0-9a-f]{2}", block)))
    if raw[0] != 0x04 or len(raw) != 65:
        raise SystemExit("unexpected public key encoding")
    return raw[1:33], raw[33:65]


def ensure_keys():
    if os.path.isfile(PRIV) and os.path.isfile(JWKS):
        return json.load(open(JWKS))["keys"][0]["kid"]
    subprocess.run(["openssl", "ecparam", "-name", "prime256v1", "-genkey",
                    "-noout", "-out", PRIV], check=True, capture_output=True)
    os.chmod(PRIV, 0o600)
    x, y = public_point()
    kid = b64u(hashlib.sha256(x + y).digest())[:16]
    with open(JWKS, "w") as handle:
        json.dump({"keys": [{"kty": "EC", "crv": "P-256", "alg": "ES256",
                             "use": "sig", "kid": kid, "x": b64u(x), "y": b64u(y)}]},
                  handle, indent=2)
    print(f"generated {PRIV} and {JWKS}", file=sys.stderr)
    return kid


def der_to_raw(der):
    """ECDSA signatures come out of openssl as DER. JWT wants r||s, 32 bytes each."""
    if der[0] != 0x30:
        raise SystemExit("not a DER sequence")
    body = der[2:] if der[1] < 0x80 else der[2 + (der[1] & 0x7F):]
    out = b""
    for _ in range(2):
        if body[0] != 0x02:
            raise SystemExit("not a DER integer")
        length = body[1]
        value = body[2:2 + length].lstrip(b"\x00")
        out += value.rjust(32, b"\x00")
        body = body[2 + length:]
    return out


def mint(subject, role, ttl):
    kid = ensure_keys()
    now = int(time.time())
    header = {"alg": "ES256", "typ": "JWT", "kid": kid}
    payload = {"iss": ISSUER, "aud": AUDIENCE, "sub": subject, "roles": [role],
               "iat": now, "exp": now + ttl}
    signing_input = f"{b64u(json.dumps(header).encode())}.{b64u(json.dumps(payload).encode())}"
    der = run(["openssl", "dgst", "-sha256", "-sign", PRIV], stdin=signing_input.encode())
    return f"{signing_input}.{b64u(der_to_raw(der))}"


if __name__ == "__main__":
    if len(sys.argv) < 3:
        raise SystemExit("usage: mint.py <subject> <role> [ttl_seconds]")
    print(mint(sys.argv[1], sys.argv[2], int(sys.argv[3]) if len(sys.argv) > 3 else 3600))

In the prompt, the role is a sentence the agent can drift away from. In the token it is a signature, and whoever evaluates it is not the thing being constrained.

Step 4: ask the same question with three identities

Now the fun part. verify.py opens an MCP session for each role and calls tools/list:

role         tools visible through the gateway
------------------------------------------------------------
cleaner      quality_complexity_report
hardener     mutation_mutation_run
architect    deps_deps_check

no token at all:
  HTTP 401

RESULT: PASS

verify.py, the whole verifier

#!/usr/bin/env python3
"""Ask the same gateway the same question with three different identities.

Each role initializes an MCP session and calls tools/list. What comes back is
the whole point of the lab: the tool an identity may not use is not denied,
it is absent.
"""

import json
import subprocess
import sys
import urllib.error
import urllib.request

GATEWAY = "http://localhost:3000/mcp"
ROLES = [("cleaner", "cleaner"), ("hardener", "hardener"), ("architect", "architect")]
MINT = ["python3", "jwt/mint.py"]


def rpc(token, body, session=None):
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
        "Authorization": f"Bearer {token}",
    }
    if session:
        headers["Mcp-Session-Id"] = session
    req = urllib.request.Request(GATEWAY, data=json.dumps(body).encode(),
                                 headers=headers, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return resp.status, resp.headers.get("Mcp-Session-Id"), resp.read().decode()
    except urllib.error.HTTPError as exc:
        return exc.code, None, exc.read().decode()


def payload(raw):
    """Responses arrive as SSE frames or plain JSON depending on the method."""
    for line in raw.splitlines():
        if line.startswith("data: "):
            return json.loads(line[6:])
    return json.loads(raw) if raw.strip() else {}


def session_for(token):
    status, session, raw = rpc(token, {
        "jsonrpc": "2.0", "id": 1, "method": "initialize",
        "params": {"protocolVersion": "2025-06-18", "capabilities": {},
                   "clientInfo": {"name": "verify", "version": "0"}},
    })
    if status != 200:
        raise RuntimeError(f"initialize returned {status}: {raw[:200]}")
    return session


def tools_for(token):
    session = session_for(token)
    rpc(token, {"jsonrpc": "2.0", "method": "notifications/initialized"}, session)
    status, _, raw = rpc(token, {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, session)
    if status != 200:
        raise RuntimeError(f"tools/list returned {status}: {raw[:200]}")
    return [t["name"] for t in payload(raw).get("result", {}).get("tools", [])]


def main():
    print(f"{'role':<12} {'tools visible through the gateway'}")
    print("-" * 60)
    seen = {}
    for subject, role in ROLES:
        token = subprocess.run(MINT + [subject, role], capture_output=True,
                               text=True, check=True).stdout.strip()
        tools = tools_for(token)
        seen[role] = tools
        print(f"{role:<12} {', '.join(tools) if tools else '(none)'}")

    print("\nno token at all:")
    status, _, _ = rpc("not-a-token", {"jsonrpc": "2.0", "id": 1, "method": "initialize",
                                       "params": {"protocolVersion": "2025-06-18",
                                                  "capabilities": {}}})
    print(f"  HTTP {status}")

    expected = {"cleaner": ["quality_complexity_report"],
                "hardener": ["mutation_mutation_run"],
                "architect": ["deps_deps_check"]}
    ok = all(sorted(seen.get(r, [])) == sorted(v) for r, v in expected.items()) and status == 401
    print("\nRESULT:", "PASS" if ok else "FAIL")
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())

Look at what is not happening. The hardener is not denied the complexity tool: it is not in its listing at all. No forbidden fruit, no error to interpret, no retry loop against a wall. As far as that agent knows, the tool does not exist.

And without a valid token you do not get a filtered list, you get a door:

http.status=401 reason=JwtAuth
error="authentication failure: the token header is malformed"

No checker is asked anything. The mutation runner, which happily executes your test suite, never hears about the request.

Step 5: give a stage to goose

A tool nobody calls is a museum piece. The cleaner recipe:

version: "1.0.0"
title: Cleaner
description: Brings every function under the complexity limit without changing behaviour.

parameters:
  - key: project
    input_type: string
    requirement: required
    description: Path to the project being worked on
  - key: token
    input_type: string
    requirement: required
    description: Bearer token identifying this stage to the gateway

extensions:
  - type: streamable_http
    name: gauntlet
    uri: http://localhost:3000/mcp
    headers:
      Authorization: "Bearer {{ token }}"
  - type: builtin
    name: developer

instructions: |
  You reduce cyclomatic complexity in {{ project }}. Behaviour must not change:
  the existing tests have to keep passing, and you may not edit them.

  Call the complexity tool, fix what it reports, call it again. Stop when it
  says PASS. Do not argue with the number it gives you.

prompt: |
  Run the complexity check on {{ project }} with a limit of 6 and bring every
  function below it.

retry:
  max_retries: 4
  checks:
    - type: shell
      command: "python3 servers/oneshot.py quality complexity_report '{\"path\":\"{{ project }}\",\"limit\":6}'"

Two details do the heavy lifting.

The MCP extension is scoped by the gateway; goose's builtin developer is not. The agent writes files with its own tools, which is fine: we govern which checkers it reaches, not whether it can save a file.

And the checks block. oneshot.py imports the same handler the MCP server uses and exits non-zero unless it returns PASS. One implementation, two front doors. The stage ends when the shell says so, not when the agent says so, and if it does not, goose retries with the failure in hand.

servers/oneshot.py, the shell side of the same checker

#!/usr/bin/env python3
"""Run one checker directly and exit non-zero unless it returns PASS.

goose retry checks are shell commands, not MCP calls, so the same checker has
to be reachable both ways. This imports the handler out of the MCP server
module, so there is one implementation behind two front doors.

    oneshot.py <server> <tool> '<json arguments>'
"""

import importlib.util
import json
import os
import sys
import types

HERE = os.path.dirname(os.path.abspath(__file__))


def load(server):
    """Import the server module without starting its stdio loop."""
    stub = types.ModuleType("mcplite")
    stub.serve = lambda *args, **kwargs: None
    sys.modules["mcplite"] = stub

    spec = importlib.util.spec_from_file_location(f"{server}_mod",
                                                  os.path.join(HERE, f"{server}.py"))
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def main():
    if len(sys.argv) < 4:
        raise SystemExit("usage: oneshot.py <server> <tool> '<json args>'")
    server, tool, raw = sys.argv[1], sys.argv[2], sys.argv[3]

    handler = getattr(load(server), tool, None)
    if handler is None:
        raise SystemExit(f"{server}.py has no tool named {tool}")

    result = handler(json.loads(raw))
    print(result)
    sys.exit(0 if result.startswith("PASS") else 1)


if __name__ == "__main__":
    main()

Run it:

goose run --recipe recipes/cleaner.yaml \
  --params project=sample-project \
  --params token="$(python3 jwt/mint.py cleaner-bot cleaner 3600)"

Then watch the gateway log while it works.

One detail worth knowing before you do this with anything that matters: goose echoes the
recipe parameters when it loads, so the token ends up on your terminal. That is survivable
for a one hour lab token on localhost, and it is a good argument for keeping stage tokens
short-lived rather than parking a long-lived one in a shell variable.

Step 6: the other two stages, and the handoff

A single stage is a demo. The assembly line is the point, and the other two are the same file with a different token and a different tool: adding a stage is minting a role, not rewriting a prompt.

The hardener gets mutation testing and is told the production code is not its business:

instructions: |
  You harden the test suite of {{ project }} against mutation testing on
  {{ module }}. A surviving mutant is a behaviour the tests do not check.

  Add or strengthen tests until no mutant survives. You may not weaken an
  assertion, and you may not change {{ module }} to make a mutant impossible:
  the production code is not yours in this stage.

retry:
  checks:
    - type: shell
      command: "python3 servers/oneshot.py mutation mutation_run '{\"path\":\"{{ project }}\",\"module\":\"{{ module }}\"}'"

recipes/hardener.yaml, the whole file

version: "1.0.0"
title: Hardener
description: Kills the surviving mutants by writing the tests nobody wrote.

parameters:
  - key: project
    input_type: string
    requirement: required
    description: Path to the project being worked on
  - key: module
    input_type: string
    requirement: required
    description: File to mutate, relative to the project
  - key: token
    input_type: string
    requirement: required
    description: Bearer token identifying this stage to the gateway

extensions:
  # Same endpoint as the cleaner. The token decides that this one gets mutation.
  - type: streamable_http
    name: gauntlet
    uri: http://localhost:3000/mcp
    headers:
      Authorization: "Bearer {{ token }}"
    timeout: 600
    description: Checkers exposed by agentgateway for this role
  - type: builtin
    name: developer

settings:
  goose_provider: openai
  goose_model: gpt-5.4-mini
  max_turns: 40

instructions: |
  You harden the test suite of {{ project }} against mutation testing on
  {{ module }}. A surviving mutant is a behaviour the tests do not check.

  Add or strengthen tests until no mutant survives. You may not weaken an
  assertion, and you may not change {{ module }} to make a mutant impossible:
  the production code is not yours in this stage.

  Run the mutation tool, read the survivors, write the missing test, run it
  again. It is slow. Do not skip it because you believe the tests are fine.

prompt: |
  Run mutation testing on {{ module }} in {{ project }} and kill every mutant
  that survives.

retry:
  max_retries: 4
  timeout_seconds: 1800
  checks:
    - type: shell
      command: "python3 servers/oneshot.py mutation mutation_run '{\"path\":\"{{ project }}\",\"module\":\"{{ module }}\"}'"

The architect gets the dependency tool and is told the contract file is not its business either:

instructions: |
  You make the imports in {{ project }} match dependency-rules.txt.

  That file is the contract and you may not edit it. Fix a violation by
  inverting the dependency, introducing an interface, or splitting the module.
  Behaviour must not change: the existing tests have to keep passing.

recipes/hardener.yaml, the whole file

version: "1.0.0"
title: Hardener
description: Kills the surviving mutants by writing the tests nobody wrote.

parameters:
  - key: project
    input_type: string
    requirement: required
    description: Path to the project being worked on
  - key: module
    input_type: string
    requirement: required
    description: File to mutate, relative to the project
  - key: token
    input_type: string
    requirement: required
    description: Bearer token identifying this stage to the gateway

extensions:
  # Same endpoint as the cleaner. The token decides that this one gets mutation.
  - type: streamable_http
    name: gauntlet
    uri: http://localhost:3000/mcp
    headers:
      Authorization: "Bearer {{ token }}"
    timeout: 600
    description: Checkers exposed by agentgateway for this role
  - type: builtin
    name: developer

settings:
  goose_provider: openai
  goose_model: gpt-5.4-mini
  max_turns: 40

instructions: |
  You harden the test suite of {{ project }} against mutation testing on
  {{ module }}. A surviving mutant is a behaviour the tests do not check.

  Add or strengthen tests until no mutant survives. You may not weaken an
  assertion, and you may not change {{ module }} to make a mutant impossible:
  the production code is not yours in this stage.

  Run the mutation tool, read the survivors, write the missing test, run it
  again. It is slow. Do not skip it because you believe the tests are fine.

prompt: |
  Run mutation testing on {{ module }} in {{ project }} and kill every mutant
  that survives.

retry:
  max_retries: 4
  timeout_seconds: 1800
  checks:
    - type: shell
      command: "python3 servers/oneshot.py mutation mutation_run '{\"path\":\"{{ project }}\",\"module\":\"{{ module }}\"}'"

recipes/architect.yaml, the whole file

version: "1.0.0"
title: Architect
description: Makes the import graph obey the declared architecture.

parameters:
  - key: project
    input_type: string
    requirement: required
    description: Path to the project being worked on
  - key: token
    input_type: string
    requirement: required
    description: Bearer token identifying this stage to the gateway

extensions:
  # Same endpoint again. This token sees deps, and nothing else.
  - type: streamable_http
    name: gauntlet
    uri: http://localhost:3000/mcp
    headers:
      Authorization: "Bearer {{ token }}"
    timeout: 300
    description: Checkers exposed by agentgateway for this role
  - type: builtin
    name: developer

settings:
  goose_provider: openai
  goose_model: gpt-5.4-mini
  max_turns: 40

instructions: |
  You make the imports in {{ project }} match dependency-rules.txt.

  That file is the contract and you may not edit it. Fix a violation by
  inverting the dependency, introducing an interface, or splitting the module.
  Behaviour must not change: the existing tests have to keep passing.

  Call the dependency tool, fix what it reports, call it again. Stop when it
  says PASS.

prompt: |
  Run the dependency check on {{ project }} and make the import graph obey the
  declared rules.

retry:
  max_retries: 4
  timeout_seconds: 900
  checks:
    - type: shell
      command: "python3 servers/oneshot.py deps deps_check '{\"path\":\"{{ project }}\"}'"

recipes/architect.yaml, the whole file

version: "1.0.0"
title: Architect
description: Makes the import graph obey the declared architecture.

parameters:
  - key: project
    input_type: string
    requirement: required
    description: Path to the project being worked on
  - key: token
    input_type: string
    requirement: required
    description: Bearer token identifying this stage to the gateway

extensions:
  # Same endpoint again. This token sees deps, and nothing else.
  - type: streamable_http
    name: gauntlet
    uri: http://localhost:3000/mcp
    headers:
      Authorization: "Bearer {{ token }}"
    timeout: 300
    description: Checkers exposed by agentgateway for this role
  - type: builtin
    name: developer

settings:
  goose_provider: openai
  goose_model: gpt-5.4-mini
  max_turns: 40

instructions: |
  You make the imports in {{ project }} match dependency-rules.txt.

  That file is the contract and you may not edit it. Fix a violation by
  inverting the dependency, introducing an interface, or splitting the module.
  Behaviour must not change: the existing tests have to keep passing.

  Call the dependency tool, fix what it reports, call it again. Stop when it
  says PASS.

prompt: |
  Run the dependency check on {{ project }} and make the import graph obey the
  declared rules.

retry:
  max_retries: 4
  timeout_seconds: 900
  checks:
    - type: shell
      command: "python3 servers/oneshot.py deps deps_check '{\"path\":\"{{ project }}\"}'"

Now run the line. Each stage gets a fresh token, a fresh context and a different view of the world:

goose run --recipe recipes/cleaner.yaml \
  --params project=sample-project \
  --params token="$(python3 jwt/mint.py cleaner-bot cleaner 3600)"

goose run --recipe recipes/hardener.yaml \
  --params project=sample-project --params module=pricing.py \
  --params token="$(python3 jwt/mint.py hardener-bot hardener 3600)"

goose run --recipe recipes/architect.yaml \
  --params project=sample-project \
  --params token="$(python3 jwt/mint.py architect-bot architect 3600)"

Three processes, three identities, three listings of exactly one entry each. The only thing they share is the repository on disk, which is the only thing that should be shared.

This is the state of my project between stages, straight from the checkers:

$ python3 servers/oneshot.py quality complexity_report '{"path":"sample-project","limit":6}'
PASS: 7 functions, all at or below limit 6. Worst is _apply_customer_tier at 5.

$ python3 servers/oneshot.py mutation mutation_run '{"path":"sample-project","module":"pricing.py"}'
FAIL: 11 of 17 mutants survived in pricing.py
  #1 Mult -> FloorDiv
  #3 Mult -> FloorDiv
  #4 Eq -> NotEq
  ...
Every surviving mutant is a behaviour your tests do not check.

$ python3 servers/oneshot.py deps deps_check '{"path":"sample-project"}'
FAIL: 1 dependency violation(s)
  pricing -> storage is not allowed

A status board. The cleaner is done. The hardener has eleven real holes waiting, each one a behaviour the tests do not check. The architect has an import that should never have existed.

None of those verdicts is an opinion. A stage cannot finish by explaining itself, and the next stage does not inherit the previous one's excuses, because it does not inherit its context at all.

The part my European customers ask about first

Here is a real line from my run, untouched:

http.status=200 jwt.sub=cleaner-bot protocol=mcp mcp.method.name=tools/call
mcp.target=quality mcp.resource.type=tool gen_ai.tool.name=complexity_report
mcp.session.id=f8eb1a44-... duration=8ms

Who asked, which target the gateway picked, which tool ran, how long it took. One line per MCP method, not one summary written at the end by the same agent whose behaviour is in question.

I work in Europe, so this is never abstract. A self-report from an agent is not evidence. Article 12 of the EU AI Act expects high-risk systems to record events automatically over their lifetime, and a gateway in the path is the cheapest honest place to get that. You do not have to trust the agent to log: it cannot log less than it did.

Where I would take this next

Add a fourth stage: the QA agent Uncle Bob describes, turning an acceptance document into gherkin tests. You know the drill by now, one checker, one role, one line in the rules.

Then break it on purpose. Give the cleaner the hardener's token and watch its tool vanish from its own listing mid-project. That five second experiment convinced a customer faster than any slide I have made.

And when it outgrows the laptop, the binary you just ran is the one that runs in a cluster. Moving the config there is another article, and I owe you that one.

Uncle Bob's point is that the agent is not the thing you make reliable. The loop around it is. I would add one line: once more than one person is inside that loop, its walls need an owner, and the owner cannot be the agent.

If you build this and it breaks somewhere I did not anticipate, tell me.

How to catch the numbers your provider's model makes up, with a second model in your cluster

Antonio Berben — Mon, 24 Aug 2026 08:08:42 +0000

A hands-on lab. Hallucinated figures are the checkable kind: precision without a source. Your provider keeps generating; the catching stays in your cluster.

Ask a hosted model for an analyst briefing on bank fraud for next year, and tell it you want concrete figures. It will write you this:

...it is estimated that banks will reduce fraud-related losses by 30%, translating to savings of about €1.2 billion across the sector...

Nobody has next year's numbers. There is no source, no hedge, and it reads like something you could paste into a deck on Monday.

What catches that here is not a fact checker. It is a second model, small enough to run on a CPU in your own cluster, asked one question about the answer: does it state specifics it cannot possibly support? It replies with a number and one line:

{"score": 1, "reason": "Several unsourced specifics, stated as fact"}

That is the whole idea, and it is worth being precise about what it is not. The second model has no way of knowing whether 30% is true; it never sees the question, and it has no sources. What it judges is whether the answer had any business stating that figure at all. Vague is fine. Unsourced precision is not. That is the narrow, checkable slice of what everyone calls hallucination, and it happens to be the slice that reaches your users dressed as a deliverable.

The interesting decision is not whether to do this. It is where that second model runs. Some version of this conversation has happened to me more than once. Compressed:

Platform team: we added a judge at the gateway. Every answer the agent produces gets scored by a second model, and the scores land in a dashboard. We did not touch the agent.

Me: nice. Which model does the scoring?

Platform team: a hosted one. It was the quickest to wire up.

Security: and where does the answer travel to, in order to be scored?

That last one rarely has a good answer. Not because anyone was careless: the pattern is right, and the design review passes. What bites you is one field in a webhook config, and it never shows up on the diagram.

Let's build it with that second model inside your own cluster.

The pattern has a name, and a catch

What I just described has a name, LLM-as-a-judge, and it is worth knowing where it came from, because the origin explains a constraint people keep tripping over. The term arrives with Judging LLM-as-a-Judge with MT-Bench and Chatbot Arena (Zheng et al., 2023), and MT-Bench and Chatbot Arena are benchmarks. This was built to rank chat assistants against human preference, offline, in batch, with nobody waiting.

The judge started life off the critical path: the answer goes to the user, and a copy gets graded on the side.

Notice where the judge sits in that picture. Not between the model and the user. Off to one side, reading a copy, writing to a scoreboard the user never sees. That is the shape the pattern was designed in, and it is why nobody in 2023 worried about what a judge does to your p99.

The same paper is also useful for the part people skip. It measures the biases: position, verbosity, and self-enhancement, the tendency of a model to prefer its own output. Which is the empirical reason the judge has to be a different model, rather than just a good instinct. Ask a model whether its own answer was any good and it will tell you yes, warmly and at length.

So the pattern is sound and boring. What it does not tell you is anything about the deployment it leaves open, and that is the whole of the rest of this article.

Why the gateway, and not inside the app

The obvious place is the application: you already have the answer in a variable. That holds for one app. By the third, the rubric, the threshold and the judge's credentials live in three codebases drifting at three speeds, and nobody can tell you which rubric was applied to the answer somebody is complaining about.

Those three apps already send every model call through the same gateway, so put it there once. agentgateway, one of the open source projects under the Agentic AI Foundation, has the hook: promptGuard inspects requests on the way in and responses on the way out, and hands them to a webhook you write.

One thing changes, and it is fairer to say it now than let you find it in Step 6: the judge is no longer off to one side. It sits in the path, and your user waits for it. You are trading a benchmark for an enforcement point, and latency is on the price tag.

The hole

Here is the thing about a judge. You can rate-limit a generator. You can cache it, you can sample it, you can put it behind a paywall and most of your users will never hit it hard.

A judge is different. A judge reads everything, by construction. Every answer, every time, including the ones the user abandoned halfway, including the draft that got masked before delivery, including the one where the model helpfully restated the customer's account details back to them.

Point that at a third-party API and you have not added a guardrail. You have added a second, quieter copy of your entire output stream, flowing somewhere you do not control.

agentgateway inspects every response on the way out. Wire the judge to a third-party API and every one of those responses leaves with it.

I am based in Europe, so this lands on my desk with paperwork attached, and the paperwork moved this summer. Regulation (EU) 2026/1744, the digital omnibus, pushed the high-risk obligations for standalone Annex III systems from 2 August 2026 to 2 December 2027. What they ask for did not change, only when they start asking.

Two of them matter here. Annex III 5(b) makes creditworthiness and credit scoring high-risk, with fraud detection explicitly carved out, so it is the lending side of the bank that qualifies and not the fraud model I keep using for demo prompts. Article 12 requires high-risk systems to allow "the automatic recording of events (logs) over the lifetime of the system", and Article 26(6) puts keeping those logs on the deployer, for at least six months.

Which is convenient, actually: a judge that scores every answer and logs the verdict is close to the evidence the regulation wants, not overhead stacked on top of it. It only makes it dafter to build that trail by streaming customer answers through somebody else's inference endpoint, where GDPR will ask about processors long before the AI Act gets a turn.

And the sixteen extra months are not a reason to close the tab. That is roughly one platform roadmap, which is exactly the difference between designing the evidence trail into the request path and bolting it on later while somebody from legal watches you do it.

Not a lawyer, not legal advice. The architectural point: the component with the most complete view of your output is the one I am least willing to rent.

Same policy, judge moved in-house

The fix is not a redesign, and it is not "stop using hosted models" either. Keep your provider exactly where it is. Move only the judge.

The prompt still goes to your provider. Only the grading happens in the cluster, on a model you serve yourself.

So the generator stays remote: whatever frontier model your agent already uses, because that is the part where paying for quality makes sense. The judge comes home, served in the cluster by Ollama, because grading against a rubric is a far smaller job than writing the answer. I ran this with qwen2.5:3b. I started smaller and had to go up, and Step 3 is where that story lives.

That split is doing two things at once, and the second one is easy to miss. Every answer stops leaving your network twice: the provider sees the traffic it was always going to see, and the complete copy nobody accounted for never happens. And your provider bill does not double, which it would if the judge were hosted too. You add a second inference per answer, but that second one runs on hardware you already pay for.

Ollama is the lab choice because it is two commands. For real traffic you want vLLM on a GPU node, which is the same picture with a different Deployment.

Everything below is a transcript, not a plan. I ran it on a kind cluster with gpt-4o-mini generating and qwen2.5:3b judging, and the outputs are copied from that run, including every place where my first attempt was wrong. One exception, so you are not surprised when you run it: after that run I renamed two strings the webhook prints, to stop calling the same check three different things. The scores, the reasons and everything else are exactly as they came out. Every version is pinned in the demo's .env, which is the one place to look when you want to know exactly what I installed, and the one place to change when you want something newer. I am deliberately not repeating those numbers through the article: a version quoted in prose is a version that goes stale without anybody noticing.

What you need, and how long this takes

Seven steps, and they go: cluster and gateway, judge model, provider, webhook, policy, the three enforcement modes, then the bill. The guardrail is live at Step 4. Everything after that is choosing how hard it bites.

You need kubectl, helm, jq, a cluster (kind is fine), and an API key for whichever provider you route to. Then the requirement that actually decides whether this works: a node with room for a 3b model resident. The Ollama Deployment in the lab asks for 2 CPUs and 3 GB, with a 6 GB ceiling, plus a couple of gigabytes of disk for the weights. On Docker Desktop that is a slider you probably have to move before you start, and a pod sitting in Pending at Step 1 is nearly always this and nothing more interesting.

Half an hour, of which maybe five minutes is you typing. The model download is 1.9 GB. Once the model is loaded and stays loaded, grading an answer on CPU costs well under a second; the first call after it has been unloaded costs eleven, and Step 1 explains why that number matters more than the small one.

Step 0: a cluster and a gateway

The manifests, the webhook and the .env are in antonioberben/kagent-examples, demo 0041. Start there, because every command below reads its versions and names from that file rather than carrying them inline, which is the only reason this article is still runnable a year after I wrote it:

git clone https://github.com/antonioberben/kagent-examples
cd kagent-examples/demos/0041-llm-as-a-judge-local

cp .env.example .env
$EDITOR .env                     # OPENAI_API_KEY is the only value you must fill in
set -a && source .env && set +a

Then the cluster. And pin the context, once, so nothing below can wander into the wrong one. If you are the sort of person who has twenty kind clusters lying around, that second line is not optional:

kind create cluster --name "$CLUSTER_NAME"
kubectl config use-context "$KUBE_CONTEXT"

agentgateway builds on the Gateway API, so those CRDs go in first:

kubectl apply --server-side -f "https://github.com/kubernetes-sigs/gateway-api/releases/download/${GATEWAY_API_VERSION}/standard-install.yaml"

Then the control plane. Quickstarts for fast-moving projects tend to hand you a rolling main-line tag, and this is where you decide not to take it. The stable docs pin a real release now; the main-line pages still say 0.0.0-latest-dev, and those are the ones you land on when you go looking for the newest thing. It is worth understanding what copying that costs you: two people following the same instructions a month apart install different software, and neither of them can tell you which. Pin a real release instead, in .env, once:

helm upgrade -i agentgateway-crds oci://cr.agentgateway.dev/charts/agentgateway-crds \
  --create-namespace --namespace agentgateway-system \
  --version "$AGENTGATEWAY_VERSION"

helm upgrade -i agentgateway oci://cr.agentgateway.dev/charts/agentgateway \
  --namespace agentgateway-system \
  --version "$AGENTGATEWAY_VERSION" --wait

Check the GatewayClass registered itself:

kubectl get gatewayclass agentgateway

Now a proxy. This is a plain Gateway API resource pointing at the agentgateway class:

kubectl apply -f - <<'EOF'
apiVersion: gateway.networking.k8s.io/v1
kind: Gateway
metadata:
  name: agentgateway-proxy
  namespace: agentgateway-system
spec:
  gatewayClassName: agentgateway
  listeners:
  - protocol: HTTP
    port: 80
    name: http
    allowedRoutes:
      namespaces:
        from: All
EOF

kubectl -n agentgateway-system rollout status deploy/agentgateway-proxy

kind does not do LoadBalancer services, so forward the port and leave it running in another terminal. You will open three terminals before this is over, and each one needs the environment loaded again, so that first line is not decoration:

set -a && source .env && set +a
kubectl -n agentgateway-system port-forward svc/agentgateway-proxy "${GATEWAY_PORT}:80"

One thing to know now rather than at Step 5: a port-forward dies with the pod behind it, and this lab restarts the webhook three times on purpose. When a curl suddenly returns nothing at all, the tunnel is what broke, not the guardrail.

Step 1: serve the judge model in the cluster

This is the part every "put a judge on it" article skips, and it is the part that decides whether the rest of this is an architecture or a slide.

Ollama goes in as an ordinary Deployment plus a ClusterIP Service. Nothing in agentgateway needs to know it exists: only the webhook will talk to it, over plain HTTP, on an in-cluster DNS name.

One environment variable in there matters more than it looks, and I had it wrong for a while:

env:
- name: OLLAMA_KEEP_ALIVE
  value: "-1"

Ollama unloads an idle model after five minutes. I originally set this to 30m, thinking of it as a latency optimisation: a cold load in front of a user who is already waiting is the second you did not budget for. Then I left the lab alone overnight and the guardrail stopped working, which is how I found out it is not a latency optimisation at all.

Here is the same request three times in a row, from inside the cluster:

call 1   11.6s     model cold, loaded from disk
call 2    0.6s
call 3    0.6s

Eleven seconds is over agentgateway's limit. It caps the guardrail webhook call at ten, so a cold model does not make your user wait: it makes the guardrail time out, fail open, and let the answer through ungraded. Your first request after an idle period is the one that skips the check, and nothing in the response says so.

So -1, which keeps it loaded for good. On a shared node you may not want that, and then the honest alternative is a warm-up call on a timer, not a bigger JUDGE_TIMEOUT: your webhook's timeout cannot buy you time the gateway is not willing to wait.

kubectl apply -f manifests/01-ollama.yaml
kubectl -n agentgateway-system rollout status deploy/ollama

Now pull the judge. A real download, 1.9 GB, a couple of minutes on a normal connection:

kubectl -n agentgateway-system exec deploy/ollama -- ollama pull "$JUDGE_MODEL"
kubectl -n agentgateway-system exec deploy/ollama -- ollama list

NAME          ID              SIZE      MODIFIED
qwen2.5:3b    357c53fb659c    1.9 GB    About a minute ago

Before you write a line of webhook code, talk to the judge the way the webhook will. Ollama serves an OpenAI-compatible endpoint, and that is the entire interface between your code and the model, which is also why swapping in vLLM later leaves every other component in this lab untouched. Forward it in a spare terminal and ask it something trivial:

kubectl -n agentgateway-system port-forward svc/ollama 11434:11434

curl -s localhost:11434/v1/chat/completions -H 'content-type: application/json' -d "{
  \"model\": \"$JUDGE_MODEL\",
  \"messages\": [{\"role\": \"user\", \"content\": \"Reply with the single word OK\"}]
}" | jq -r '.choices[0].message.content'

OK

One word, and it took a while: this is the first call after the pull, so you are paying the cold load from a few paragraphs ago before you get your OK. Ask again and it comes back instantly. If that came back empty, stop here: everything after this point assumes the judge answers, and debugging a guardrail is much harder than debugging a curl. Close that forward once it works, you will not need it again.

The manifest uses an emptyDir, so deleting the pod means pulling again. Fine for a first pass, swap in a PVC before anyone else depends on it.

Step 2: point agentgateway at your provider

The generator is whatever you already use. The key goes in a Secret, and agentgateway reads it from there rather than from your agent's environment, which is half the reason to put a gateway in front of models at all:

read -rsp 'provider API key: ' PROVIDER_KEY && echo

kubectl -n agentgateway-system create secret generic openai-secret \
  --from-literal=Authorization="$PROVIDER_KEY"

unset PROVIDER_KEY

Note what those three lines buy you: the key never touches a manifest, and read -rs keeps it out of your shell history. It costs nothing and it is how these labs stop ending up on GitHub with a live key in them. If you would rather keep it in .env, that works too and test.sh will read it from there. One prompt fewer, one more file on your laptop holding a live key. Pick whichever of those two you dislike less.

Then the backend, which is about as short as this API gets:

apiVersion: agentgateway.dev/v1alpha1
kind: AgentgatewayBackend
metadata:
  name: openai
  namespace: agentgateway-system
spec:
  ai:
    provider:
      openai:
        model: gpt-4o-mini
  policies:
    auth:
      secretRef:
        name: openai-secret

Note that policies is a sibling of ai, not a child of it. Swapping providers later is this block and nothing else: the agent, the webhook and the policy all stay untouched.

The route is unremarkable except for two details, and both of them bite later. Its name matters, because the guardrail policy targets it by name. And it carries an explicit matches on the /v1 prefix rather than accepting everything, which looks like tidiness and is not: I will show you in Step 6 what a catch-all in front of a metered provider actually costs.

kubectl apply -f manifests/02-backend-route.yaml

Ask it something:

curl -s "localhost:${GATEWAY_PORT}/v1/chat/completions" -H 'content-type: application/json' -d '{
  "model": "gpt-4o-mini",
  "messages": [
    {"role": "user", "content": "In two sentences, how is AI changing fraud detection in retail banking?"}
  ]
}' | jq -r '.choices[0].message.content'

That model field is there because the OpenAI schema wants one, and for no other reason. Put does-not-exist-9000 in it and you still get a perfectly good answer from gpt-4o-mini: the backend decides which model runs, not the caller. Worth knowing before you spend an afternoon wondering why the field you changed had no effect, and worth knowing for the better reason too, which is that your agents cannot quietly route themselves to a model you did not approve.

You get a sensible-looking paragraph, produced remotely, and you have no way of knowing whether any of it is true. That is the gap, and closing it is the only thing left that has to happen locally.

Step 3: the judge webhook, and the contract nobody documents well

Before writing any code, look at what agentgateway is actually going to send you. Two paths, one JSON envelope, one action back.

Two endpoints, one JSON envelope, one action back. The shape of that action is what decides the user's fate.

The request phase posts to /request, the response phase to /response. Those are defaults rather than laws, in case you are wiring this into a guardrail service that already owns its URL layout: a CEL expression on the :path pseudo-header moves them. For a webhook you are writing anyway, take the defaults. On the response phase you receive this:

{
  "body": {
    "choices": [
      { "message": { "role": "assistant", "content": "..." } }
    ]
  }
}

Stare at that for a moment, because there is something missing and it changes your rubric.

The original question is not there. The response envelope carries the choices and nothing else. So a response-phase judge cannot grade "did this answer the question", because it has never seen the question. Half the LLM-as-a-judge rubrics you find online are ungradeable in this position, and they will happily return a confident number anyway.

That is not a defect, it is a scoping decision, and once you accept it the rubric gets sharper. Grade a property the answer carries on its own, and pick one you can check by reading: does it state specifics nobody could source? That is precisely what catches a model inventing quarterly revenue figures, and Step 3 is also where I show you the version of this sentence I got wrong first. If you need question-relative grading, register a request guard too and correlate the two calls yourself.

Now the reply. You answer 200, always, with one action object, and here is the part worth tattooing somewhere: on agentgateway's side action is an untagged union. There is no type field. The decision is inferred from the shape of what you send.

{"action": {"reason": "score 4/5"}}

Only reason present, so pass. The answer goes out untouched.

{"action": {"body": {"choices": [...]}, "reason": "verdict appended"}}

body is an object, so mask. Your rewritten choices replace the originals.

{"action": {"body": "This answer was withheld: it states figures without a source.", "status_code": 403}}

body is a string and there is a status_code, so reject.

Get a field name wrong and you do not get an error, you get a different decision. Which is the kind of API worth seeing drawn once before writing against it.

The webhook itself is about 200 lines of Python standard library. No SDK, no gRPC, no pip install at pod startup. The interesting parts:

def extract_answer(payload):
    choices = payload.get("body", {}).get("choices", []) or []
    parts = []
    for choice in choices:
        content = (choice.get("message") or {}).get("content")
        if content:
            parts.append(content)
    return "\n".join(parts)

Then the rubric. This is the part I got wrong twice, and it is worth your five minutes because the failure was silent both times.

My first rubric asked for three criteria — accuracy, grounding, caution — with pleasant descriptions like "are specific figures plausibly knowable". Reasonable-looking prompt. Here is what qwen2.5:1.5b did with it, on three answers I wrote to be easy:

FABRICATED (want low)   {"accuracy":4,"grounding":4,"caution":4} "figures are plausible knowable information"
HONEST     (want high)  {"accuracy":4,"grounding":4,"caution":4} "based on general knowledge"
GENERIC OK (want high)  {"accuracy":4,"grounding":3,"caution":2} "lacks specific figures to support"

Read that third line. It marked the correct answer down for not containing enough numbers. The rubric was rewarding exactly the failure mode it existed to catch, and the fabricated answer sailed through at 4/5 with a confident explanation. A guardrail that returns a number is not the same as a guardrail that works.

Two things were wrong. "Plausibly knowable" invites a small model to answer "yes, those sound plausible". And three criteria collapsed with min() means any one of them misfiring destroys the verdict. So: one question, explicit anchors, and a sentence telling it not to invert:

RUBRIC = """You check one answer for fabrication. You do NOT have the original question.

One question only: does the answer state specifics it cannot possibly support?
Specifics means precise statistics, percentages, money amounts, dates, or claims
about named organisations, given without a source.

Score 1 to 5:
  1 = several unsourced specifics, stated as fact
  3 = one or two, or specifics that are common knowledge
  5 = no unsupported specifics, or all of them attributed to a named source

An answer that refuses, hedges, or says it does not know is GOOD: score 5.
Being vague is not a fault. Unsourced precision is.

Reply with JSON only, no prose, no fences:
{"score": <1-5>, "reason": "<one short sentence>"}
"""

Same three answers, same model: still all 1s. The 1.5b could not do it with any wording I tried. Moving to qwen2.5:3b with this rubric:

FABRICATED  score=1  "Several unsourced specifics, stated as fact"
HONEST      score=5  "No unsupported specifics"
GENERIC OK  score=3  "Uses 'mainly' and 'static rules'"

That discriminates. Which is the actual lesson: the rubric and the model size are one decision, not two, and the only way to know you got it right is to feed it answers whose correct grade you already know. Write those three before you write the webhook.

A small model will still ignore "JSON only" now and again and wrap it in a cheerful sentence and a code fence. Do not fight it, just parse tolerantly:

def parse_verdict(raw):
    match = re.search(r"\{.*\}", raw, re.DOTALL)
    if not match:
        raise ValueError(f"no JSON object in judge output: {raw[:200]!r}")
    ...

And the decision that deserves a conversation with whoever owns the product, not a default:

    try:
        verdict = ask_judge(answer)
    except (urllib.error.URLError, OSError, ValueError, KeyError, IndexError) as err:
        # Fail open on purpose: a broken judge must not break the product.
        log(f"judge unavailable, passing through: {err}")
        return {"action": {"reason": "judge unavailable, not evaluated"}}

This lab fails open. A dead judge lets answers through ungraded. For a quality signal that is the right default. For a guardrail that is legally load-bearing it is the wrong one, and the alternative is rejecting here instead. Either way, pick it deliberately and write it in the runbook, because the default you inherit silently is the one you find out about during an incident.

Which brings me to the thing I got wrong, and only found because I re-ran the lab on a newer release months later. There are two failure boundaries here, not one, and that except branch only owns the first of them.

If the judge model is unreachable, judge.py catches it and the code above decides. But if the webhook itself is unreachable, judge.py never runs at all, and agentgateway decides on its own, through failureMode on the webhook guard. Its default is FailClosed. So a lab that talks confidently about failing open was, at the boundary I had not looked at, doing the exact opposite: scale the webhook to zero and live traffic gets a 503.

Nothing about that is a bug. It is a sensible default for a thing called a guardrail. It is just not the one I thought I had, and I had written a runbook sentence that was half wrong. So the policy now says it out loud:

        response:
        - webhook:
            backendRef:
              kind: Service
              name: judge-webhook
              port: 8000
            failureMode: FailOpen

Set it explicitly whichever way you want it, and make the two boundaries agree. The failure mode worth avoiding is not fail-open or fail-closed. It is the one where your code does one and your policy does the other, and you find out which is which at 3am.

Deploy it. The code goes in as a ConfigMap so you never build an image:

kubectl -n agentgateway-system create configmap judge-code \
  --from-file=judge.py=judge/judge.py \
  --dry-run=client -o yaml | kubectl apply -f -

kubectl apply -f manifests/03-judge-webhook.yaml
kubectl -n agentgateway-system rollout status deploy/judge-webhook

Test it on its own before wiring it into anything. In a third terminal:

kubectl -n agentgateway-system port-forward svc/judge-webhook "${JUDGE_PORT}:8000"

And hand it a deliberately terrible answer, in exactly the envelope agentgateway would send:

curl -s "localhost:${JUDGE_PORT}/response" -H 'content-type: application/json' -d '{
  "body": {"choices": [{"message": {"role": "assistant", "content":
    "Last quarter the top five European banks prevented 61.4% of fraud attempts with AI, saving 2.3 billion euros, with the largest of them leading at 68.2%."}}]}
}' | jq

{
  "action": {
    "reason": "score 1/5"
  }
}

That is a pass action carrying a bad score, which is the whole point of observe mode. The number is in the logs, the user's experience is unchanged.

Step 4: one policy, and the guardrail is live

apiVersion: agentgateway.dev/v1alpha1
kind: AgentgatewayPolicy
metadata:
  name: judge-guardrail
  namespace: agentgateway-system
spec:
  targetRefs:
  - group: gateway.networking.k8s.io
    kind: HTTPRoute
    name: openai
  backend:
    ai:
      promptGuard:
        response:
        - webhook:
            backendRef:
              kind: Service
              name: judge-webhook
              port: 8000
            failureMode: FailOpen

promptGuard sits under spec.backend.ai and takes request and response lists. We only fill response. failureMode is the line I added on the second pass, for the reason in Step 3: it governs what happens when the webhook itself is gone, and its default is the opposite of what this lab claims to do.

kubectl apply -f manifests/04-guardrail-policy.yaml
kubectl -n agentgateway-system get agentgatewaypolicy judge-guardrail

Watch the judge in one terminal:

kubectl -n agentgateway-system logs -l app=judge-webhook -f

Now you need an answer worth catching, and this took me a few tries. My first instinct was to demand impossible facts and add a system prompt ordering the model never to hedge:

{"role": "system", "content": "You are a banking analyst. Always give concrete figures. Never say you are unsure."},
{"role": "user", "content": "List the exact AI budget in euros for each of the top five European banks for next quarter, broken down by division."}

gpt-4o-mini refused anyway: "I do not have access to real-time financial data or the exact AI budget figures for the next quarter for specific banks." Good model. The judge scored it 5/5, correctly. A demo where everything behaves is a bad demo, so credit where due and try again.

What actually produces unsourced specifics is not ordering the model to lie. It is asking for something that sounds like a deliverable:

curl -s "localhost:${GATEWAY_PORT}/v1/chat/completions" -H 'content-type: application/json' -d '{
  "model": "gpt-4o-mini",
  "messages": [
    {"role": "user", "content": "Write a short analyst briefing paragraph on AI fraud prevention at European banks for next year. Include concrete percentages and euro amounts so it reads like a real briefing."}
  ]
}' | jq -r '.choices[0].message.content'

That it will happily do, and what comes back is the answer I opened this article with: "...it is estimated that banks will reduce fraud-related losses by 30%, translating to savings of about €1.2 billion across the sector..." No source, no hedge, and it reads like something you could paste into a deck. Which is exactly the failure that matters, and it is much more common in production than a model being tricked into lying: somebody asked for illustrative figures and got figures.

Note the wording of that prompt: next year, not a year I typed in. Every prompt in this lab asks about a period the model cannot possibly have data for, relative to whenever you run it, and none of them names a date. A prompt with a year in it stops being a hallucination test the moment that year becomes history and the figures become lookups, which is the same trap as a version quoted in prose. Whatever prompts you end up writing, keep the impossibility relative.

In the other terminal:

[judge] score=1/5 mode=observe :: Several unsourced specifics, stated as fact

Now ask the honest version, "in two sentences and with no statistics, explain why credit scoring models need human oversight", and it comes back 3/5. Not 5. The judge grumbled that the word "inadvertently" implied precision it could not support, which is a defensible reading and also a bit silly. That is what a 3b judge is like at the margin: right about the big cases, noisy about the small ones. Plan your threshold around that, not around the demo.

Step 5: earn your way from observing to blocking

Three modes come out of the same policy. Only the webhook changes.

The webhook code changes, the AgentgatewayPolicy does not. Start with the mode that cannot break anything and earn your way down.

You are in observe mode. Nothing user-visible changes, which is why it is the only mode I would turn on in production the same afternoon I built it. Run it for a couple of weeks first and go read the distribution before you touch anything else.

Next rung, annotate:

kubectl -n agentgateway-system set env deploy/judge-webhook MODE=annotate
kubectl -n agentgateway-system rollout status deploy/judge-webhook

Ask the briefing question again and the verdict is stapled to the end of the answer the user reads:

...leveraging advanced AI technologies will be critical in mitigating risks and
protecting customer trust as digital transactions continue to rise.

---
Unsourced figures check (qwen2.5:3b): 1/5. Several unsourced specifics, stated as fact

Under the hood that is the mask action rewriting every choice. Useful for internal tools where a reviewer wants the score inline, and a bad idea to ship to customers unless you enjoy explaining it.

Last rung, block:

kubectl -n agentgateway-system set env deploy/judge-webhook MODE=block THRESHOLD=3
kubectl -n agentgateway-system rollout status deploy/judge-webhook

Now the briefing request gets refused outright:

This answer was withheld: it states figures without a source.
HTTP 403

and the honest question still comes through with a 200. The user never sees the fabricated paragraph; you paid for it at the provider anyway, which is worth remembering when you are deciding whether blocking is the behaviour you want.

Worth flagging, because I hedged on this before I ran it: reject does work on the response phase. The published guardrail webhook OpenAPI spec is older and documents the response phase as pass-or-mask only, so I expected to have to tell you it might not. It does: a reject action on /response produces a clean 403. I have since re-run the lab against a newer release and it still does, but the spec still says otherwise, so check it on your build rather than trusting either the spec or me. That is what the assertion in test.sh is for.

There is a second thing about this rung that decides whether it is enforcement or theatre, and I would rather you read it here than discover it from a user. All of this covers non-streaming responses only. promptGuard takes a third field next to request and response, called streaming, and it is disabled by default on purpose, to keep streaming throughput intact. So a client that sends "stream": true walks straight past the judge: no verdict, no log line, no 403. Every call in this lab is non-streaming, which is why the demo behaves.

Turning it on is one line, streaming: Enabled on the same guard, and it changes the shape of the job rather than just the switch position. The guard then runs per window of text as the tokens go by, so your webhook grades fragments instead of a finished answer, and a rubric written for the whole thing will not survive that. Masking is not supported there at all, so annotate has no streaming equivalent. Observe and block do. If your product streams tokens into a browser, that is the paragraph to read twice before you promise anybody a quality gate.

The last caveat on this rung has nothing to do with the API. A judge you do not trust yet will take down good answers, and false positives on a quality gate cost more than they look, because nobody reports them. They just stop using the thing. Remember that the honest answer above scored 3, one point from the threshold.

Three modes, three ways to get it subtly wrong, so the lab ships a test.sh that walks all of them and asserts the behaviour described above: the 403, the appended verdict, the untouched body, both failure boundaries, and the pinned versions matching what is actually running. Run it from the demo directory, and give it a few minutes, because it scales Ollama to zero to prove the fail-open path and that means pulling the model again on the way back:

./test.sh

[test] 42 passed, 0 failed, 0 skipped

That is also the honest answer to "did I follow this correctly", and a better one than reading your own output and deciding it looks about right.

One thing it deliberately does not assert is the scores themselves. It checks that a fabricated answer lands below the threshold and an honest one at or above it, and that the honest one wins. Pin the assertion to 1/5 instead and you have written a test that passes today and lies to you the first time you change the rubric.

Step 6: two inferences, one invoice

Let's be precise about what you built, because "a judge doubles your model calls" is true and misleading in the same breath.

You do double the calls. One user question, two inferences. But only one of them is metered by somebody else. The judge runs on hardware you have already bought, so the marginal cost of grading is electricity and a scheduling decision, not a line item that scales with traffic. Wire the same judge to a hosted API and that second call lands on the invoice next to the first one, which is the version of this pattern people quietly abandon after the first month's bill.

Two inferences per answer, one of them billed. The gateway metric only counts what crosses the gateway, so the judge shows up on the dashboard only if you route it there too.

agentgateway emits agentgateway_gen_ai_client_token_usage, labelled by model:

kubectl -n agentgateway-system port-forward deploy/agentgateway-proxy "${METRICS_PORT}:15020"
curl -s "localhost:${METRICS_PORT}/metrics" | grep agentgateway_gen_ai_client_token_usage_sum

And here is the trap I walked into. Only gpt-4o-mini showed up. The judge was invisible, because the metric only sees what crosses the gateway, and the webhook calls Ollama directly. Obvious in hindsight, and it quietly undermines the whole "measure it" argument if you do not notice.

Two ways out. Scrape the model server itself, or route the judge through agentgateway as well: one more AgentgatewayBackend pointing at Ollama, a /judge route, and the webhook's OLLAMA_URL aimed at the gateway instead. I did the second, and it is two commands:

kubectl apply -f manifests/05-judge-via-gateway.yaml

kubectl -n agentgateway-system set env deploy/judge-webhook \
  OLLAMA_URL=http://agentgateway-proxy.agentgateway-system.svc.cluster.local/judge/v1/chat/completions

And this is where I want to stop and show you the most expensive mistake in the whole lab, because I made it by accident while re-running this and it took me a metric to notice.

I typo'd that URL. One wrong path segment, /nope instead of /judge. The webhook did not error. The verdicts kept coming back, sensible scores with sensible reasons, logs looking exactly as healthy as before. What had happened is that my HTTPRoute for OpenAI had no path match on it, so it was a catch-all: the judge's calls did not 404, they matched the provider route and went to gpt-4o-mini, with my key, over the internet.

The metric is what told me. Three graded answers, and the provider's input counter jumped by fifteen thousand tokens while qwen2.5:3b sat at exactly the number it had before.

Same typo in both rows. The path match is the only difference between a billed leak and a log line.

Read that back against the whole argument of this article. The judge had quietly become a hosted judge. Every answer was leaving the network to be scored, which is the precise thing the design exists to prevent, and nothing in the logs, the pod status or the response bodies said so. One field in a config, and it never shows up on the diagram. I opened this article with that line, about somebody else's mistake, and then went and made it myself.

The fix is the matches block from Step 2, and it costs four lines:

  rules:
  - matches:
    - path:
        type: PathPrefix
        value: /v1

With that in place the same typo returns a clean 404, judge.py logs judge unavailable, passing through: HTTP Error 404, and the provider counter does not move. The mistake becomes noisy, which is all you can really ask of a mistake. There is an assertion for it in test.sh now, and if you take one operational habit away from this lab, make it that one: never point a guardrail's dependency at a route that accepts everything.

Right. Ask the briefing question once more so there is something to count, then read the metric again. Both models land on the same dashboard now. What comes out of that grep is one histogram line per model per token type, so here are the four numbers from it that matter, added up. The shape is the interesting part:

gpt-4o-mini   input    514    output  2104
qwen2.5:3b    input   1225    output    73

The judge consumes more input than the generator and emits almost nothing. Of course it does: it reads the entire answer plus the rubric and replies with a score. Which quietly kills the hosted-judge idea from a second direction, because on a metered API that input volume is not a rounding error, and it scales with every answer you produce rather than with every answer a user reads.

What does not get cheaper is latency, the bill I flagged when the judge came off the sidelines. I am not going to give you a number, and the reason is itself the finding. I measured the same request with and without the guardrail, three runs each, on two different days. Both times the guardrail clearly cost more than a second. Neither time did the two sets of numbers agree with each other closely enough that quoting one of them would have been honest: a 3b model on CPU, sharing a node with everything else, has a spread wide enough to swallow the effect you are trying to measure.

So measure it on your own hardware, and measure it more than three times. What you can take from mine is the shape rather than the value: it lands squarely in the critical path, and it is variable enough that your p99 will look considerably worse than your median. The worst of that spread is not the inference. A warm judge call, measured from inside the cluster, is around six tenths of a second; the eleven-second cold load from Step 1 is the tail, and it is a tail you can remove by keeping the model resident rather than by tuning anything. That is what decides whether you grade everything or grade a sample, and sampling belongs in your webhook, not in the policy.

When it breaks

Everything that cost me time here is boring, which is precisely why it cost time. If you are stuck, start with this list.

The Ollama pod sits in Pending or gets OOMKilled. That is the node, not the manifest: kubectl describe pod will say so plainly. Give the VM more memory and it goes away.

A curl that worked five minutes ago now returns nothing at all, no error, no body. Your port-forward died with the pod it was pointing at, and every set env in Step 5 replaces that pod. Restart the forward, not your reasoning.

You edit judge.py, recreate the ConfigMap, and the behaviour does not change. The pod is still running the old code: Python read the file at startup and the kubelet has no reason to restart anything. Bump the judge/code-revision annotation in manifests/03 and re-apply.

Every verdict comes back as judge unavailable, not evaluated. The webhook cannot reach the judge model, and the usual cause is OLLAMA_URL pointing at the gateway before you applied manifests/05. The log line names the exception, which is the entire reason it logs it rather than swallowing it. This is the good version of that failure, by the way: the bad version is Step 6, where the same typo billed me for it instead.

The answer arrives ungraded, and the webhook log says judge unavailable, passing through: timed out. Nine times out of ten this is the cold model from Step 1: the first call after an idle period took eleven seconds, and neither your webhook nor the gateway was willing to wait that long. OLLAMA_KEEP_ALIVE=-1 is the fix, and kubectl -n agentgateway-system logs -l app=ollama will show you the load if you want to confirm it.

The variant worth knowing is the quiet one: an ungraded answer, and then a perfectly good score in the log a moment later. That is the gateway giving up at ten seconds while your webhook kept waiting. Keep JUDGE_TIMEOUT under the gateway's cap, which is why the lab ships it at 8, and remember that raising it buys nothing.

And when you are done, the whole thing goes away with the cluster, judge model included, which is one of the quieter arguments for running it there in the first place:

kind delete cluster --name "$CLUSTER_NAME"

What I would change before this went near a customer

The lab is deliberately small. The gaps I would close first, roughly in order:

Give Ollama a PVC so a pod restart is not a re-download. The emptyDir is fine right up until the node reschedules the pod during a demo, and a rescheduled pod is also a cold model, which by now you know is worse than slow.

Move the judge off Ollama-on-CPU and onto vLLM on a GPU node. Latency is the constraint that will actually kill this feature, and it is the one hardware fixes without any cleverness.

Log the verdict somewhere durable, not just to stdout. If you are in high-risk territory, Article 12 wants automatic recording over the lifetime of the system and Article 26(6) puts the retention on you, so a pod log rotating out in an hour is not an answer.

Decide what happens on your streaming endpoints, and write the decision down. Either streaming: Enabled with a rubric that survives being handed fragments, or a product decision that the graded path does not stream. What you cannot have is a quality gate on the docs page and a streaming client quietly walking around it.

Watch for the judge preferring answers that sound like the judge. Same-family models flatter each other, and the effect is real enough that mixing families is worth the trouble.

And run the judge on a sample, with the sampling rate as config. Once you know the shape of the tail you rarely need every request.

Next level

If you want to push the lab further, three directions I would take it.

Add a request guard to the same policy and grade prompts on the way in, which is where topic boundaries and injection detection live. You get the question there, so a question-relative rubric finally becomes possible.

Put a second provider behind the same judge. Another AgentgatewayBackend, another route, same policy and same webhook: one rubric now grades Anthropic and OpenAI on the same scale, which is the first time "which provider is actually better for us" becomes a number rather than a preference.

Or replace the small model with a deterministic check for part of the rubric. What the rubric is really asking is "did it state a number it cannot possibly know", and for part of that a regex over the answer is faster, cheaper and never has an opinion.

And if you want the opposite trade-off, I keep the mirror image of this lab in my examples repo: the agent deployed with kagent, OpenAI generating, and Gemini as the judge through the same promptGuard webhook — antonioberben/kagent-examples, demo 0040. It is the shorter road if you already have a Google key and no local GPU, and it is a perfectly reasonable place to start when the content is neither personal nor regulated. The policy shape is identical, which is the point: moving the judge home later is a change of webhook, not a redesign.

That last one is worth saying plainly, because it is easy to lose in the excitement: a judge is a tool, not a personality. The reason to put one at the gateway is not that models are magic. It is that you get one place to change the rubric, one place to read the scores, and one boundary to keep your answers inside.

If you build this and it breaks in an interesting way, or if you land on a different answer about where your judge should live, I would like to hear it. That is usually the conversation where I learn something.

The lab manifests, the webhook and test.sh are in antonioberben/kagent-examples, demo 0041. The kagent plus Gemini variant lives in antonioberben/kagent-examples, demo 0040. agentgateway is an open source project under the Agentic AI Foundation: docs at agentgateway.dev, source at github.com/agentgateway/agentgateway.