LLM APIs in 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:

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

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

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

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:

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:

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)

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:

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:

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:

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:

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

#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:

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:

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:

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:

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:

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:

Turn usage into dollars with a small function:

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):

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.

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()

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

#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.