Claude API Mechanics & Integration

1. Setup

What you need: Python 3.9+ (or Node 18+), an editor, and credentials. About 60 minutes.

python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\Activate.ps1
pip install anthropic

# TypeScript equivalent
# npm install @anthropic-ai/sdk

Credential resolution is worth knowing because the exam tests secrets handling (Domain 7) and because an unset ANTHROPIC_API_KEY does not mean you have no credentials. The SDK resolves, first match wins: ANTHROPIC_API_KEY, then ANTHROPIC_AUTH_TOKEN, then an OAuth profile created by ant auth login, then Workload Identity Federation. A bare constructor picks up whichever won.

import anthropic

# Resolves from the environment. Prefer this. Never hardcode a key.
client = anthropic.Anthropic()

# Async client for concurrent workloads
async_client = anthropic.AsyncAnthropic()

2. The Response Is a List of Typed Blocks

The single most common integration bug is response.content[0].text. That works until the day you enable thinking or tools, and then block zero is a thinking block or a tool_use block and you get an AttributeError in production. Always narrow by .type.

response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=16000,
    system="You are a precise technical assistant.",
    messages=[{"role": "user", "content": "What is the capital of France?"}],
)

for block in response.content:
    if block.type == "text":
        print(block.text)
    elif block.type == "thinking":
        print("[reasoning]", block.thinking)
    elif block.type == "tool_use":
        print("[tool]", block.name, block.input)

The API is stateless. There is no server-side conversation. You resend the full history on every turn, which is why context management (Lab 5) and caching (Lab 6) matter so much economically.

Stop reasons drive your control flow

Never parse the text to decide what to do next. Branch on stop_reason.

3. Thinking and Effort: the Current Shapes

Use adaptive thinking: Claude decides when and how much to reason. Depth is controlled by effort, which lives inside output_config, not at the top level.

response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=16000,
    thinking={"type": "adaptive", "display": "summarized"},
    output_config={"effort": "high"},   # low | medium | high | xhigh | max
    messages=[{"role": "user", "content": "Plan a migration for this schema..."}],
)

response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=16000,
    thinking={"type": "enabled", "budget_tokens": 8000},   # removed
    temperature=0.7,                                        # removed
    top_p=0.9,                                              # removed
    messages=[...],
)

Two subtleties the exam likes. First, display defaults to "omitted" on current models; thinking blocks still arrive, but with empty text. If you render reasoning in a UI, you must ask for "summarized" or your users see a long pause and then nothing. Second, on Opus-tier models, omitting thinking entirely means no thinking at all; set {"type": "adaptive"} explicitly if you want it.

4. Streaming

Stream whenever output could be long. This is not only a UX choice: the SDK refuses non-streaming requests it estimates will exceed the HTTP timeout, and large max_tokens without streaming raises before a single token is generated. Rule of thumb: non-streaming max_tokens around 16,000; streaming, go to 64,000 and beyond.

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=64000,
    messages=[{"role": "user", "content": "Write a detailed migration plan."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final = stream.get_final_message()   # full Message: stop_reason, usage, content

print("\nstop:", final.stop_reason, "| out tokens:", final.usage.output_tokens)

The helper accumulates state for you. If you only want the raw event iterator with lower memory use, pass stream=True to messages.create() instead, but then nothing is accumulated and you assemble the final message yourself.

5. Multi-Format Input

Images and documents are content blocks inside the user turn, not a different endpoint.

import base64

with open("chart.png", "rb") as f:
    img = base64.standard_b64encode(f.read()).decode("utf-8")

with open("report.pdf", "rb") as f:
    pdf = base64.standard_b64encode(f.read()).decode("utf-8")

response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=16000,
    messages=[{
        "role": "user",
        "content": [
            {"type": "image",
             "source": {"type": "base64", "media_type": "image/png", "data": img}},
            # Document block goes BEFORE the text block it refers to.
            {"type": "document",
             "source": {"type": "base64", "media_type": "application/pdf", "data": pdf}},
            {"type": "text", "text": "Does the chart agree with the report's Q3 figures?"},
        ],
    }],
)

Images also accept {"type": "url", "url": "..."}. For files reused across many requests, upload once via the Files API and reference by file_id instead of re-encoding megabytes on every call.

6. Error Handling That Distinguishes Retryable From Fatal

The exam's Domain 4 objective is literally "error type identification, recovery strategy selection." The mistake is a single broad except that swallows a 404 and a 429 identically, then retries both. Catch a chain, most specific first.

import anthropic

try:
    response = client.messages.create(...)

except anthropic.NotFoundError:                 # 404 - bad model ID. Never retry.
    raise
except anthropic.BadRequestError as e:          # 400 - malformed request. Never retry.
    log.error("fix the request: %s", e.message)
    raise
except anthropic.AuthenticationError:           # 401 - bad/missing key. Never retry.
    raise
except anthropic.RateLimitError as e:           # 429 - retryable.
    wait = int(e.response.headers.get("retry-after", "60"))
    backoff(wait)
except anthropic.APIStatusError as e:           # any other non-2xx
    if e.status_code >= 500:                    # 500 / 529 - retryable
        backoff()
    else:
        raise
except anthropic.APIConnectionError:            # network failure, no response
    backoff()

Two more traps. Every response exposes response._request_id; despite the underscore it is public, and it is the first thing Anthropic support will ask for. And the SDK timeout unit differs by language: seconds in Python and Ruby, milliseconds in TypeScript.

7. Realtime or Batch?

This is the exam's own Sample 1, so learn the decision rather than the trivia. The Message Batches API processes large asynchronous workloads within a 24-hour window at 50% of standard token cost. It is the right answer whenever the work is high-volume and latency-tolerant.

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"doc-{i}",                       # YOUR correlation key
        params=MessageCreateParamsNonStreaming(
            model="claude-opus-4-8",
            max_tokens=1024,
            messages=[{"role": "user", "content": f"Summarize: {doc}"}],
        ),
    )
    for i, doc in enumerate(documents)
])

# Poll until processing_status == "ended", then stream results.
results = {}
for result in client.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        msg = result.result.message
        results[result.custom_id] = next(
            (b.text for b in msg.content if b.type == "text"), ""
        )
    elif result.result.type == "errored":
        # "invalid_request" = fix and resubmit. Anything else = safe to retry.
        handle_error(result.custom_id, result.result.error)

8. Lab Exercise: A Resilient Client

Objective: build a small module that survives the failure modes above, then prove each one.

1. Block-safe extraction. Write extract_text(response) that walks response.content and concatenates only text blocks. Enable thinking={"type": "adaptive"} and confirm it still works where content[0].text would have crashed.
2. Stop-reason dispatch. Write a handler that branches on all six stop reasons. Force max_tokens by setting max_tokens=16 on a long request and confirm your code detects truncation rather than treating it as a complete answer.
3. Prove the 400s. Send one request with budget_tokens and one with temperature. Capture the BadRequestError and read the message. Now you will never write them again.
4. Stream with a final message. Stream a long generation, print deltas as they arrive, then read stop_reason and usage off get_final_message().
5. Typed error chain. Trigger a NotFoundError with a bogus model ID. Confirm your handler does not retry it. Then set max_retries=0 and observe the difference.
6. Batch, keyed correctly. Submit 20 short requests as a batch with meaningful custom_ids. Deliberately reassemble results positionally, observe the mismatch, then fix it by keying on custom_id.
7. Multi-format. Send an image and a PDF in one user turn and ask a question that requires both.

Step 6 is the one that shows up in production. Do it.

Key Takeaways

• Everything is POST /v1/messages. Tools, thinking, structured output, vision, and caching are parameters, not endpoints.
• content is a list of typed blocks. Narrow on .type. Never index block zero.
• Branch on stop_reason, never on the text. Guard stop_details: it is populated only for refusal.
• Current shapes: thinking={"type": "adaptive"} and output_config={"effort": ...}. budget_tokens, temperature, top_p, top_k, and assistant prefill all return 400.
• Stream anything long. Non-streaming past roughly 16k max_tokens risks an HTTP timeout the SDK will refuse outright.
• Catch a typed error chain: 400/401/404 are fatal, 429/5xx/network are retryable. The SDK already retries the retryable ones twice.
• Batches: 24-hour window, 50% cost, no latency SLA, single-turn only, and results are unordered: key by custom_id.