Integrate Claude models with first-class patterns for tool use, prompt caching, extended thinking, streaming, and safe production operation.
- Basic Messages call (stateless; you pass full history each time):
from anthropic import Anthropic
client = Anthropic()
resp = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
system="You are a precise technical writer. Cite sources inline.",
messages=[
{"role": "user", "content": "Summarize the SLSA v1.0 provenance spec in 5 bullets."},
],
)
print(resp.content[0].text)
print(resp.usage)
- Streaming for chat UX (lowest TTFT):
with client.messages.stream(
model="claude-sonnet-4-5",
max_tokens=2048,
messages=[{"role": "user", "content": "Explain mTLS in one paragraph."}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
final = stream.get_final_message()
- Tool use (function calling) — model proposes a tool call; you execute and return
tool_result:
tools = [{
"name": "get_weather",
"description": "Current weather for a city.",
"input_schema": {
"type": "object",
"properties": {"city": {"type": "string"}, "unit": {"type": "string", "enum": ["c", "f"]}},
"required": ["city"],
},
}]
messages = [{"role": "user", "content": "What's the weather in Berlin in C?"}]
while True:
resp = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
tools=tools,
messages=messages,
)
if resp.stop_reason != "tool_use":
print(next(b.text for b in resp.content if b.type == "text"))
break
tool_results = []
for block in resp.content:
if block.type == "tool_use":
result = run_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
messages.append({"role": "assistant", "content": resp.content})
messages.append({"role": "user", "content": tool_results})
- Prompt caching for stable, large prefixes (system prompt, docs, tool defs). Mark the last block of the cacheable prefix:
resp = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
system=[
{"type": "text", "text": LARGE_STYLE_GUIDE,
"cache_control": {"type": "ephemeral"}},
],
messages=[{"role": "user", "content": user_q}],
)
Cache hits cost ~10% of normal input tokens; misses cost ~125%. Worth it when prefix is reused >2×.
- Extended thinking for hard reasoning (visible chain-of-thought):
resp = client.messages.create(
model="claude-opus-4",
max_tokens=4096,
thinking={"type": "enabled", "budget_tokens": 16000},
messages=[{"role": "user", "content": HARD_PROBLEM}],
)
for block in resp.content:
if block.type == "thinking":
log.debug("reasoning: %s", block.thinking)
elif block.type == "text":
print(block.text)
- Vision — pass images as base64 or URL blocks alongside text:
{"role": "user", "content": [
{"type": "image", "source": {"type": "url", "url": "https://.../diagram.png"}},
{"type": "text", "text": "What is this diagram showing?"},
]}
- Batches for non-interactive bulk work (50% cheaper, ≤24 h):
batch = client.messages.batches.create(requests=[
{"custom_id": f"doc-{i}",
"params": {"model": "claude-haiku-4", "max_tokens": 512,
"messages": [{"role": "user", "content": doc}]}}
for i, doc in enumerate(docs)
])
- Robust client wrapper — retries with backoff, honor
retry-after, redact PII before logging, cap max_tokens and total context. The SDK retries transient 5xx / 429 by default; configure max_retries and timeout explicitly.
- Cost & latency telemetry — record
usage.input_tokens, usage.output_tokens, usage.cache_read_input_tokens, usage.cache_creation_input_tokens, model name, and end-to-end latency per call. Alert on cost-per-request anomaly.
- Eval before swap — when changing model or prompt, run a fixed eval set (≥50 cases) measuring task accuracy, cost, and latency p50/p95. Don't ship on vibes.
