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
agentgatewaybinary, v1.5.0. -
goosev1.51.0, plus a configured provider: rungoose configureonce, or exportGOOSE_PROVIDER,GOOSE_MODELand 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.
Top comments (1)
“The agent is not the thing you make reliable; the loop around it is” captures the engineering shift well. The next boundary is protecting the checker itself: an agent that cannot call the architecture tool may still try to edit its config, fixtures, or CI wiring unless those artifacts are outside its write scope. Binding each role token to a workflow run and commit SHA could make the authorization and the evidence line up. Does agentgateway support that kind of short-lived, run-scoped claim today?