Calling LLMs in Python: SDKs, streaming, structured output, tools, retries

This is the core of the job. Everything you learned about dicts, JSON, generators, Pydantic, and HTTP converges here. The good news for you specifically: the Python SDKs mirror their JS/TS siblings almost method-for-method, so the shapes will feel familiar. (Model IDs like gpt-5.1 and claude-sonnet-4-5 below are illustrative — providers rev them constantly, so check the current model list rather than trusting a string in a book.)

The mental model: messages in, a response object out, statelessness throughout

An LLM chat call is a list of messages, each with a role (system, user, or assistant) and content. The API returns a response object; you dig the text out of it and, if you care, the token counts. The single most important fact: the API is stateless. The model remembers nothing between calls. "Conversation memory" is an illusion you maintain by resending the entire message history every time. This connects directly to your dict/JSON-reading skills — a response is just a nested object you navigate.

from openai import OpenAI      # OpenAI SDK
from anthropic import Anthropic # Anthropic SDK

openai_client = OpenAI()       # reads OPENAI_API_KEY from the environment
anthropic_client = Anthropic() # reads ANTHROPIC_API_KEY from the environment

Neither constructor takes a hard-coded key in real code — they read OPENAI_API_KEY / ANTHROPIC_API_KEY from the environment. Never hard-code an API key in source; that is the most common way secrets leak into git.

OpenAI — Chat Completions:

resp = openai_client.chat.completions.create(
    model="gpt-5.1",
    messages=[
        {"role": "system", "content": "You are a terse assistant."},
        {"role": "user", "content": "Capital of France?"},
    ],
    temperature=0.7,     # 0 = deterministic-ish, higher = more varied
    max_tokens=100,      # cap on the response length
)
print(resp.choices[0].message.content)   # => "Paris."
print(resp.usage.total_tokens)           # => e.g. 23  (billing lives here)

The text is buried at resp.choices[0].message.content — a list of choices, each with a message, whose content is the string. Token usage sits on resp.usage.

OpenAI — the Responses API (the newer, agent-oriented surface) flattens this:

resp = openai_client.responses.create(
    model="gpt-5.1",
    input="Capital of France?",   # a string, or a messages-style list
)
print(resp.output_text)   # => "Paris."  — convenience accessor, no digging

Anthropic — Messages:

resp = anthropic_client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=100,        # required on Anthropic
    system="You are a terse assistant.",   # system is a top-level param, not a message
    messages=[{"role": "user", "content": "Capital of France?"}],
)
print(resp.content[0].text)              # => "Paris."
print(resp.usage.input_tokens, resp.usage.output_tokens)

Note the shape differences: Anthropic puts system at the top level (not in messages), requires max_tokens, and returns a content list of blocks (resp.content[0].text) because a response can mix text, tool calls, and thinking. Same concept, different object graph — which is exactly why reading the response object carefully matters.

JS → Python: These SDKs are near-identical to openai and @anthropic-ai/sdk for Node. client.chat.completions.create({...}) in TS becomes client.chat.completions.create(...) with keyword args in Python. resp.choices[0].message.content is the same path in both languages. The main Python-isms: keyword arguments instead of an options object, and snake_case (max_tokens) instead of camelCase.

Streaming: tokens as a generator

For responsive UIs you stream, receiving tokens as they're generated. This is the generator/iterator pattern from earlier chapters, applied to the network. You iterate; each element is a partial chunk.

# OpenAI — stream=True turns the call into an iterator of chunks
stream = openai_client.chat.completions.create(
    model="gpt-5.1",
    messages=[{"role": "user", "content": "Write a haiku."}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content   # the NEW text in this chunk (or None)
    if delta:
        print(delta, end="", flush=True)     # print without newline, flush immediately

The token text lives on delta (the increment), not message (the whole thing) — a classic gotcha. Anthropic gives you a higher-level helper that accumulates state for you:

# Anthropic — a streaming context manager with a convenience text iterator
with anthropic_client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=200,
    messages=[{"role": "user", "content": "Write a haiku."}],
) as stream:
    for text in stream.text_stream:          # just the text deltas
        print(text, end="", flush=True)
    final = stream.get_final_message()       # the assembled full message, after the loop

Both SDKs also ship async clients (AsyncOpenAI, AsyncAnthropic) whose streams you consume with async for — the same shape under await, for when you're fanning out many calls concurrently.

Structured output: the reliability backbone

Free-text output is unparseable. When you need data — a classification, an extraction, fields for a database — you constrain the model to a schema and validate the result. This is where Chapter 8's Pydantic pays off: you define a model, hand its schema to the LLM, and get back a validated object instead of hoping a string is valid JSON.

from pydantic import BaseModel, ValidationError

class Contact(BaseModel):
    name: str
    email: str
    wants_demo: bool

# OpenAI — .parse() takes a Pydantic class and returns a typed, validated instance
try:
    completion = openai_client.chat.completions.parse(
        model="gpt-5.1",
        messages=[{"role": "user",
                   "content": "Jane Doe, jane@co.com, would love a demo."}],
        response_format=Contact,   # the SDK sends the JSON schema and parses the reply
    )
    contact = completion.choices[0].message.parsed   # a validated Contact instance
    print(contact.name, contact.wants_demo)          # => Jane Doe True
except ValidationError as e:
    print("Model returned data that didn't fit the schema:", e)

The equivalent on the Responses API is openai_client.responses.parse(..., text_format=Contact), reading resp.output_parsed. Anthropic reaches the same goal via tool use (below) with a single tool whose schema is your model, or by instructing JSON output and validating with Contact.model_validate_json(text). Whichever path, the discipline is identical: define a schema, get JSON back, validate it, and handle the ValidationError. Code that does json.loads(resp_text) on raw model output with no schema and no error handling is the fragile version this replaces.

Tool / function calling: the mechanical heart of agents

An agent is a loop: the model, given tools, can request a tool call instead of answering; your code runs the real function and feeds the result back; the model continues. Nothing about it is magical — it's a while loop over a stateless API.

import json

def get_weather(location: str) -> str:
    return f"18°C and clear in {location}"   # a real implementation would hit an API

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get the current weather for a location.",
        "parameters": {
            "type": "object",
            "properties": {"location": {"type": "string"}},
            "required": ["location"],
        },
    },
}]

messages = [{"role": "user", "content": "What's the weather in Paris?"}]

while True:
    resp = openai_client.chat.completions.create(
        model="gpt-5.1", messages=messages, tools=tools,
    )
    msg = resp.choices[0].message
    messages.append(msg)                     # keep the assistant turn in history

    if not msg.tool_calls:                   # no tool requested -> we're done
        print(msg.content)
        break

    # The model asked to call one or more tools. Run each, append the result.
    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)   # arguments arrive as a JSON string
        result = get_weather(**args)                 # dispatch to the real function
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,          # ties the result to the request
            "content": result,
        })
    # loop again — the model now sees the tool result and continues

That is the entire mechanism. Anthropic's version is structurally the same — the model returns a tool_use content block, you append a tool_result block keyed by its id, and re-call messages.create until stop_reason is no longer tool_use. Note the two things that must be right: tool arguments come back as a JSON string you must parse (and validate — the model can emit malformed or hallucinated args), and every tool result must carry the id linking it to the request.

Robustness: retries, cost, and prompt injection

Real deployments wrap calls in retry logic because rate limits (429) and transient 5xx errors are normal, not exceptional. The SDKs retry a couple of times by default, but explicit exponential backoff — waiting longer after each failure — is the standard pattern. The tenacity library expresses it declaratively:

from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
import openai

@retry(
    retry=retry_if_exception_type((openai.RateLimitError, openai.APIStatusError)),
    wait=wait_exponential(min=1, max=30),   # 1s, 2s, 4s ... capped at 30s
    stop=stop_after_attempt(5),
)
def ask(prompt: str) -> str:
    resp = openai_client.chat.completions.create(
        model="gpt-5.1", messages=[{"role": "user", "content": prompt}],
    )
    return resp.choices[0].message.content

The hand-rolled equivalent is a for loop with time.sleep(2 ** attempt) between tries — same idea, more code. Only retry the retryable errors (rate limits, 5xx, timeouts); a 400 "bad request" will fail identically every time.

Two more habits. Token/cost awareness: every response carries usage; you're billed per token, so log it, and know that resending long histories every turn (statelessness, remember) makes cost grow with conversation length. Prompt templating and its injection risk: you'll build prompts with f-strings, str.format, or Jinja templates —

prompt = f"Summarise this support ticket:\n\n{user_ticket}"

— and the moment user_ticket contains untrusted text, that text can carry instructions the model may follow ("ignore previous instructions and…"). Treat interpolated user content as data, not trusted instructions: keep it clearly delimited, put your real instructions in the system role, and never let a template blindly concatenate a secret or another user's data into the prompt.

JS → Python: The Python and Node SDKs are siblings — openai.chat.completions.create, stream=True, the tool-call loop, and Pydantic-vs-Zod structured output all map one-to-one (response_format=Contact ↔ response_format: zodResponseFormat(Contact)). Async Python (AsyncOpenAI + await) is the direct analogue of the promise-based JS client you already know. tenacity's @retry decorator is the Python spelling of a retry wrapper like p-retry.

Reading & judging: The high-value smells in LLM code are specific. Hard-coded API keys in source (should be os.environ). No retry/backoff around the API call (one rate-limit blip kills the job). No timeout on the underlying HTTP (a hung call wedges the worker — the SDKs set defaults, but a custom httpx client can strip them). Unvalidated model JSON used directly — json.loads(output) with no schema and no try/except. Swallowed API errors — a bare except: pass around the call hides rate limits and auth failures. Unbounded concurrency — a for over thousands of prompts firing all at once with no semaphore. Partial/failed tool calls dropped — a loop that assumes exactly one well-formed tool call and crashes on zero, many, or malformed arguments. And injection-unsafe templating — untrusted user text concatenated straight into an instruction prompt with no delimiting or role separation. Spotting these is most of what "reviewing AI code" actually means.

Try it: With a key in your environment, run the streaming example and watch tokens print live. Then take the tool-calling loop and add a print(f"[tool] {call.function.name}({args})") before dispatch — now you can see the agent's decisions. Finally, break the structured-output example on purpose (ask for a Contact from text that has no email) and confirm you get a ValidationError you can catch, not a silent bad object.

Sources: OpenAI Structured Outputs guide, OpenAI Function calling guide, OpenAI Streaming guide, openai-python parsed responses (DeepWiki).

---

---