# LLM APIs in Python

> Call Claude from Python with the official SDK: messages, system prompts, history, streaming, stop reasons, errors, retries and cost.

AI Engineer roadmap · LLM foundations · 20 min read · https://levelcode.org/roadmaps/ai-engineer/llm-apis-python

Every AI feature you build starts with the same move: your code sends a request to a model and handles what comes back. This lesson teaches that move properly in Python, using Anthropic's official `anthropic` SDK. You will set up a client, send messages with a system prompt, keep a multi-turn conversation, stream responses, read usage and stop reasons, handle errors and retries, and estimate cost. Other providers' APIs follow a very similar shape, so the concepts transfer. (If you build in TypeScript, the TypeScript version of this lesson covers the same ground.)

## Setup: SDK, key and client

You need Python 3.12+ and an API key from the Claude Console. Install the SDK into a virtual environment:

```bash
python -m venv .venv
source .venv/bin/activate
pip install anthropic
```

Export the key as an environment variable, which the SDK reads automatically:

```bash
export ANTHROPIC_API_KEY="sk-ant-..."
```

> [!WARNING]
> Never put the key in source code or commit it to Git. Anyone with the key can spend your money. In a real project, keep it in your deployment platform's secret store, or in a `.env` file listed in `.gitignore`. If a key leaks, revoke it in the Console immediately.

Now create a client. One client per process is enough. It reuses HTTP connections, so create it once at module level rather than per request:

```python
import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from the environment
MODEL = "claude-opus-5-5"
```

`claude-opus-5-5` is the most capable current Opus model. `claude-sonnet-5-5` and `claude-haiku-5-5` are cheaper and faster options. Use model IDs exactly as written, with no date suffix.

## Your first request

The core endpoint is **Messages**. You send a list of messages and get one assistant message back:

```python
response = client.messages.create(
    model=MODEL,
    max_tokens=1024,
    messages=[{"role": "user", "content": "In one sentence, what is a context window?"}],
)

print(response.stop_reason)
print(response.usage.input_tokens, response.usage.output_tokens)
for block in response.content:
    if block.type == "text":
        print(block.text)
```

```text
end_turn
19 58
A context window is the maximum amount of text, measured in tokens, that a model can take into account at once, including both your input and its response.
```

Three parameters are required:

- `model`: which model to run.
- `max_tokens`: a hard cap on how many tokens the model may generate in this response, including any thinking. It is a ceiling, not a target. If the model hits it, the response is cut off mid-sentence.
- `messages`: the conversation so far, as a list of `{"role": ..., "content": ...}` dicts. The first message must have role `user`.

The response's `content` is a **list of content blocks**, not a string. Current models think before answering, so you may see a `thinking` block before the `text` block, and later lessons add `tool_use` blocks. Always select blocks by `type` rather than assuming `content[0]` is text. A small helper keeps that tidy:

```python
from anthropic.types import Message

def text_of(message: Message) -> str:
    return "".join(block.text for block in message.content if block.type == "text")
```

You can tune how hard the model thinks with `output_config={"effort": ...}`, using `"low"`, `"medium"`, `"high"`, `"xhigh"` or `"max"`. Low effort is faster and cheaper for simple tasks; higher effort helps on hard ones. Some older tutorials pass `temperature` or a thinking `budget_tokens`, or pre-fill the start of the assistant's reply. Current Claude models reject all three with a 400 error, so leave them out.

## System prompts and conversations

The `system` parameter sets the model's role, rules and context for the whole conversation. It is separate from `messages`:

```python
response = client.messages.create(
    model=MODEL,
    max_tokens=1024,
    system="You are a concise assistant for a cycling shop. Answer in at most three sentences.",
    messages=[{"role": "user", "content": "Do I need tubeless tyres?"}],
)
print(text_of(response))
```

The API is **stateless**. The model remembers nothing between calls, so to hold a conversation you send the whole history every time, alternating `user` and `assistant` turns:

```python
messages = [
    {"role": "user", "content": "My name is Priya and I ride gravel."},
    {"role": "assistant", "content": "Nice to meet you, Priya! Gravel is great fun."},
    {"role": "user", "content": "What tyre width would you suggest for me?"},
]
```

Wrap the bookkeeping in a small class. Two details matter. First, append the assistant's **entire** `response.content` (the list of blocks), not just its text, so the history is exactly what the model produced. Second, only commit the new turns after the call succeeds, so a failed request doesn't leave a dangling user message:

```python
from anthropic.types import MessageParam

class Chat:
    def __init__(self, system: str, model: str = MODEL) -> None:
        self.system = system
        self.model = model
        self.history: list[MessageParam] = []

    def send(self, user_text: str) -> str:
        user_turn: MessageParam = {"role": "user", "content": user_text}
        response = client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=self.system,
            messages=[*self.history, user_turn],
        )
        self.history += [user_turn, {"role": "assistant", "content": response.content}]
        return text_of(response)

chat = Chat("You are a friendly cycling coach.")
print(chat.send("My name is Priya."))
print(chat.send("What's my name?"))  # works because the history was re-sent
```

> [!IMPORTANT]
> Because every call re-sends the whole history, a long conversation costs more with each turn. Turn 50 pays for turns 1 to 49 again. Keep the history append-only (don't rewrite earlier turns or the system prompt between calls), and cap or summarise very long conversations.

## Reading the response: stop reasons and usage

Before you use a response, check **why the model stopped**. `stop_reason` tells you whether the content is complete:

| `stop_reason` | Meaning | What to do |
| --- | --- | --- |
| `end_turn` | Finished naturally | Use the content |
| `max_tokens` | Hit your `max_tokens` cap | Content is truncated: raise the cap or ask for less |
| `stop_sequence` | Generated one of your `stop_sequences` | Use the content |
| `refusal` | Declined for safety reasons | Don't parse the content as an answer; show a fallback |
| `model_context_window_exceeded` | Ran out of context window | Shorten or summarise the input |
| `tool_use` | Wants to call a tool | Covered in the tool use lesson |

```python
def answer_or_raise(response: Message) -> str:
    match response.stop_reason:
        case "end_turn" | "stop_sequence":
            return text_of(response)
        case "max_tokens":
            raise RuntimeError("Response truncated; raise max_tokens or ask for less")
        case "refusal":
            return "Sorry, I can't help with that request."
        case other:
            raise RuntimeError(f"Unexpected stop_reason: {other}")
```

`response.usage` reports what you are billed for. `input_tokens` covers the uncached input, `output_tokens` covers everything generated (thinking included), and `cache_creation_input_tokens` and `cache_read_input_tokens` report prompt caching, which the cost lesson covers. Log usage for every call. It is your only reliable view of spend.

## Streaming

A long answer can take many seconds to finish. **Streaming** delivers text as it is generated, so users see words immediately. The SDK's `stream()` helper is a context manager:

```python
with client.messages.stream(
    model=MODEL,
    max_tokens=4096,
    system="You are a concise technical writer.",
    messages=[{"role": "user", "content": "Explain HTTP caching in two paragraphs."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

print(f"\n[{final.stop_reason}, {final.usage.output_tokens} output tokens]")
```

`text_stream` yields only text deltas. `get_final_message()` returns the same assembled `Message` you would get without streaming, so you can still check `stop_reason`, read `usage` and append `final.content` to your history. Stream anything user-facing, and anything with a large `max_tokens`: for requests it estimates will run longer than about ten minutes, the SDK refuses to make a non-streaming call and raises an error telling you to stream.

## Errors, retries and timeouts

Network calls fail. The SDK raises typed exceptions, and you should catch them by type, never by matching message text:

| Exception | Cause | Retry? |
| --- | --- | --- |
| `anthropic.APIConnectionError` (and its subclass `APITimeoutError`) | Network failure or timeout | Yes |
| `anthropic.RateLimitError` (429) | Too many requests or tokens per minute | Yes, after a delay |
| `anthropic.APIStatusError` with `status_code >= 500`, including `OverloadedError` (529) | Server-side problem | Yes |
| `anthropic.BadRequestError` (400) | Invalid request: bad parameter, malformed history | No, fix the code |
| `anthropic.AuthenticationError` (401), `PermissionDeniedError` (403) | Bad key or missing permission | No |
| `anthropic.NotFoundError` (404) | Wrong model ID or endpoint | No |

The SDK **already retries** connection errors, 408, 409, 429 and 5xx responses twice, with exponential backoff, and honours the server's `retry-after` header. Tune it on the client, or per call with `with_options`:

```python
client = anthropic.Anthropic(max_retries=4, timeout=60.0)  # timeout in seconds

quick = client.with_options(max_retries=0, timeout=10.0).messages.create(
    model="claude-haiku-5-5",
    max_tokens=1024,  # leave room for thinking; it counts toward max_tokens
    messages=[{"role": "user", "content": "Reply with OK."}],
)
```

Your own code decides what happens when retries run out. Catch the specific errors, show a useful message, and log the request ID so you can trace the failure:

```python
import logging

log = logging.getLogger(__name__)

def safe_ask(question: str) -> str:
    try:
        response = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            messages=[{"role": "user", "content": question}],
        )
    except anthropic.RateLimitError:
        return "We're busy right now. Please try again in a minute."
    except anthropic.APIConnectionError:
        return "Couldn't reach the AI service. Check your connection."
    except anthropic.APIStatusError as e:
        log.error("Claude API error %s (request %s): %s", e.status_code, e.request_id, e.message)
        return "Something went wrong on our side."
    return answer_or_raise(response)
```

The order matters: `RateLimitError` is a subclass of `APIStatusError`, so it must come first. Don't wrap 400-class errors in a retry loop. Sending the same invalid request again will fail the same way.

## Cost math

You pay per million tokens (MTok), with separate input and output prices:

| Model | Input $/MTok | Output $/MTok |
| --- | --- | --- |
| `claude-opus-5-5` | 4.00 | 20.00 |
| `claude-sonnet-5-5` | 2.00 | 10.00 |
| `claude-haiku-5-5` | 0.10 | 0.50 (prompts up to 100K tokens) |

Turn `usage` into dollars with a small function:

```python
from anthropic.types import Usage

PRICES = {  # (input, output) in USD per million tokens
    "claude-opus-5-5": (4.00, 20.00),
    "claude-sonnet-5-5": (2.00, 10.00),
    "claude-haiku-5-5": (0.10, 0.50),
}

def cost_usd(model: str, usage: Usage) -> float:
    price_in, price_out = PRICES[model]
    return (usage.input_tokens * price_in + usage.output_tokens * price_out) / 1_000_000
```

For a call with 1,500 input and 600 output tokens on Opus 5.5, that is `1,500 × 4 / 1M + 600 × 20 / 1M = $0.006 + $0.012 = $0.018`. Output dominates even though there are fewer output tokens, because each one costs five times more. To estimate before sending, count the tokens (the endpoint is free):

```python
count = client.messages.count_tokens(
    model=MODEL,
    system="You are a concise technical writer.",
    messages=[{"role": "user", "content": "Explain HTTP caching in two paragraphs."}],
)
print(count.input_tokens)
```

Don't estimate Claude token counts with another provider's tokenizer library; counts differ between model families.

## Practice

Build `chat.py`, a terminal chat program that:

1. takes a system prompt from the first command-line argument (with a sensible default);
2. streams each reply as it is generated;
3. keeps the full conversation history;
4. after each reply, prints the stop reason, the tokens used and the running cost in dollars;
5. warns if a reply was truncated or refused;
6. survives API errors without crashing or corrupting the history, and exits cleanly on `quit` or Ctrl-C.

**Solution**

```python
import sys

import anthropic
from anthropic.types import MessageParam, Usage

MODEL = "claude-sonnet-5-5"
PRICES = {"claude-opus-5-5": (4.00, 20.00), "claude-sonnet-5-5": (2.00, 10.00), "claude-haiku-5-5": (0.10, 0.50)}

def cost_usd(model: str, usage: Usage) -> float:
    price_in, price_out = PRICES[model]
    return (usage.input_tokens * price_in + usage.output_tokens * price_out) / 1_000_000

def main() -> None:
    system = sys.argv[1] if len(sys.argv) > 1 else "You are a helpful, concise assistant."
    client = anthropic.Anthropic()
    history: list[MessageParam] = []
    total = 0.0

    while True:
        try:
            user_text = input("\nyou> ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if user_text in {"quit", "exit"}:
            break
        if not user_text:
            continue

        user_turn: MessageParam = {"role": "user", "content": user_text}
        try:
            print("claude> ", end="", flush=True)
            with client.messages.stream(
                model=MODEL,
                max_tokens=4096,
                system=system,
                messages=[*history, user_turn],
            ) as stream:
                for text in stream.text_stream:
                    print(text, end="", flush=True)
                final = stream.get_final_message()
        except KeyboardInterrupt:
            print("\n[interrupted; turn discarded]")
            continue
        except anthropic.APIError as e:
            print(f"\n[API error: {e.message}; turn discarded, try again]")
            continue

        history += [user_turn, {"role": "assistant", "content": final.content}]
        call_cost = cost_usd(MODEL, final.usage)
        total += call_cost
        print(
            f"\n[{final.stop_reason} | in {final.usage.input_tokens} / out {final.usage.output_tokens} tokens"
            f" | ${call_cost:.4f} this turn, ${total:.4f} total]"
        )
        if final.stop_reason == "max_tokens":
            print("[warning: reply was truncated]")
        elif final.stop_reason == "refusal":
            print("[warning: the model declined this request]")

    print(f"\nBye. Session cost: ${total:.4f}")

if __name__ == "__main__":
    main()
```

```bash
python chat.py "You are a patient Python tutor."
```

The user turn is only added to `history` after the stream completes, so an error or Ctrl-C mid-reply discards the whole turn and the next request is still well-formed. `anthropic.APIError` is the base class of both connection and status errors. The SDK has already retried the retryable ones before this handler sees them. Watch the "in" token count climb each turn: that is the cost of re-sending history.

## Key takeaways

- Create one `anthropic.Anthropic()` client per process; it reads `ANTHROPIC_API_KEY` from the environment, so keys never appear in code.
- `messages.create` needs `model`, `max_tokens` and `messages`. The response `content` is a list of typed blocks, so select by `block.type`.
- The API is stateless: send the full history each call, append the assistant's whole `content`, and commit turns only after success.
- Check `stop_reason` (`max_tokens`, `refusal` and others) before trusting content, and log `usage` on every call.
- Stream user-facing responses with `messages.stream()` and `get_final_message()`.
- The SDK retries transient errors (connection, 429, 5xx) for you. Catch typed exceptions, never retry 400s, and compute cost as tokens × price per million.

## Quiz

1. Your chatbot forgets everything the user said in earlier turns. What is the most likely bug?

   A. The client was created without a session ID
   B. Each request sends only the latest user message instead of the full conversation history
   C. max_tokens is too low
   D. The system prompt is missing

   **Answer: B.** The Messages API is stateless. The model only knows what is in the current request, so you must re-send every earlier user and assistant turn. There is no server-side session to enable.

2. A response comes back with stop_reason "max_tokens". What does that tell you?

   A. The model generated up to the max_tokens cap and the content is likely cut off
   B. The prompt exceeded the context window
   C. The request was rate limited
   D. The model finished its answer normally

   **Answer: A.** max_tokens is a hard cap on generated tokens. Hitting it means the output stopped mid-stream. Raise the cap, ask for a shorter answer, or treat the content as incomplete. Running out of context window has its own stop reason, model_context_window_exceeded.

3. Which failure should your code NOT retry automatically?

   A. anthropic.RateLimitError (429)
   B. anthropic.APIConnectionError
   C. anthropic.BadRequestError (400)
   D. An APIStatusError with status_code 529 (overloaded)

   **Answer: C.** A 400 means the request itself is invalid, for example a bad parameter or malformed history. Sending it again fails the same way. Rate limits, connection errors and 5xx/529 are transient, and the SDK already retries them with backoff.

4. A call on claude-sonnet-5-5 ($2 input / $10 output per million tokens) uses 10,000 input tokens and 1,000 output tokens. What does it cost?

   A. $0.012
   B. $0.12
   C. $0.002
   D. $0.03

   **Answer: D.** Input: 10,000 x $2 / 1,000,000 = $0.02. Output: 1,000 x $10 / 1,000,000 = $0.01. Total $0.03. Output tokens are priced five times higher, so they matter even when there are fewer of them.