AskAiml
Sign in Get started

Calling LLM APIs (OpenAI, Anthropic, and open source)

One mental model across providers: messages, roles, streaming, timeouts, retries, and self-hosted parity.

💬 Module 1 8 min read Not started

Why this matters

Every agentic system you will build sends LLM API calls. The patterns in this lesson — message roles, streaming, error handling, async — are reused in every framework on top. Master the raw API and the frameworks become trivial.

In 2026 OpenAI ships two APIs (Chat Completions and the newer Responses API), and Anthropic ships a single Messages API with prompt caching and a built-in Tool Runner. You will learn both because production code uses both.

Learning objectives

  1. Make synchronous and asynchronous calls to OpenAI Responses, OpenAI Chat Completions, and Anthropic Messages APIs.
  2. Handle streaming, tool calls, and structured outputs.
  3. Use prompt caching to cut Claude costs by up to 90%.
  4. Switch providers behind a single function (provider-agnostic code).
  5. Handle retries, timeouts, and rate-limit errors like a senior engineer.

1. The three message roles

Every chat-style API uses these:

RoleWhat it is
systemBoss instructions ("You are a polite SQL expert")
userWhat the human typed
assistantWhat the model previously said (history)
toolResult of a tool call (newer APIs use this)

The model is stateless. You must include the full history every time.

2. OpenAI — the Responses API (2026 default)

OpenAI now recommends the Responses API for all new projects. It is an evolution of Chat Completions with cleaner agentic primitives (instructions separated from input, typed output items, built-in web_search, file_search, computer_use tools).

python
# pyproject.toml: uv add openai
from openai import OpenAI
client = OpenAI()  # reads OPENAI_API_KEY from env

resp = client.responses.create(
    model="gpt-5.5",  # or "gpt-4.1", "gpt-4.1-mini", "gpt-5-nano"
    instructions="You answer in 1 sentence.",
    input="Why is the sky blue?",
)
print(resp.output_text)

output_text is the helper that gives you the final string. For agents you will iterate over resp.output because it contains typed items: message, reasoning, function_call, function_call_output, web_search_call, etc.

Multi-turn

python
history = [
    {"role": "user", "content": "I want to buy a laptop."},
    {"role": "assistant", "content": "Sure! What is your budget?"},
    {"role": "user", "content": "Around 80,000 INR."},
]

resp = client.responses.create(
    model="gpt-4.1-mini",
    instructions="You are a helpful shopping advisor.",
    input=history,
)
print(resp.output_text)

Streaming

python
with client.responses.stream(
    model="gpt-4.1-mini",
    instructions="Answer briefly.",
    input="Explain RAG.",
) as stream:
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)

Async

python
import asyncio
from openai import AsyncOpenAI

aclient = AsyncOpenAI()

async def ask(q: str) -> str:
    r = await aclient.responses.create(
        model="gpt-4.1-mini",
        input=q,
    )
    return r.output_text

async def main():
    qs = ["What is RAG?", "What is LangGraph?", "What is MCP?"]
    answers = await asyncio.gather(*(ask(q) for q in qs))
    for q, a in zip(qs, answers):
        print(q, "->", a[:60], "...")

asyncio.run(main())

3. OpenAI — Chat Completions (still widely used)

You will see this in older code and many tutorials. Same model, different shape.

python
resp = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[
        {"role": "system", "content": "You answer in 1 sentence."},
        {"role": "user", "content": "Why is the sky blue?"},
    ],
    temperature=0.2,
)
print(resp.choices[0].message.content)
print(resp.usage)  # prompt_tokens, completion_tokens, total_tokens

When to use which:

  • New projects → Responses API.
  • Existing code, simple use cases → Chat Completions is still fully supported.
  • Need built-in web_search, file_search, computer_use tools → Responses API only.

4. Anthropic Messages API

python
# uv add anthropic
import anthropic
aclient = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY

msg = aclient.messages.create(
    model="claude-opus-4-7",  # or claude-sonnet-4-6, claude-haiku-4-5
    max_tokens=1024,
    system="You answer in 1 sentence.",
    messages=[{"role": "user", "content": "Why is the sky blue?"}],
)
print(msg.content[0].text)

Note max_tokens is required in Anthropic.

Streaming

python
with aclient.messages.stream(
    model="claude-haiku-4-5",
    max_tokens=512,
    messages=[{"role": "user", "content": "Explain MCP."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Prompt caching (the cost-killer)

Anthropic's prompt caching gives up to 90% off on cached input tokens. Use it whenever you have a long, stable prefix (system prompt, document, code base context).

python
big_doc = open("contract.txt").read()  # tens of thousands of tokens

resp = aclient.messages.create(
    model="claude-opus-4-7",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": f"You analyze legal contracts.\n\nCONTRACT:\n{big_doc}",
            "cache_control": {"type": "ephemeral"},  # 5-minute cache
        }
    ],
    messages=[{"role": "user", "content": "List the termination clauses."}],
)
print(resp.usage)
# Look for cache_creation_input_tokens and cache_read_input_tokens

Rules to remember:

  • Cached prefix must be identical byte-for-byte. A timestamp in the system prompt = silent cache miss.
  • Minimum block size: 1024 tokens for Sonnet/Opus, 2048 for Haiku.
  • Default TTL is 5 minutes; new 1-hour TTL costs 2x to write but lasts 12x longer.
  • Cache hits show in usage.cache_read_input_tokens. Always check this in dev.

You can also use the simpler automatic caching mode by passing cache_control at the top level:

python
resp = aclient.messages.create(
    model="claude-opus-4-7",
    max_tokens=1024,
    cache_control={"type": "ephemeral"},
    system=long_system_prompt,
    messages=history,
)

This caches the last cacheable block automatically and is ideal for multi-turn chat where the conversation history grows.

5. Provider-agnostic wrapper (a 30-line skill that pays a salary premium)

Senior engineers wrap providers behind one interface so the rest of the codebase does not care.

python
from typing import Protocol, Literal
from openai import OpenAI
import anthropic

class LLMClient(Protocol):
    def complete(self, system: str, user: str, *, model: str, temperature: float = 0.2) -> str: ...

class OpenAIBackend:
    def __init__(self): self.c = OpenAI()
    def complete(self, system, user, *, model, temperature=0.2):
        r = self.c.responses.create(model=model, instructions=system, input=user, temperature=temperature)
        return r.output_text

class AnthropicBackend:
    def __init__(self): self.c = anthropic.Anthropic()
    def complete(self, system, user, *, model, temperature=0.2):
        r = self.c.messages.create(
            model=model, max_tokens=1024, system=system,
            messages=[{"role":"user","content":user}], temperature=temperature,
        )
        return r.content[0].text

def make_client(provider: Literal["openai","anthropic"]) -> LLMClient:
    return {"openai": OpenAIBackend, "anthropic": AnthropicBackend}[provider]()

Now your agent code calls client.complete(system, user, model=...) and you can swap providers without touching it.

A more powerful pre-built option: litellm — a single library that proxies 100+ providers (OpenAI, Anthropic, Google, Azure, Bedrock, Together, Groq, Ollama). It is the de-facto standard for provider abstraction in 2026.

python
# uv add litellm
from litellm import completion
r = completion(
    model="anthropic/claude-haiku-4-5",  # or "openai/gpt-4.1-mini", "groq/llama-3.3-70b"
    messages=[{"role":"user","content":"Hi"}],
)
print(r.choices[0].message.content)

6. Errors, retries, timeouts (production hygiene)

You will hit rate limits and 5xx errors. Wrap every call.

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

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    retry=retry_if_exception_type((
        openai.RateLimitError,
        openai.APITimeoutError,
        openai.InternalServerError,
    )),
)
def safe_call(prompt: str) -> str:
    r = client.responses.create(model="gpt-4.1-mini", input=prompt, timeout=30)
    return r.output_text

The 5 errors you must handle:

  • RateLimitError (429) — back off + retry.
  • APITimeoutError — your timeout fired; retry once.
  • InternalServerError (5xx) — provider problem; retry.
  • BadRequestError (400) — your bug; do not retry.
  • AuthenticationError (401) — bad key; do not retry.

7. Cost awareness (count before you call)

python
import tiktoken

def estimate_cost(prompt: str, model: str = "gpt-4.1-mini") -> dict:
    enc = tiktoken.encoding_for_model("gpt-4.1")  # close enough for 4.1-mini
    n = len(enc.encode(prompt))
    # 2026 published rates (always check the model page!):
    # gpt-4.1-mini: $0.40 / 1M input, $1.60 / 1M output
    in_cost = n / 1_000_000 * 0.40
    return {"tokens": n, "input_usd": round(in_cost, 6)}

print(estimate_cost("hello"))

Habit: instrument every LLM call with token usage logs. You cannot optimize what you do not measure.

Hands-on lab (2 hours)

Build multi_llm.py:

  1. Defines a LLMClient protocol and three backends: OpenAIResponses, OpenAIChat, Anthropic.
  2. Adds tenacity-based retry with exponential backoff.
  3. Loads 5 prompts from prompts.txt.
  4. Calls all three providers in parallel (asyncio.gather).
  5. Saves a CSV with: prompt, provider, latency_ms, input_tokens, output_tokens, estimated_cost_usd.
  6. Prints a summary table sorted by cost-per-quality (you decide quality with a tiny rubric).

Acceptance criteria:

  • Async + parallel: total runtime ≤ 1.5x the slowest single call.
  • Retries triggered on simulated 429 (you can monkeypatch).
  • README explains which model wins for which prompt with data.

Common pitfalls

  1. Forgetting max_tokens for Anthropic — request fails immediately.
  2. Long system prompts without caching — paying full price every call.
  3. Mixing sync/async in tests — separate test files for each.
  4. Catching all exceptions — never except Exception: — be specific.
  5. Logging the full prompt — leaks user data. Hash or redact.

Self-check

  1. Two differences between Responses API and Chat Completions.
  2. What goes in cache_control and what does the TTL mean?
  3. When should you NOT retry on a 429?
  4. What is the minimum cacheable block size for Haiku 4.5?
  5. Why is tiktoken not exact for gpt-4.1-mini but still useful?

References

Sign in to save your progress and earn badges.