Calling LLM APIs from Python: Messages, Streaming, Structured Output, Costs and Reliability

pip install anthropic            # official SDK (also: openai, google-genai)
# Never paste keys into notebooks. Set an environment variable instead:
export ANTHROPIC_API_KEY="sk-ant-..."          # macOS/Linux
setx ANTHROPIC_API_KEY "sk-ant-..."            # Windows (new terminal afterwards)
# or put it in a .env file listed in .gitignore and load it with python-dotenv

import anthropic
client = anthropic.Anthropic()                 # reads ANTHROPIC_API_KEY from the environment

response = client.messages.create(
    model="claude-opus-5-5",                   # current default model; claude-sonnet-5-5 / claude-haiku-5-5 are cheaper
    max_tokens=16000,                          # upper limit on output tokens
    system="You are a concise data analyst. Answer in at most 3 bullet points.",
    messages=[{"role": "user", "content": "Why might a churn model have high accuracy but low recall?"}],
)
for block in response.content:                 # the reply is a list of content blocks
    if block.type == "text":
        print(block.text)
print(response.stop_reason, response.usage.input_tokens, response.usage.output_tokens)

import json, re, time, random, types, pandas as pd, numpy as np
class FakeClient:
    """Mimics client.messages.create(...) for offline practice: returns canned but realistic responses and token usage."""
    def __init__(self, fail_first=0): self.calls, self.fail_first = 0, fail_first; self.messages = self
    def create(self, model, max_tokens, messages, system=None, **kw):
        self.calls += 1
        if self.calls <= self.fail_first: raise RuntimeError("429 rate_limit_error: please retry after 1s")
        last = messages[-1]["content"] if isinstance(messages[-1]["content"], str) else json.dumps(messages[-1]["content"])
        t = last.lower()
        if "json" in t or "output_config" in kw:
            neg = any(w in t for w in ["drain", "poor", "slow", "late", "worst", "regret", "cheap"])
            text = json.dumps({"overall": "negative" if neg else "positive", "aspects": [a for a in ["battery", "screen", "sound", "delivery", "price"] if a in t], "summary": last[:60]})
        elif "my name is" in " ".join(m["content"] for m in messages if isinstance(m["content"], str)).lower() and "name" in t:
            text = "Your name is " + re.search(r"my name is (\w+)", " ".join(m["content"] for m in messages if isinstance(m["content"], str)), re.I).group(1) + "."
        else:
            text = "• Accuracy counts both classes; with few churners, predicting 'stay' scores well.\n• Recall only counts churners caught.\n• Check precision/recall at the decision threshold."
        n_in = int(len((system or "").split() + " ".join(str(m["content"]) for m in messages).split()) * 1.3); n_out = int(len(text.split()) * 1.3)
        return types.SimpleNamespace(content=[types.SimpleNamespace(type="text", text=text)], stop_reason="end_turn", model=model,
                                     usage=types.SimpleNamespace(input_tokens=n_in, output_tokens=n_out, cache_read_input_tokens=0))
client = FakeClient()
r = client.messages.create(model="claude-opus-5-5", max_tokens=16000, system="You are a concise data analyst.",
                           messages=[{"role": "user", "content": "Why might a churn model have high accuracy but low recall?"}])
print(next(b.text for b in r.content if b.type == "text"))
print(r.stop_reason, r.usage.input_tokens, "in /", r.usage.output_tokens, "out tokens")

history = []
def chat(user_text):
    history.append({"role": "user", "content": user_text})
    resp = client.messages.create(model="claude-opus-5-5", max_tokens=2000, messages=history)
    reply = next(b.text for b in resp.content if b.type == "text")
    history.append({"role": "assistant", "content": reply})
    return reply
chat("My name is Asha and I analyse telecom churn.")
print(chat("What is my name?"))
print(len(history), "messages are sent with the next request")

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=64000,
    messages=[{"role": "user", "content": "Write a 300-word data-quality checklist for a new dataset."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()          # full message with usage, when you need it

from pydantic import BaseModel
from typing import Literal, List

class ReviewAnalysis(BaseModel):
    overall: Literal["positive", "negative", "neutral"]
    aspects: List[str]
    summary: str

response = client.messages.parse(
    model="claude-opus-5-5",
    max_tokens=16000,
    messages=[{"role": "user", "content": "Analyse this review: 'Display is bright but the battery drains fast.'"}],
    output_format=ReviewAnalysis,
)
result = response.parsed_output              # a validated ReviewAnalysis object
print(result.overall, result.aspects)

from pydantic import BaseModel, ValidationError
from typing import Literal, List
class ReviewAnalysis(BaseModel):
    overall: Literal["positive", "negative", "neutral"]
    aspects: List[str]
    summary: str
reviews = pd.read_csv("customer_reviews.csv").sample(5, random_state=2)
rows = []
for t in reviews.review_text:
    resp = client.messages.create(model="claude-haiku-5-5", max_tokens=1000, messages=[{"role": "user", "content": f"Return JSON for this review: {t}"}])
    parsed = ReviewAnalysis.model_validate_json(resp.content[0].text)
    rows.append({"review": t[:55], "overall": parsed.overall, "aspects": parsed.aspects})
print(pd.DataFrame(rows).to_string(index=False))

response = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=16000,
    thinking={"type": "adaptive", "display": "summarized"},   # the model decides how much to think
    output_config={"effort": "high"},                          # low | medium (default on Opus 5.5) | high | xhigh | max
    messages=[{"role": "user", "content": "Our churn fell from 27% to 22% after a price cut that also changed the customer mix. Can we credit the price cut?"}],
)
for b in response.content:
    print(b.type, (b.thinking if b.type == "thinking" else b.text)[:300])

PRICE = {"claude-opus-5-5": (4.00, 20.00), "claude-sonnet-5-5": (2.00, 10.00), "claude-haiku-5-5": (0.10, 0.50)}   # USD per million tokens (input, output), Oct 2026 list prices
def cost_usd(model, tokens_in, tokens_out): pin, pout = PRICE[model]; return (tokens_in * pin + tokens_out * pout) / 1e6
job = {"reviews": 100_000, "tokens_in": 350, "tokens_out": 80}            # classify 1 lakh reviews
usd_inr = 88
for m in PRICE:
    c = cost_usd(m, job["reviews"] * job["tokens_in"], job["reviews"] * job["tokens_out"])
    print(f"{m:18s} ${c:8,.2f}  ≈ ₹{c * usd_inr:10,.0f}  | with Batch API (50% off) ≈ ₹{c * usd_inr / 2:9,.0f}")

# Prompt caching: mark the long, stable part of the prompt
response = client.messages.create(
    model="claude-opus-5-5", max_tokens=16000,
    system=[{"type": "text", "text": LONG_POLICY_DOCUMENT, "cache_control": {"type": "ephemeral"}}],
    messages=[{"role": "user", "content": "Does the refund policy cover damaged boxes?"}],
)
print(response.usage.cache_read_input_tokens)     # > 0 on repeat calls when the cache hits

# Token counting
print(client.messages.count_tokens(model="claude-opus-5-5", messages=[{"role": "user", "content": "Hello"}]).input_tokens)

import anthropic
try:
    response = client.messages.create(model="claude-opus-5-5", max_tokens=16000, messages=msgs)
except anthropic.RateLimitError as e:          # 429: slow down, queue, or retry later
    retry_after = int(e.response.headers.get("retry-after", "30"))
except anthropic.BadRequestError as e:         # 400: fix the request (do not retry)
    print(e.message)
except anthropic.APIStatusError as e:          # other HTTP errors
    print(e.status_code, e.message)
except anthropic.APIConnectionError:           # network problem
    ...
if response.stop_reason == "max_tokens": ...   # output was cut off: raise max_tokens or stream
if response.stop_reason == "refusal": ...      # declined for safety: check response.stop_details

def call_with_retry(fn, tries=4, base=0.05):
    for attempt in range(1, tries + 1):
        try: return fn()
        except RuntimeError as e:
            if "429" not in str(e) or attempt == tries: raise
            wait = base * 2 ** (attempt - 1) + random.uniform(0, base); print(f"  attempt {attempt} failed ({e}); retrying in {wait:.2f}s"); time.sleep(wait)
flaky = FakeClient(fail_first=2)
r = call_with_retry(lambda: flaky.messages.create(model="claude-haiku-5-5", max_tokens=500, messages=[{"role": "user", "content": "hello"}]))
print("succeeded after", flaky.calls, "calls")

from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
batch = client.messages.batches.create(requests=[
    Request(custom_id=f"review-{i}", params=MessageCreateParamsNonStreaming(
        model="claude-haiku-5-5", max_tokens=500, messages=[{"role": "user", "content": f"Classify: {text}"}]))
    for i, text in enumerate(review_texts)])
# later: poll client.messages.batches.retrieve(batch.id).processing_status until "ended",
# then iterate client.messages.batches.results(batch.id) and match results by custom_id (order is not guaranteed)