Stabilizing Claude API Structured Responses in Production — Notes on tool_use, JSON Schema, and Layered Validation

Write JSON Schema "Just Strict Enough"

The stability of your structured responses is mostly decided by how you write the JSON Schema. The first instinct is to make the schema maximally strict, but in my experience overly tight constraints actively raise the failure rate. The shape I use in production is what I call "just strict enough."

Concretely, four guidelines:

First, always include an escape hatch in your enums. Adding "other" or "unspecified" gives the model somewhere to go when it cannot confidently fit an existing category. Routing "other" to a manual-review bucket downstream keeps the pipeline stable rather than forcing dishonest classifications.

Second, set minLength and maxLength on every string field. This rejects both empty strings and the rare case where the model decides to be helpful and returns "primary_category": "nature (broad scenery including mountains and lakes)". A maxLength: 32 is a cheap insurance policy.

Third, set additionalProperties: false in production. Without it, the model occasionally adds "helpful" extra fields that crash downstream code.

Fourth, bound numeric fields with minimum and maximum. Probabilities are 0..1, ratings are 1..5. The tighter the range, the less spread you see in actual outputs.

# Before — too strict, fragile
"input_schema": {
    "type": "object",
    "properties": {
        "primary_category": {
            "type": "string",
            "enum": ["nature", "abstract", "city", "minimal", "people"],  # no escape hatch
        },
        "color_palette": {
            "type": "array",
            "items": {"type": "string", "pattern": "^#[0-9A-Fa-f]{6}$"},  # heavy regex
            "minItems": 5,
            "maxItems": 5,  # exactly five colors required
        },
    },
    "required": ["primary_category", "color_palette"],
}

# After — strict but not breakable
"input_schema": {
    "type": "object",
    "properties": {
        "primary_category": {
            "type": "string",
            "enum": ["nature", "abstract", "city", "minimal", "people", "other"],
        },
        "color_palette": {
            "type": "array",
            "items": {"type": "string", "minLength": 4, "maxLength": 7},
            "minItems": 1,
            "maxItems": 5,
        },
    },
    "required": ["primary_category", "color_palette"],
    "additionalProperties": False,
}

The "before" schema insisted on exactly five colors with strict hex format. Some images simply do not have five distinct colors, and occasionally the model emitted a near-miss like #ff that the regex rejected. We were paying for API retries on what was essentially our own pickiness. After loosening, Anthropic-side rejections fell to a couple per month.

The natural worry is: "if I loosen the schema, won't I let broken data through?" The answer is yes — and that is exactly what the next layer absorbs.

Three layers: SDK types, JSON Schema, domain rules

The SDK's types are not as load-bearing as they look. block.input is essentially dict[str, Any] at runtime; the API-side JSON Schema check is what actually enforces the shape. Even then, the schema does not know your business rules — for instance, "if needs_human_review=true, then confidence>0.9 is contradictory."

So I build validation as three layers, each with its own failure strategy.

L3 lives in your application code, not in the model. Retrying domain failures wastes calls because the model isn't going to suddenly become coherent. Instead, I treat them as "task complete, escalate to human review."

from dataclasses import dataclass
 
@dataclass
class ClassificationResult:
    primary_category: str
    tags: list[str]
    confidence: float
    needs_human_review: bool
    domain_violations: list[str]
 
 
def validate_domain(raw: dict) -> ClassificationResult:
    violations: list[str] = []
 
    # Rule 1: high confidence but flagged for review is contradictory
    if raw.get("confidence", 0) > 0.92 and raw.get("needs_human_review"):
        violations.append("high_confidence_but_review_requested")
 
    # Rule 2: "other" is always promoted to human review
    if raw.get("primary_category") == "other" and not raw.get("needs_human_review"):
        raw["needs_human_review"] = True
        violations.append("auto_promoted_other_to_review")
 
    # Rule 3: trim tags and drop empties; insert a fallback if everything is gone
    cleaned_tags = [t.strip() for t in raw.get("tags", []) if t and t.strip()]
    if not cleaned_tags:
        violations.append("empty_tags_after_cleaning")
        cleaned_tags = ["uncategorized"]
 
    return ClassificationResult(
        primary_category=raw.get("primary_category", "other"),
        tags=cleaned_tags,
        confidence=float(raw.get("confidence", 0.0)),
        needs_human_review=bool(raw.get("needs_human_review", False)),
        domain_violations=violations,
    )

I log the domain_violations list. That list becomes the cleanest first-pass signal for "the model's behavior has drifted." If high_confidence_but_review_requested spikes in a given week, that is the first thing I check when a model version changes. Comparing violation counts before and after a Claude release is the cheapest regression test I have.

Streaming and partial JSON: where to be paranoid

With stream=True, a tool_use block arrives as content_block_start, then a sequence of input_json_delta chunks, then content_block_stop. The temptation is to parse the partial JSON as it streams in so the UI can update incrementally. Two traps lie there.

First, partial JSON ({"primary_category": "natu) is not valid input to a standard json.loads. Second, you need a clear answer to "what if the socket dies before content_block_stop?"

The compromise I have settled on:

• The UI shows a "thinking" skeleton during partial. Only the finalized result is rendered.
• I do extract one progress hint from partial: by structuring the prompt to put primary_category first, that key tends to resolve early. The moment it stabilizes, I tint the skeleton with a category color so it feels alive.
• If the connection drops before content_block_stop, I log the partial state and throw, requeuing the task.

Practically, I pair a tolerant parser like partial-json with one hard rule: only tool_use.input from a completed content_block_stop event is allowed to flow into business logic. The partial parse is purely cosmetic.

// TypeScript SDK — partial JSON for UI hints only
import Anthropic from "@anthropic-ai/sdk";
import { parse as parsePartial } from "partial-json";
 
const client = new Anthropic();
 
async function classifyWithProgressHint(imageInput: any) {
  let accumulatedJson = "";
  let preview: { primary_category?: string } = {};
  let finalInput: Record<string, unknown> | null = null;
 
  const stream = client.messages.stream({
    model: "claude-haiku-4-5",
    max_tokens: 512,
    tools: [TOOL_SCHEMA],
    tool_choice: { type: "tool", name: "classify_wallpaper" },
    messages: [{ role: "user", content: [imageInput, { type: "text", text: "classify" }] }],
  });
 
  for await (const event of stream) {
    if (event.type === "content_block_delta" && event.delta.type === "input_json_delta") {
      accumulatedJson += event.delta.partial_json;
      try {
        preview = parsePartial(accumulatedJson);
        if (preview.primary_category) {
          notifyUiHint(preview.primary_category);
        }
      } catch {
        // ignore partial parse errors
      }
    }
    if (event.type === "content_block_stop" && event.content_block?.type === "tool_use") {
      finalInput = event.content_block.input as Record<string, unknown>;
    }
  }
 
  if (!finalInput) {
    throw new Error("tool_use did not complete");
  }
  return finalInput;
}

The wider tension is between "give the user feedback faster" and "do not lie about data we have not received." Splitting the responsibility — partial drives animation, finalized drives logic — is what stayed the most stable in production.

Classify failures, retry only where retrying helps

Not every failure deserves the same treatment. I bucket structured-response failures into four kinds, each with its own policy.

Looking at my wallpaper logs, second attempts on schema-violation failures succeed about 14% of the time. A third attempt adds only another 4 points or so. Cutting at attempt two saves both money and elapsed time, and the trapped logs become the best raw material for prompt tuning.

import time
from anthropic import APIStatusError, APITimeoutError
 
def classify_with_observability(image_block: dict, max_retries: int = 3) -> ClassificationResult:
    last_error: Exception | None = None
    for attempt in range(1, max_retries + 1):
        try:
            raw = classify_wallpaper_tool(image_block)
            result = validate_domain(raw)
            log_metric("classification.success", {"attempt": attempt})
            return result
        except APITimeoutError as e:
            last_error = e
            log_metric("classification.timeout", {"attempt": attempt})
            time.sleep(2 ** attempt)
        except APIStatusError as e:
            last_error = e
            log_metric("classification.api_status", {"status": e.status_code, "attempt": attempt})
            if e.status_code == 400 and attempt >= 2:
                raise
            time.sleep(2 ** attempt)
 
    raise RuntimeError(f"classification failed after {max_retries} attempts: {last_error}")

Logging metrics per attempt is what makes "retries are paying for themselves" visible in a dashboard. Without it, retries quietly cost money while solving nothing. I push these to OpenTelemetry and Grafana, but Cloud Run log-based metrics are perfectly fine for getting started.

Extended thinking, when and when not

For schemas with deep nesting or oneOf branches, pairing extended thinking with tool_use lifts the "first-pass valid response" rate noticeably. In the schemas I tested, "deep" being more than five nested levels, that lift was about 22 percentage points.

Two caveats. First, with tool_choice="auto" and extended thinking enabled, the model occasionally finishes thinking and answers without calling the tool. I force tool_choice={"type": "tool", "name": "..."} to keep the tool in the loop. Second, latency rises by roughly 1.5–2x. That makes extended thinking unsuitable for interactive UIs where the user is waiting; it is much more comfortable in batch or asynchronous flows.

I skip extended thinking on simple wallpaper tagging. I do enable it for the multi-step App Store review reply pipeline, where the model has to pick a category, choose a tone, and draft the message. There, the rate of needs_human_review=true dropped from about 8.4% per month without extended thinking to 2.1% with it. That is not the kind of gap simple retries can close — it comes from the model getting more space to reason about the choice.

A practical template for a one-person scale

The closing artifact is the one file that wraps all of the above. This is close to what I use in both the wallpaper pipeline and the App Store review reply pipeline.

"""structured_call.py — Shared layer for stable structured Claude API responses."""
from __future__ import annotations
import time
import logging
from dataclasses import dataclass
from typing import Any, Callable, TypeVar
 
from anthropic import Anthropic, APIStatusError, APITimeoutError
 
T = TypeVar("T")
logger = logging.getLogger(__name__)
client = Anthropic()
 
 
@dataclass
class StructuredResult[T]:
    value: T
    domain_violations: list[str]
    attempt_count: int
 
 
def call_with_tool(
    *,
    model: str,
    tool: dict,
    messages: list[dict],
    domain_validator: Callable[[dict], tuple[T, list[str]]],
    max_retries: int = 3,
    extended_thinking: bool = False,
) -> StructuredResult[T]:
    """Run a tool_use call and pipe the result through domain validation."""
    create_kwargs: dict[str, Any] = {
        "model": model,
        "max_tokens": 1024,
        "tools": [tool],
        "tool_choice": {"type": "tool", "name": tool["name"]},
        "messages": messages,
    }
    if extended_thinking:
        create_kwargs["thinking"] = {"type": "enabled", "budget_tokens": 4096}
 
    last_error: Exception | None = None
    for attempt in range(1, max_retries + 1):
        try:
            resp = client.messages.create(**create_kwargs)
            for block in resp.content:
                if block.type == "tool_use" and block.name == tool["name"]:
                    value, violations = domain_validator(dict(block.input))
                    return StructuredResult(value=value, domain_violations=violations, attempt_count=attempt)
            raise RuntimeError("tool_use block not found")
 
        except APITimeoutError as e:
            last_error = e
            logger.warning("timeout attempt=%s err=%s", attempt, e)
            time.sleep(min(2 ** attempt, 8))
 
        except APIStatusError as e:
            last_error = e
            logger.warning("api_status attempt=%s status=%s", attempt, e.status_code)
            if e.status_code == 400 and attempt >= 2:
                break
            time.sleep(min(2 ** attempt, 8))
 
    raise RuntimeError(f"structured call failed: {last_error}")

I run image classification, text summarization, and review-reply template selection through this same call_with_tool. The payoff of consolidation is that adding a domain rule propagates to every feature, and retry policy lives in exactly one place.

Twelve years of shipping apps solo has taught me that LLM integrations have two phases — the fun phase, and the operational phase. The first few weeks are all wonder, but a month into production the edge cases pile up. The three-layer design (tool_use, JSON Schema, domain rules) is how I keep that pile from wearing me down: it doesn't pretend the edges aren't there, but it doesn't let them drain me either.

If you are stabilizing structured Claude API responses in your own production stack, I hope some of this is useful. Thank you for reading.