Skip to content

C3 · lesson 6 · 10 min read

Validation and retry loops

Check every structured answer against stop_reason, schema and business rules, feed specific errors back for a bounded retry, and escalate when retries run out.

Ընտրեք տեքստը՝ հայերեն թարգմանությունը տեսնելու համար։Highlight text to see Armenian. Machine-assisted translation; English is the source of truth.

Structured outputs guarantee syntax: the JSON parses and matches the schema. They don’t guarantee the answer is right, and there are cases where even the syntax guarantee doesn’t hold. Anything that feeds automation needs a validation step and a plan for what happens when validation fails.

What can still go wrong

Failure How you see it What to do
Output cut off stop_reason: "max_tokens", JSON incomplete Retry with a higher max_tokens
Safety classifier declined stop_reason: "refusal", may not match schema Don’t parse it; check stop_details.category
Rule the schema can’t express Parses fine, but e.g. end time before start Validate in code, send the error back
Value outside a range minimum/maxLength aren’t supported Validate in code
Enum casing differs ("Page" vs "page") Normal stop reason Compare case-insensitively
Prompt-only JSON (no schema) Parse error, missing field, extra prose Parse, validate, retry, or switch to a schema

A refusal deserves its own handling. With the reasoning_extraction category, the prompt asked the model to print its reasoning (for example a reasoning field in the schema), and the fix is to change the prompt, not to retry. Ask for a short explanation field instead.

The loop

call Claudestop_reason?validateuse itappend errors, retry (attempt < N)escalateinvalidok
Check the stop reason, validate, retry with the errors, stop at a cap.

The key idea is that a retry is not the same request sent again. You append Claude’s previous answer and a user message that lists exactly what was wrong. Claude fixes the specific problem instead of rolling the dice again.

import os
import sys
from datetime import datetime

import anthropic
from pydantic import BaseModel, ValidationError, model_validator

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
MAX_ATTEMPTS = 3

SCHEMA = {
    "type": "object",
    "properties": {
        "service": {"type": "string"},
        "environment": {"type": "string", "enum": ["dev", "staging", "prod"]},
        "start": {"type": "string", "format": "date-time"},
        "end": {"type": "string", "format": "date-time"},
        "rollback_plan": {"type": "string"},
    },
    "required": ["service", "environment", "start", "end", "rollback_plan"],
    "additionalProperties": False,
}


class ChangeWindow(BaseModel):
    service: str
    environment: str
    start: datetime
    end: datetime
    rollback_plan: str

    @model_validator(mode="after")
    def check_rules(self):
        if self.end <= self.start:
            raise ValueError("end must be after start")
        if (self.end - self.start).total_seconds() > 4 * 3600:
            raise ValueError("change windows may not exceed 4 hours")
        if self.environment == "prod" and len(self.rollback_plan) < 20:
            raise ValueError("prod changes need a concrete rollback_plan")
        return self


messages = [{"role": "user", "content": f"<change_request>\n{sys.stdin.read()}\n</change_request>\n\nExtract the change window."}]
errors = "no attempt made"
for attempt in range(1, MAX_ATTEMPTS + 1):
    response = client.messages.create(
        model="claude-haiku-5-5",
        max_tokens=2048,
        system="You extract change windows from change requests. Use UTC ISO 8601 timestamps.",
        messages=messages,
        output_config={"format": {"type": "json_schema", "schema": SCHEMA}},
    )
    text = "".join(b.text for b in response.content if b.type == "text")
    if response.stop_reason == "refusal":
        details = getattr(response, "stop_details", None)
        sys.exit(f"Refused ({details}); fix the prompt, don't retry.")
    if response.stop_reason != "end_turn":
        errors = f"stop_reason={response.stop_reason}"
    else:
        try:
            print(ChangeWindow.model_validate_json(text).model_dump_json(indent=2))
            sys.exit(0)
        except ValidationError as exc:
            errors = "; ".join(e["msg"] for e in exc.errors())
    print(f"attempt {attempt} failed: {errors}", file=sys.stderr)
    messages += [
        {"role": "assistant", "content": text or "(no output)"},
        {"role": "user", "content": f"That record failed validation: {errors}. "
                                    "Re-read the change request and return a corrected record."},
    ]

sys.exit(f"Giving up after {MAX_ATTEMPTS} attempts; last errors: {errors}. Route to a human.")

Design rules that make the loop safe:

  • Cap the attempts. Two or three retries is plenty. A loop without a cap is an outage waiting for a pathological input.
  • Make errors specific. “Invalid” teaches nothing. “end must be after start” tells Claude what to fix.
  • Escalate, don’t guess. When the cap is hit, exit non-zero and send the input to a person or a dead-letter queue. Never silently pass a half-valid record on.
  • Log each failure (attempt number, stop reason, error text), but not secrets or whole payloads.

The same idea for tools

In a tool loop, your code is the validator. When a tool call has bad arguments or the tool fails, return a tool_result with "is_error": true and an instructive message such as "Namespace 'paymnts' not found. Known namespaces: payments, billing." Claude reads it and usually retries with corrected input. The docs say it retries two to three times before giving up and telling the user. With strict: true the schema part of that problem disappears, but rules like “this namespace must exist” still need your check.

Recap

  • Schemas guarantee shape, not truth. Check stop_reason first, then business rules in code.
  • max_tokens means truncated output: retry larger. A refusal is not a format error, so don’t just retry it.
  • Retries append the previous answer plus specific errors; cap them and escalate at the cap.
  • For tools, report problems with is_error: true and a message that says what to try next.

See it in action

Validate, then retry with the errors
validinvaliderrors sent backbudget spentrequestschema + rulesClaudevalidateJSON Schemause the resultretry with errorsstop after N tries

1/5 Ask for structured output: give the schema (or a strict tool definition) and say what each field means.

all steps
  1. Ask for structured output: give the schema (or a strict tool definition) and say what each field means.
  2. Never trust the reply blindly. Parse it and validate it against the schema in your own code.
  3. Valid? Use it.
  4. Invalid? Send the validation errors back to Claude and ask for a corrected reply. Specific errors fix things faster than a bare "try again".
  5. Cap the retries. After a few failures, stop and surface the error instead of looping forever.
structured output with validation and retry · illustrative session
transcript
$ python triage_json.py incident-4711.log
attempt 1  → {"severity": "urgent", "service": "checkout", ...}
invalid: severity 'urgent' is not one of ['sev1','sev2','sev3']
# the validation error goes back to Claude in the next turn
attempt 2  → {"severity": "sev2", "service": "checkout", ...}
valid ✓  written to incident-4711.json

$ ./try-it.sh

Try it: a validated change-window extractor with bounded retries

  1. Install anthropic and pydantic, export ANTHROPIC_API_KEY, and save the script from this lesson as window.py.
  2. Run it on a change request: python window.py < change.txt. Confirm it prints a JSON record or exits non-zero with the last errors.
  3. Write a change request whose end time is before its start time and watch the business-rule error go back to Claude on the retry.
  4. Set max_tokens=40 and check that the loop treats stop_reason: max_tokens as a failure instead of parsing half a record.

Check yourself

4 questions. Original practice questions, not taken from the real exam.

  1. A structured-outputs call that extracts a large inventory record sometimes returns JSON that doesn't parse. Those responses have stop_reason "max_tokens".

    1. What is the right handling?

  2. Extracted change windows sometimes have an end time before the start time. Your retry code catches the validation error and resends exactly the same request, up to ten times.

    2. What is the best improvement?

  3. Your pipeline uses structured outputs with a strict schema for maintenance windows before it writes them to the change calendar.

    3. Which two checks does your code still need? (Choose two.)

    Choose 2.

  4. In a Kubernetes agent loop, Claude calls get_pods with namespace "paymnts". Your tool code finds no such namespace.

    4. What should your code send back?

Answer every question to check.