Workshop: Freeze Agent Output Contracts Before You Swap Endpoints in 80 Minutes

Swapping an inference endpoint without a frozen JSON contract is how agent demos pass and production loops fail. This workshop treats each model reply as a versioned API, not as free-form text that happens to look structured. Students pin required fields, enumerations, and forbidden keys, then hash a golden fixture so later runs fail closed on silent drift. The same harness works against any HTTP completion endpoint, including a local stub, so the lesson stays vendor-neutral.

The outline below is a teaching plan, not a production incident report. Commands and modules are workshop starters you can rerun; they are not claims about live traffic, latency, or model rankings. Keep personal API keys off the projector, and keep prompt files free of customer data while you practice.

Audience and prerequisites

This session is for instructors and students who already have a tiny agent loop that returns JSON. It also fits platform engineers who change completion URLs more often than they change unit tests. You need Python 3.11+, pytest, and the ability to POST a prompt to an HTTP endpoint that returns a JSON object.

Skip the session if your agent streams unstructured prose with no schema, or if your only goal is a qualitative vibe check. The contract ledger will not grade reasoning quality, retrieval freshness, or safety policy. It only answers whether the reply still matches the shape your downstream code already assumed.

Eighty-minute agenda

Use a visible timer. When a block overruns, cut discussion rather than skipping the fail-closed test, because that test is the artifact students take home.

1. 0–8 min — Frame the miss. Show a passing test that only asserts tool == "triage" while the JSON body quietly changes type.
2. 8–20 min — Define the ledger. Agree on required keys, enums, ranges, and keys that must never appear.
3. 20–38 min — Exercise 1. Record one golden fixture and print its SHA-256.
4. 38–55 min — Exercise 2. Encode fail-closed rules and watch pytest fail on extra keys.
5. 55–72 min — Exercise 3. Point the same fixture at a second endpoint without editing assertions.
6. 72–80 min — Decision table and limitations. Classify warn versus fail, then list who should not ship this as a safety gate.

Failure mode: tests that only watch tool names

Many agent suites assert that a tool was selected, then treat the remaining payload as an untyped blob. That pattern survives a model swap even when confidence flips from a float to a string, or when priority gains a new label such as urgent. Downstream routers then throw at runtime, while CI stays green because the tool name never changed.

A second miss is leakage. Completions sometimes echo raw_prompt, chain-of-thought, or a key-shaped string that a student pasted into a notebook. Field allowlists catch that class of accident faster than a human reading a transcript. The workshop therefore freezes three layers at once: required types, closed enumerations, and forbidden keys.

Current public debate about agents outgrowing their tests is useful as motivation, not as a scoring rubric. You do not need a new leaderboard. You need a replayable object that fails when the JSON contract moves.

Artifact: frozen contract ledger

Create a directory named agent_contracts/ and keep fixtures next to the checker. The golden file is the source of truth; the checker is only an interpreter of that file. Students should rerun the same commands after any endpoint change, including a move from a laptop stub to a shared classroom URL.

Golden fixture

{
  "schema_version": "triage.v1",
  "category": "billing",
  "priority": "p1",
  "confidence": 0.64,
  "rationale_id": "r-0142"
}

Save that object as agent_contracts/golden/triage_v1.json. Do not pretty-print it differently across machines if you want stable hashes; the checker below canonicalizes keys before hashing.

Checker module

# agent_contracts/ledger.py
from __future__ import annotations

import hashlib
import json
from typing import Any

REQUIRED = {
    "category": str,
    "priority": str,
    "confidence": float,
    "rationale_id": str,
}
ALLOWED_PRIORITY = {"p0", "p1", "p2", "p3"}
FORBIDDEN_KEYS = {"raw_prompt", "api_key", "internal_cot", "tool_dump"}
OPTIONAL = {"notes", "schema_version"}


def canonical_dumps(payload: dict[str, Any]) -> str:
    return json.dumps(payload, sort_keys=True, separators=(",", ":"))


def fixture_hash(payload: dict[str, Any]) -> str:
    body = canonical_dumps(payload).encode("utf-8")
    return hashlib.sha256(body).hexdigest()


def check_contract(payload: Any) -> list[str]:
    errors: list[str] = []
    if not isinstance(payload, dict):
        return ["payload_not_object"]

    for key, expected in REQUIRED.items():
        if key not in payload:
            errors.append(f"missing:{key}")
            continue
        value = payload[key]
        if expected is float and isinstance(value, bool):
            errors.append(f"type:{key}:bool")
        elif expected is float and isinstance(value, int) and not isinstance(value, bool):
            value = float(value)
        elif not isinstance(value, expected):
            errors.append(f"type:{key}:{type(payload[key]).__name__}")

    allowed = set(REQUIRED) | OPTIONAL
    for key in sorted(set(payload) - allowed):
        errors.append(f"unexpected:{key}")
    for key in FORBIDDEN_KEYS:
        if key in payload:
            errors.append(f"forbidden:{key}")

    priority = payload.get("priority")
    if isinstance(priority, str) and priority not in ALLOWED_PRIORITY:
        errors.append(f"enum:priority:{priority}")

    confidence = payload.get("confidence")
    if isinstance(confidence, bool):
        errors.append("range:confidence")
    elif isinstance(confidence, (int, float)) and not (0.0 <= float(confidence) <= 1.0):
        errors.append("range:confidence")
    return errors

Pytest gate

# tests/test_triage_contract.py
import json
from pathlib import Path

from agent_contracts.ledger import check_contract, fixture_hash

GOLDEN = Path("agent_contracts/golden/triage_v1.json")


def test_golden_file_parses_and_passes_contract():
    payload = json.loads(GOLDEN.read_text(encoding="utf-8"))
    assert check_contract(payload) == []
    digest = fixture_hash(payload)
    assert len(digest) == 64


def test_string_confidence_is_a_contract_break():
    payload = json.loads(GOLDEN.read_text(encoding="utf-8"))
    payload["confidence"] = "0.64"
    assert "type:confidence:str" in check_contract(payload)


def test_forbidden_key_fails_closed():
    payload = json.loads(GOLDEN.read_text(encoding="utf-8"))
    payload["internal_cot"] = "..."
    assert "forbidden:internal_cot" in check_contract(payload)

Run the local gate before any remote call:

python -m pip install pytest
python -m pytest tests/test_triage_contract.py -q

Exercise 1 — Record the golden path (18 minutes)

Students call their current loop once, write the JSON to disk, and refuse to hand-edit fields that the model did not emit. If the model omitted rationale_id, that is a contract failure in the generator, not a license to invent an identifier in the fixture. Label any repaired example as synthetic in notes so later readers do not treat it as a live trace.

Proposed recording client (unexecuted until you fill ENDPOINT):

# scripts/record_fixture.py
from __future__ import annotations

import json
import os
import urllib.request
from pathlib import Path

from agent_contracts.ledger import check_contract, fixture_hash

ENDPOINT = os.environ["AGENT_COMPLETE_URL"]
PROMPT = {
    "task": "triage",
    "ticket": "Invoice #441 was charged twice after the March workspace upgrade.",
    "schema_version": "triage.v1",
}


def complete(prompt: dict) -> dict:
    req = urllib.request.Request(
        ENDPOINT,
        data=json.dumps(prompt).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read().decode("utf-8"))


def main() -> None:
    payload = complete(PROMPT)
    errors = check_contract(payload)
    Path("agent_contracts/last_errors.json").write_text(
        json.dumps(errors, indent=2), encoding="utf-8"
    )
    if errors:
        raise SystemExit(f"contract_failed:{errors}")
    out = Path("agent_contracts/golden/triage_v1.json")
    out.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    print(fixture_hash(payload))


if __name__ == "__main__":
    main()

export AGENT_COMPLETE_URL="http://127.0.0.1:8080/complete"
python scripts/record_fixture.py

Print the hash on the board. Students should treat a hash change without a schema_version bump as an incident, even when the test count stays the same.

Exercise 2 — Encode fail-closed rules (17 minutes)

Add one mutation per pair of students. Useful mutations include stringified numbers, an extra debug_trace key, a priority of urgent, and a confidence of 1.7. Each mutation must produce a stable error token such as enum:priority:urgent, not a traceback from application code.

If your loop wraps the model object in { "ok": true, "data": { ... } }, contract the inner object, not the envelope. Mixing transport metadata into the domain schema is how later retries look like product changes. Keep schema_version on the domain object so two envelopes can carry the same contract.

Exercise 3 — Swap the endpoint, keep the fixture (17 minutes)

Point AGENT_COMPLETE_URL at a second completion URL without editing pytest. The golden file stays read-only. If the second endpoint cannot satisfy triage.v1, the workshop result is a documented incompatibility, not a rewritten fixture. That is the entire point of freezing the contract before the swap.

Classroom groups often need a shared completion URL so nobody pastes personal keys into a projector demo. Disclosure: This article was prepared as part of MonkeyCode's product outreach. MonkeyCode is an open-source project that currently offers free model access and a free server option, which can host this replay runner for a lab section. Confirm live availability on the project page before you schedule the session, because free capacity is not a contractual SLA and must not be described with invented quotas or hardware claims.

A local stub remains the correct fallback when the shared box is busy:

# scripts/stub_server.py
from http.server import BaseHTTPRequestHandler, HTTPServer
import json

FIXTURE = {
    "schema_version": "triage.v1",
    "category": "billing",
    "priority": "p1",
    "confidence": 0.64,
    "rationale_id": "r-0142",
}

class Handler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        length = int(self.headers.get("Content-Length", "0"))
        _ = self.rfile.read(length)
        body = json.dumps(FIXTURE).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

if __name__ == "__main__":
    HTTPServer(("127.0.0.1", 8080), Handler).serve_forever()

python scripts/stub_server.py

Worked example: ticket triage loop

The domain rule for this example is narrow on purpose. A ticket about duplicate charges must come back as category=billing, priority in {p0,p1,p2,p3}, and confidence in [0,1]. rationale_id is an opaque string your notes system already knows; the model may not invent a new citation format mid-workshop.

Proposed loop (label as a teaching stub, not a production agent):

# agent_contracts/loop.py
from __future__ import annotations

import json
from typing import Any, Callable

from agent_contracts.ledger import check_contract

CompleteFn = Callable[[dict[str, Any]], dict[str, Any]]


def triage_once(ticket: str, complete: CompleteFn) -> dict[str, Any]:
    prompt = {
        "task": "triage",
        "ticket": ticket,
        "schema_version": "triage.v1",
        "fields": ["category", "priority", "confidence", "rationale_id"],
    }
    payload = complete(prompt)
    errors = check_contract(payload)
    if errors:
        raise ValueError("contract_failed:" + ",".join(errors))
    return payload

A replay test that students can keep after class:

# tests/test_triage_loop.py
import json
from pathlib import Path

from agent_contracts.loop import triage_once


def test_loop_rejects_drifted_endpoint():
    golden = json.loads(
        Path("agent_contracts/golden/triage_v1.json").read_text(encoding="utf-8")
    )

    def drifted(_prompt):
        broken = dict(golden)
        broken["priority"] = "urgent"
        return broken

    try:
        triage_once("Invoice charged twice", drifted)
        raise AssertionError("expected contract failure")
    except ValueError as err:
        assert "enum:priority:urgent" in str(err)

python -m pytest tests/test_triage_loop.py tests/test_triage_contract.py -q

If both files pass, students have a rerunnable cassette of shape, not of prose. They can later attach a real completer without rewriting the assertions. That is the workshop’s take-home artifact.

Decision table for contract severity

Bump schema_version when you intentionally widen the contract. Do not widen it to make a weaker model look compatible. Compatibility theater is how this workshop turns into theater.

Limitations

JSON contracts do not prove that billing was the right category, only that the field existed as a string. They also do not bound token spend, tool fan-out, or retry storms; those are different workshops with different meters. Free or shared endpoints can change behavior without notice, so a passing lab on Monday is not evidence for a capacity plan on Friday.

Canonical hashing is sensitive to key order only after canonicalization; it is still sensitive to semantically equal numbers encoded as 0.64 versus 0.640. If you need numeric tolerance, add an explicit rounding step before the hash, and document it in the fixture README. Do not hide rounding inside the completer.

Who should not run this workshop

• Teams that need a safety certifier for medical, legal, or financial decisions should not treat this ledger as that certifier.
• Groups without permission to send tickets to a shared server should stay on the local stub.
• Courses that only want a chat demo will find the fail-closed posture frustrating, and they should pick a different lab.
• Operators who cannot state a schema_version should fix naming first; the checker cannot invent a product contract.

Close-out checklist

Students leave with four files: the golden JSON, ledger.py, two pytest modules, and a one-row note of the endpoint URL they last used. The URL is configuration, not curriculum. If the next lab swaps endpoints, the curriculum stays still, and the contract either holds or it does not.