Module 4: Bedrock Guardrails And Defense In Depth
6. Hands-on: the output schema validator
Description
Lesson 4 demonstrated it with a concrete example: a response can pass Bedrock Guardrails' six policies with no objection and still not have the shape Shipments needs. This lesson builds the piece that closes that exact gap: guardrails/post_invoke_checks.py, a deterministic validator confirming, field by field, that a response — real or representative — matches ShipmentFields, the five-field contract parse_manifest() (Module 1, lesson 3) already produces for the deterministic path, before any code tries writing it to Shipments. It ran for real to write this lesson; the pytest "What to expect" block is literal output, sixteen fixed cases.
Connection to the module
Where lesson 5 protects extract-shipment-manifest-fields's input, this lesson protects its output — the same two-layer pattern this module's lesson 1 already previewed in its map. Neither depends on the other; neither depends on Bedrock existing.
Analogy: quality control at the end of the line, not at the front door
A serious logistics warehouse doesn't trust that merchandise coming in the door is correctly packed — it also has quality control at the end of the packing line, right before a box heads out to the truck: does the box have every item the shipping guide says it should have? Is something missing? Is there something extra that shouldn't be there? That control doesn't care whether the merchandise inside is dangerous or not — that was already checked at an earlier point; it cares exclusively about whether the box, as it stands, matches what the system expects to receive. post_invoke_checks.py is exactly that final quality control, applied to a Bedrock response instead of a physical box: it doesn't ask again "is this safe?" (the managed guardrail already evaluated that in lesson 3) — it asks, exclusively, "does this have the exact shape Shipments needs to receive?"
Step 1 — The complete script
guardrails/post_invoke_checks.py, at the root of andes-cargo-infra/:
#!/usr/bin/env python3
"""post_invoke_checks.py -- a deterministic, local schema validator that runs
AFTER extract-shipment-manifest-fields gets a response back from Bedrock (or,
in this $0 lab, a representative dict standing in for one -- see Module 4,
lesson 7), and BEFORE that response is ever written to Shipments.
This is defense in depth on the OUTPUT side (pre_invoke_checks.py is the input
side): Bedrock Guardrails' contextual grounding policy (bedrock.tf, Module 4,
lesson 3) can tell you a response is grounded in the manifest text -- it says
nothing about whether that response has the exact five fields Shipments
requires. A perfectly grounded, perfectly ungrounded-free response that is
missing weightKg is still not writable to Shipments. That gap is this script's
entire job (Module 4, lesson 4).
ShipmentFields is the name this guide gives to that contract: exactly the five
fields parse_manifest() already produces for the deterministic path (Module 1,
lesson 3) -- shipmentId, originCountry, destinationCountry, carrier, weightKg
-- all non-empty strings, because write_shipment_record() expects the LLM path
to hand it the same shape the deterministic path always has.
Never uses random or datetime.now(). Given the same candidate dict, this script
always produces the same PASS/FAIL result. Run the test suite with:
pytest guardrails/test_post_invoke_checks.py -v
"""
from __future__ import annotations
import argparse
import json
import sys
from dataclasses import dataclass, field
# The exact contract Shipments requires -- the same five keys parse_manifest()
# (Module 1, lesson 3) produces from a well-formed clave=valor manifest. The
# LLM path must match this shape exactly, or write_shipment_record() would
# receive a dict it was never built to handle.
SHIPMENT_FIELDS_SCHEMA = (
"shipmentId",
"originCountry",
"destinationCountry",
"carrier",
"weightKg",
)
@dataclass(frozen=True)
class ValidationResult:
candidate: dict
missing_fields: tuple[str, ...] = field(default_factory=tuple)
empty_fields: tuple[str, ...] = field(default_factory=tuple)
unexpected_fields: tuple[str, ...] = field(default_factory=tuple)
weight_not_numeric: bool = False
@property
def is_valid(self) -> bool:
return not (
self.missing_fields
or self.empty_fields
or self.unexpected_fields
or self.weight_not_numeric
)
@property
def errors(self) -> tuple[str, ...]:
errors: list[str] = []
if self.missing_fields:
errors.append(f"missing required field(s): {', '.join(self.missing_fields)}")
if self.empty_fields:
errors.append(f"empty value for field(s): {', '.join(self.empty_fields)}")
if self.unexpected_fields:
errors.append(f"unexpected field(s) not in ShipmentFields: {', '.join(self.unexpected_fields)}")
if self.weight_not_numeric:
errors.append("weightKg is not a numeric string")
return tuple(errors)
def _is_numeric_string(value: object) -> bool:
if not isinstance(value, str):
return False
try:
float(value)
except ValueError:
return False
return True
def validate_shipment_fields(candidate: dict) -> ValidationResult:
"""Validate that `candidate` matches ShipmentFields exactly: all five
required keys present, every value a non-empty string, weightKg parseable
as a number, and no extra keys the deterministic path never produces."""
required = set(SHIPMENT_FIELDS_SCHEMA)
present = set(candidate.keys())
missing = tuple(sorted(required - present))
unexpected = tuple(sorted(present - required))
empty = tuple(
k
for k in SHIPMENT_FIELDS_SCHEMA
if k in candidate and (not isinstance(candidate[k], str) or candidate[k].strip() == "")
)
weight_not_numeric = "weightKg" in candidate and "weightKg" not in empty and not _is_numeric_string(candidate["weightKg"])
return ValidationResult(
candidate=candidate,
missing_fields=missing,
empty_fields=empty,
unexpected_fields=unexpected,
weight_not_numeric=weight_not_numeric,
)
def format_report(result: ValidationResult) -> str:
status = "PASS" if result.is_valid else "FAIL"
lines = [f"post_invoke_checks: {status}"]
if result.is_valid:
lines.append(" All five ShipmentFields present, non-empty, weightKg numeric.")
else:
for err in result.errors:
lines.append(f" - {err}")
return "\n".join(lines)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Deterministic ShipmentFields schema check, before a candidate dict is written to Shipments."
)
parser.add_argument("--json", required=True, dest="json_text", help="candidate fields as a JSON object")
args = parser.parse_args(argv)
try:
candidate = json.loads(args.json_text)
except json.JSONDecodeError as exc:
print(f"error: --json is not valid JSON: {exc}", file=sys.stderr)
return 2
if not isinstance(candidate, dict):
print("error: --json must decode to a JSON object", file=sys.stderr)
return 2
result = validate_shipment_fields(candidate)
print(format_report(result))
return 0 if result.is_valid else 1
if __name__ == "__main__":
raise SystemExit(main())Four kinds of failure, each a distinct category of "this box doesn't match the shipping guide": missing fields (missing_fields), empty values (empty_fields — an empty string or one with only whitespace counts as if the field didn't exist), unexpected fields (unexpected_fields — something ShipmentFields never expected to receive), and a specific type case (weight_not_numeric — weightKg has to be interpretable as a number, even when stored as a string, exactly as parse_manifest() already delivers it). main()'s exit code does distinguish PASS from FAIL (0 or 1) — unlike lesson 5's pre_invoke_checks.py, this check does decide whether something continues: a response failing here should never reach Shipments.
Step 2 — Running the validator, for real
A representative candidate that does match ShipmentFields — Module 1, lesson 3's shipment 4471, now in the shape a successful extraction would produce:
python3 guardrails/post_invoke_checks.py --json '{"shipmentId": "4471", "originCountry": "Peru", "destinationCountry": "Chile", "carrier": "AndesExpress", "weightKg": "120"}'What to expect (literal — run for real, same environment as this guide; the input JSON is a hand-built representative candidate, never the output of a real invocation, see lesson 7):
post_invoke_checks: PASS
All five ShipmentFields present, non-empty, weightKg numeric.
Now, exactly lesson 4's Gap 2 case — the same candidate, with no weightKg:
python3 guardrails/post_invoke_checks.py --json '{"shipmentId": "4471", "originCountry": "Peru", "destinationCountry": "Chile", "carrier": "AndesExpress"}'What to expect (literal):
post_invoke_checks: FAIL
- missing required field(s): weightKg
Exit code 1 — this is, exactly, the candidate lesson 4 demonstrated all six of Bedrock Guardrails' policies would approve with no objection. post_invoke_checks.py is the only layer in this entire module rejecting it, and it does so with the same determinism as any other check in this guide: the same JSON, always the same result.
Step 3 — The pytest suite, sixteen fixed cases
guardrails/test_post_invoke_checks.py:
"""pytest suite for post_invoke_checks.py -- fixed, deterministic candidate
dicts only. No random, no datetime.now(): the same candidate must always
produce the same PASS/FAIL result, on any machine. Run with:
pytest guardrails/test_post_invoke_checks.py -v
Every candidate here is a REPRESENTATIVE dict -- never the output of a real
Bedrock invocation (Module 4, lesson 7 explains exactly why none was invoked).
These are hand-built fixtures shaped like what a real response would look
like, used only to exercise this script's own logic.
"""
import pytest
from post_invoke_checks import SHIPMENT_FIELDS_SCHEMA, format_report, main, validate_shipment_fields
VALID_SHIPMENT_4471 = {
"shipmentId": "4471",
"originCountry": "Peru",
"destinationCountry": "Chile",
"carrier": "AndesExpress",
"weightKg": "120",
}
def test_valid_candidate_passes():
result = validate_shipment_fields(VALID_SHIPMENT_4471)
assert result.is_valid is True
assert result.errors == ()
def test_missing_weight_kg_fails():
candidate = {k: v for k, v in VALID_SHIPMENT_4471.items() if k != "weightKg"}
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.missing_fields == ("weightKg",)
def test_missing_multiple_fields_reports_all_of_them():
candidate = {"shipmentId": "4472", "carrier": "AndesExpress"}
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.missing_fields == ("destinationCountry", "originCountry", "weightKg")
def test_empty_string_value_fails():
candidate = dict(VALID_SHIPMENT_4471, carrier="")
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.empty_fields == ("carrier",)
def test_whitespace_only_value_counts_as_empty():
candidate = dict(VALID_SHIPMENT_4471, originCountry=" ")
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.empty_fields == ("originCountry",)
def test_non_numeric_weight_fails():
candidate = dict(VALID_SHIPMENT_4471, weightKg="approximately 120kg")
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.weight_not_numeric is True
def test_numeric_weight_as_plain_number_string_passes():
candidate = dict(VALID_SHIPMENT_4471, weightKg="45.5")
result = validate_shipment_fields(candidate)
assert result.is_valid is True
def test_unexpected_extra_field_fails():
"""A representative case of exactly what Module 4, lesson 4 warns about:
a response that IS grounded and DOES have the five required fields, but
also invents a sixth field (e.g. a hallucinated confidence score in the
wrong shape) that Shipments was never built to receive."""
candidate = dict(VALID_SHIPMENT_4471, confidenceScore="very high")
result = validate_shipment_fields(candidate)
assert result.is_valid is False
assert result.unexpected_fields == ("confidenceScore",)
def test_empty_dict_reports_all_five_missing():
result = validate_shipment_fields({})
assert result.is_valid is False
assert set(result.missing_fields) == set(SHIPMENT_FIELDS_SCHEMA)
def test_schema_has_exactly_five_fields():
assert len(SHIPMENT_FIELDS_SCHEMA) == 5
def test_report_format_for_pass():
result = validate_shipment_fields(VALID_SHIPMENT_4471)
report = format_report(result)
assert "PASS" in report
def test_report_format_for_fail_lists_each_error():
candidate = {"shipmentId": "4471"}
result = validate_shipment_fields(candidate)
report = format_report(result)
assert "FAIL" in report
assert "missing required field(s)" in report
def test_cli_end_to_end_pass(capsys):
exit_code = main(["--json", '{"shipmentId": "4471", "originCountry": "Peru", "destinationCountry": "Chile", "carrier": "AndesExpress", "weightKg": "120"}'])
captured = capsys.readouterr()
assert exit_code == 0
assert "PASS" in captured.out
def test_cli_end_to_end_fail_missing_field(capsys):
exit_code = main(["--json", '{"shipmentId": "4471", "originCountry": "Peru", "destinationCountry": "Chile", "carrier": "AndesExpress"}'])
captured = capsys.readouterr()
assert exit_code == 1
assert "FAIL" in captured.out
def test_cli_rejects_invalid_json(capsys):
exit_code = main(["--json", "{not valid json"])
captured = capsys.readouterr()
assert exit_code == 2
assert "error:" in captured.err
def test_cli_rejects_non_object_json(capsys):
exit_code = main(["--json", "[1, 2, 3]"])
captured = capsys.readouterr()
assert exit_code == 2
assert "error:" in captured.errSixteen cases: the valid candidate, a missing field, multiple missing fields, an empty value, a whitespace-only value, a non-numeric weightKg, a valid numeric weightKg, an unexpected field (lesson 4's exact Gap 2), a completely empty dictionary, the schema's own shape, the report's format in both states, the command-line interface's complete flow in both cases, and rejecting invalid JSON or JSON that doesn't decode to an object.
cd guardrails/
pytest test_post_invoke_checks.py -v -p no:randomlyWhat to expect (literal — run for real):
============================= test session starts ==============================
platform darwin -- Python 3.14.0, pytest-9.1.1, pluggy-1.6.0
rootdir: guardrails/
collected 16 items
test_post_invoke_checks.py::test_valid_candidate_passes PASSED [ 6%]
test_post_invoke_checks.py::test_missing_weight_kg_fails PASSED [ 12%]
test_post_invoke_checks.py::test_missing_multiple_fields_reports_all_of_them PASSED [ 18%]
test_post_invoke_checks.py::test_empty_string_value_fails PASSED [ 25%]
test_post_invoke_checks.py::test_whitespace_only_value_counts_as_empty PASSED [ 31%]
test_post_invoke_checks.py::test_non_numeric_weight_fails PASSED [ 37%]
test_post_invoke_checks.py::test_numeric_weight_as_plain_number_string_passes PASSED [ 43%]
test_post_invoke_checks.py::test_unexpected_extra_field_fails PASSED [ 50%]
test_post_invoke_checks.py::test_empty_dict_reports_all_five_missing PASSED [ 56%]
test_post_invoke_checks.py::test_schema_has_exactly_five_fields PASSED [ 62%]
test_post_invoke_checks.py::test_report_format_for_pass PASSED [ 68%]
test_post_invoke_checks.py::test_report_format_for_fail_lists_each_error PASSED [ 75%]
test_post_invoke_checks.py::test_cli_end_to_end_pass PASSED [ 81%]
test_post_invoke_checks.py::test_cli_end_to_end_fail_missing_field PASSED [ 87%]
test_post_invoke_checks.py::test_cli_rejects_invalid_json PASSED [ 93%]
test_post_invoke_checks.py::test_cli_rejects_non_object_json PASSED [100%]
============================== 16 passed in 0.02s ==============================
Sixteen out of sixteen — including test_unexpected_extra_field_fails, the case making explicit, in runnable code, exactly the same argument lesson 4 developed in prose: a response with an added confidenceScore, a field neither parse_manifest() nor Shipments ever expected, fails this check even though it would have passed all six of Bedrock Guardrails' policies with no problem.
Common mistakes
Accepting weightKg as a Python number (int/float) instead of a string, and being surprised when _is_numeric_string() rejects it (wrong-type assumption mistake). What happens: someone builds a candidate with "weightKg": 120 (no quotes, a Python/JSON integer) instead of "weightKg": "120" (a string). How to spot it: _is_numeric_string() starts with if not isinstance(value, str): return False — a real integer never passes that first check, so the candidate gets marked weight_not_numeric. How to fix it: ShipmentFields, as this guide defines it, requires all five fields to be strings — the same type parse_manifest() produces for all its values, including weight, in Module 1, lesson 3 — if your representative extraction produces a weightKg as a native JSON number, convert it to a string before passing it to this validator, exactly as extract-shipment-manifest-fields's real code would need to.
Confusing a field with a null/None value with a missing field (not distinguishing "absent" from "present but empty" mistake). What happens: someone builds a candidate with "carrier": null in JSON, expecting it to be reported as missing_fields, and is surprised to see it show up instead in a different category or pass unmarked. How to spot it: check the empty_fields logic in Step 1 — the condition is not isinstance(candidate[k], str) or candidate[k].strip() == ""; a None value (which in Python isn't a string) falls into the first part of that condition (not isinstance(...)), so it does get marked as empty_fields, not missing_fields. How to fix it: this distinction matters for the error message, not just academic precision — missing_fields means "the key was never in the dictionary," empty_fields means "the key is there, but its value is useless" (empty, whitespace-only, or a non-string type) — two different root causes an engineer debugging a failed extraction needs to distinguish.
Writing a new test that only checks is_valid, without checking errors' specific content or the failing fields (too-weak-an-assertion mistake). What happens: someone adds a new test case with assert result.is_valid is False, without also confirming which specific field caused the failure. How to spot it: if your test would pass just as "green" for any kind of failure — a missing field, an empty one, an unexpected one — without distinguishing which. How to fix it: follow Step 3's sixteen-case pattern — every one expecting a failure verifies, besides is_valid is False, the exact field populated in missing_fields/empty_fields/unexpected_fields/weight_not_numeric. A test that only confirms "it failed, somehow" doesn't protect against the day the internal logic changes and the check starts failing for the wrong reason, silently.
Exercises
Exercise 1 — Build, yourself, a JSON candidate that fails for TWO different reasons at once (for example, a missing field and another with an empty value), and predict what format_report() would show. Verify your prediction by running the script.
See solution
For example: {"shipmentId": "4475", "originCountry": "", "carrier": "AndesExpress", "weightKg": "60"} (missing destinationCountry, and originCountry is empty). format_report() would show two error lines, one for each populated category: - missing required field(s): destinationCountry and - empty value for field(s): originCountry — errors's design as a tuple walking all four possible categories, adding a line for each one with content, is precisely what lets it report several problems at once, instead of stopping at the first one it finds.
Exercise 2 — Explain why this validator rejects an UNEXPECTED field (unexpected_fields), instead of simply ignoring it and accepting the rest of the fields as valid. What real risk does this decision avoid, beyond "the extra data doesn't make sense"?
See solution
Silently ignoring extra fields would be a reasonable decision in some contexts, but not this one: write_shipment_record() (inherited from Module 1, lesson 3) expects a dictionary with a specific shape, and an unexpected additional field — for example, a confidenceScore the model decided to add on its own — could, depending on how that inherited code is written, cause unanticipated behavior if that code someday changed to use **candidate less carefully, or if an extra field accidentally propagated into a log or another table. Explicitly rejecting it, instead of silently ignoring it, forces any new field the model starts producing to be a conscious decision to update SHIPMENT_FIELDS_SCHEMA, not an accident slipping through unnoticed — the same "never guess, never silently accept" philosophy bedrock_cost_estimate.py (Module 2, lesson 7) already applied when rejecting a model with no known price.
Exercise 3 — Predict what would happen if, in Module 5, someone tried using this same script to validate a candidate with a SIXTH field Andes Cargo decided to add in the future (for example, packageCount). Would you need to change post_invoke_checks.py, SHIPMENT_FIELDS_SCHEMA, or both?
See solution
Only SHIPMENT_FIELDS_SCHEMA would need to change — adding "packageCount" to the tuple — no other line of post_invoke_checks.py would need to be touched. validate_shipment_fields()'s entire logic is written in terms of SHIPMENT_FIELDS_SCHEMA as a single source of truth (required = set(SHIPMENT_FIELDS_SCHEMA)), never with the five field names written directly inside the function — the same design principle MODEL_PRICING_USD_PER_MILLION_TOKENS already applied in Module 2, lesson 7: a central table or schema, referenced by the code, instead of values repeated in multiple places someone would have to remember to keep in sync.
Summary and next step
This lesson built post_invoke_checks.py, the deterministic validator closing lesson 4's exact Gap 2: you confirmed, with that lesson's same hypothetical candidate (missing weightKg), that this script does reject it — FAIL, exit code 1 — where all six of Bedrock Guardrails' policies would find nothing to object to. You ran the complete sixteen-case suite with pytest, including the explicit unexpected-field case, and confirmed that SHIPMENT_FIELDS_SCHEMA, as a single source of truth, is the only thing that would change if Andes Cargo's contract ever grew.
Before moving on you should be able to: explain the difference between missing_fields and empty_fields, with an example of each; build a candidate that fails for two simultaneous reasons and predict the exact report; and explain why an unexpected field gets rejected instead of silently ignored.
With both halves of the custom defense in depth built — input (lesson 5) and output (this lesson) —, lesson 7 marks, with this entire ecosystem's same honesty, the exact limit no custom check and no terraform plan can cross: Bedrock Guardrails' real blocking of a genuine attack attempt, something only a real invocation — outside this $0 lab's scope — could confirm.
Resources
- Python Docs —
dataclasses— reference forValidationResult, the type structuring this validator's result. - Python Docs —
json— reference for handlingjson.JSONDecodeErrorin the command-line interface. - This module, lesson 4 (
04-why-a-managed-guardrail-is-not-enough-alone.md) — Gap 2, the direct source for thetest_unexpected_extra_field_failstest case and the missing-weightKgexample run in Step 2. - This guide's Module 1, lesson 3 (
03-andes-cargos-ai-workload-when-the-deterministic-parser-is-not-enough.md) — the source forparse_manifest()and the exact five fieldsSHIPMENT_FIELDS_SCHEMAencodes.