Claude doesn’t ship a JSON mode or structured outputs endpoint, yet most production pipelines need machine-readable responses. Getting reliable JSON from Claude is a matter of controlling the prompt, forcing the start of the response, and validating strictly. The following steps show a pattern that holds up under load.
Step 1: Lock the schema in the system prompt
Vague instructions produce vague output. Write a system prompt that specifies exact keys, types, and constraints, then anchor it with a single minimal example. The example should match the schema exactly and avoid any commentary.
SYSTEM_PROMPT = """You are a data extraction engine.
Respond with exactly one JSON object matching this schema:
{
"title": string,
"score": float between 0 and 1,
"tags": array of strings
}
Do not include markdown fences, explanations, or trailing text.
Example:
{"title": "Local runoff", "score": 0.82, "tags": ["water", "quality"]}
"""If your real schema is large, do not paste the entire thing into the example. Show a compact subset that demonstrates nesting and array handling. Keep the system prompt static across calls so it can be cached later.
A common mistake is asking for “JSON if possible”. That invites prose. State the contract: “Respond with a single JSON object, no markdown, no commentary.” The model follows negative constraints better when they are explicit.
Step 2: Force the opening brace with an assistant prefill
Anthropic’s Messages API lets you seed the assistant turn by passing an assistant message containing a single {. This eliminates the “Sure! Here is your JSON:” preamble that breaks parsers. The model continues from that character, so your extraction code prepends the brace to the returned text.
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=SYSTEM_PROMPT,
messages=[
{"role": "user", "content": "Extract from: The creek flooded after the storm, likely due to poor drainage."},
{"role": "assistant", "content": "{"}
],
)
raw = "{" + response.content[0].textThe prefill must be exactly the opening brace, not a partial key. If you prefill {"title": the model may close the object early and append text. Keep it to { and let the schema drive the rest.
If you call Claude through an OpenAI-compatible gateway such as n4n.ai, the same prefill technique works via the assistant role in the chat sequence, and cache-control hints you attach are forwarded to the provider.
Step 3: Defensively extract and parse
Never trust the raw string. Even with prefill, the model can emit a trailing comma, a missing brace on token truncation, or stray whitespace. Use a parser that locates the first { and last } and attempts json.loads. For incremental safety, use json.JSONDecoder.raw_decode to parse the first complete object and ignore trailing garbage.
import json
import re
def extract_json(text: str) -> dict:
text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text.strip(), flags=re.MULTILINE)
start = text.find("{")
end = text.rfind("}")
if start == -1 or end == -1:
raise ValueError("No JSON object found")
candidate = text[start:end+1]
decoder = json.JSONDecoder()
obj, _ = decoder.raw_decode(candidate)
return obj
try:
data = extract_json(raw)
except (json.JSONDecodeError, ValueError) as e:
print("Parse failed:", e)When max_tokens is too low, the response truncates mid-object. Detect this by catching JSONDecodeError and checking whether the candidate ends with }; if not, retry with a higher limit or a continuation prompt that supplies the missing closing structure.
Step 4: Validate against a schema
Prompting is not a type system. Use pydantic to enforce types, ranges, and required fields. This catches the score returned as "0.82" or a missing tags array.
from pydantic import BaseModel, Field, conlist, confloat
class Extracted(BaseModel):
title: str
score: confloat(ge=0.0, le=1.0)
tags: conlist(str, min_length=0) = Field(default_factory=list)
try:
validated = Extracted(**data)
except ValidationError as e:
print("Schema violation:", e)For complex schemas, add custom validators (@validator) to reject semantically invalid combinations. The verifier should run in the same process that calls Claude, failing loudly before the data reaches a database or queue.
Step 5: Retry with targeted feedback
On parse or validation failure, open a new turn that includes the error and asks for correction. Keep the original user input and system prompt; append the assistant’s bad output and the error message, then re-prefill {.
def call_with_retry(client, system, user_text, max_retries=3):
messages = [
{"role": "user", "content": user_text},
{"role": "assistant", "content": "{"}
]
for attempt in range(max_retries):
resp = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=system,
messages=messages,
)
raw = "{" + resp.content[0].text
try:
data = extract_json(raw)
return Extracted(**data)
except (json.JSONDecodeError, ValidationError) as e:
messages.append({"role": "assistant", "content": resp.content[0].text})
messages.append({"role": "user", "content": f"Invalid: {e}. Return only corrected JSON."})
messages.append({"role": "assistant", "content": "{"})
raise RuntimeError("Exceeded retries")This loop turns occasional malformed output into a self-correcting pipeline. In practice, two retries resolve the vast majority of cases for moderate schemas. Add exponential backoff if you hit rate limits, and log the final error with the raw text for later analysis.
Step 6: Cache the static prefix
The system prompt and example are identical across calls. Anthropic supports cache_control on the system block to avoid re-pricing the prefix. Set cache_control: {"type": "ephemeral"} on the system prompt.
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=[{
"type": "text",
"text": SYSTEM_PROMPT,
"cache_control": {"type": "ephemeral"}
}],
messages=[...],
)This cuts latency and token cost on high-volume extraction. If you use a routing gateway, ensure it forwards the cache hint—n4n.ai does this automatically for Anthropic calls. The cache TTL is short (around five minutes), so it helps bursty workloads more than steady low-QPS traffic.
Step 7: Stream and reconstruct for large objects
If your JSON can exceed a few thousand tokens, stream the response and accumulate chunks. You can still use raw_decode on the accumulated buffer once the stream ends, or parse incrementally with a push parser. Set stop_sequences to ["\n"}] only if you are confident the model will not need newlines inside strings; usually it is safer to rely on the validator.
with client.messages.stream(
model="claude-3-5-sonnet-20241022",
max_tokens=4096,
system=SYSTEM_PROMPT,
messages=[{"role": "user", "content": "Large doc..."}, {"role": "assistant", "content": "{"}]
) as stream:
chunks = []
for text in stream.text_stream:
chunks.append(text)
raw = "{" + "".join(chunks)
data = extract_json(raw)Streaming does not change the contract; it only changes how you collect the bytes. Validate the final object exactly as in Step 4.
Verify success
A successful run produces an Extracted instance with correct types and no exceptions. Write a smoke test that sends a known input and asserts the parsed fields:
def test_extraction():
out = call_with_retry(client, SYSTEM_PROMPT, "Storm caused sewer overflow near Main St.")
assert isinstance(out.score, float)
assert 0.0 <= out.score <= 1.0
assert isinstance(out.tags, list)
assert isinstance(out.title, str) and out.titleRun it in CI with a mocked client or a small allowance for live calls. Monitor production parse-failure rate; if it climbs above 1%, tighten the example or add a second negative example showing what not to do (e.g., “do not return a list of objects”). Getting reliable JSON from Claude without structured outputs is not magic—it’s a disciplined loop of prefill, parse, validate, and retry. The moment you treat the model as a probabilistic encoder with a strict verifier downstream, the format stops being a problem.