L2 · 03 · Calling LLM APIs Properly

The Level 1 examples made a single happy-path HTTP call. Real code needs configuration, typed requests, timeouts, retries with back-off, streaming, token and cost tracking, and good error handling. This page builds a small, production-shaped LLM client in Scala 3 (sttp + circe) and Python (httpx).

Anatomy of an LLM API call

Configuration, never hard-coded

Which errors to retry

Exponential back-off with jitter

Wait base × 2attempt, capped, multiplied by a random factor. The randomness ("jitter") stops many clients from retrying in lock-step and overloading the server again. Limit total attempts (3–5) and total time.

Scala 3 client (sttp + circe)

//> using scala 3
//> using dep com.softwaremill.sttp.client4::core:4.0.25
//> using dep io.circe::circe-core:0.14.15
//> using dep io.circe::circe-parser:0.14.15
import io.circe.{Codec, Decoder, Encoder}
import io.circe.parser.decode
import io.circe.syntax.*
import sttp.client4.*
import sttp.model.Uri
import scala.concurrent.duration.*
import scala.util.{Random, Try}

// ---------- Wire types (OpenAI-compatible chat completions) ----------
final case class Message(role: String, content: String) derives Codec.AsObject
final case class ChatRequest(model: String, messages: List[Message], temperature: Double, max_tokens: Int)
    derives Encoder.AsObject
final case class Usage(prompt_tokens: Int, completion_tokens: Int) derives Decoder
final case class Choice(message: Message, finish_reason: Option[String]) derives Decoder
final case class ChatResponse(choices: List[Choice], usage: Option[Usage]) derives Decoder:
  def text: String = choices.headOption.map(_.message.content).getOrElse("")

// ---------- Config & errors ----------
final case class LlmConfig(baseUrl: String, model: String, apiKey: Option[String])
object LlmConfig:
  def fromEnv: LlmConfig = LlmConfig(
    baseUrl = sys.env.getOrElse("LLM_BASE_URL", "http://localhost:11434/v1"),
    model   = sys.env.getOrElse("LLM_MODEL", "llama3.2"),
    apiKey  = sys.env.get("LLM_API_KEY")
  )

enum LlmError:
  case Http(status: Int, body: String, retryAfterSec: Option[Int])
  case Network(message: String)
  case Decode(message: String)

  def retryable: Boolean = this match
    case Http(s, _, _) => Set(408, 409, 429, 500, 502, 503, 504)(s)
    case Network(_)    => true
    case Decode(_)     => false

// ---------- Client ----------
final class LlmClient(cfg: LlmConfig, backend: SyncBackend = DefaultSyncBackend()):
  private val endpoint = Uri.unsafeParse(s"${cfg.baseUrl.stripSuffix("/")}/chat/completions")

  def chat(
      messages: List[Message],
      temperature: Double = 0.2,
      maxTokens: Int = 800,
      maxAttempts: Int = 4
  ): Either[LlmError, ChatResponse] =
    val body = ChatRequest(cfg.model, messages, temperature, maxTokens).asJson.noSpaces

    def once(): Either[LlmError, ChatResponse] =
      val base   = basicRequest.post(endpoint).contentType("application/json").body(body).readTimeout(90.seconds)
      val req    = cfg.apiKey.fold(base)(key => base.auth.bearer(key))
      Try(req.send(backend)).toEither.left.map(e => LlmError.Network(e.toString)).flatMap { resp =>
        resp.body match
          case Right(json) => decode[ChatResponse](json).left.map(e => LlmError.Decode(e.getMessage))
          case Left(err)   => Left(LlmError.Http(resp.code.code, err.take(500), resp.header("Retry-After").flatMap(_.toIntOption)))
      }

    def backoff(attempt: Int, retryAfter: Option[Int]): FiniteDuration =
      retryAfter.map(_.seconds).getOrElse:
        val capMs = math.min(30000.0, 500.0 * math.pow(2, attempt))
        (capMs * (0.5 + Random.nextDouble() / 2)).toLong.millis

    def loop(attempt: Int): Either[LlmError, ChatResponse] =
      once() match
        case Left(e) if e.retryable && attempt < maxAttempts =>
          val wait = backoff(attempt, e match { case LlmError.Http(_, _, ra) => ra; case _ => None })
          System.err.println(s"[llm] attempt $attempt failed ($e), retrying in ${wait.toMillis} ms")
          Thread.sleep(wait.toMillis)
          loop(attempt + 1)
        case result =>
          result.foreach(r => r.usage.foreach(u => System.err.println(s"[llm] tokens in=${u.prompt_tokens} out=${u.completion_tokens}")))
          result

    loop(1)

/** Prices are per million tokens and vary by model: read them from config, don't hard-code. */
def estimateCostUsd(u: Usage, inPerMillion: BigDecimal, outPerMillion: BigDecimal): BigDecimal =
  (BigDecimal(u.prompt_tokens) * inPerMillion + BigDecimal(u.completion_tokens) * outPerMillion) / 1000000

@main def ask(question: String): Unit =
  val client = LlmClient(LlmConfig.fromEnv)
  val result = client.chat(List(
    Message("system", "You are a concise assistant for retail store staff."),
    Message("user", question)
  ))
  result match
    case Right(r) =>
      println(r.text)
      if r.choices.headOption.flatMap(_.finish_reason).contains("length") then
        System.err.println("[llm] warning: answer was cut off (max_tokens reached)")
    case Left(err) => System.err.println(s"LLM call failed: $err")

# Local model, no key needed:
ollama pull llama3.2
scala-cli LlmClient.scala -- "How should staff handle a pickup order that is not on the shelf?"

# Hosted provider or company gateway:
export LLM_BASE_URL="https://<your-provider-or-gateway>/v1"
export LLM_MODEL="<approved-model>"
export LLM_API_KEY="$(<read from your secret manager>)"

Design points:

• Errors are values (Either[LlmError, ...]), with retryability decided in one place.
• Both timeouts: read timeout on the request, and a connection timeout on the backend if you need it. LLM calls can legitimately take 30–90 s, so set the read timeout accordingly.
• Usage is logged, content is not. Prompts can contain customer data, so log token counts, latency and model, and avoid logging full prompts in production.
• finish_reason == "length" means the answer was truncated. Treat it as an error for structured output.
• For effectful code (Cats Effect / ZIO), the same design applies: use sttp's effect backends and the effect library's retry and timeout combinators instead of Thread.sleep.

Python client (httpx)

import json
import os
import random
import sys
import time
from collections.abc import Iterator

import httpx

RETRYABLE = {408, 409, 429, 500, 502, 503, 504}


class LlmClient:
    def __init__(self) -> None:
        self.base_url = os.environ.get("LLM_BASE_URL", "http://localhost:11434/v1").rstrip("/")
        self.model = os.environ.get("LLM_MODEL", "llama3.2")
        key = os.environ.get("LLM_API_KEY")
        self.headers = {"Authorization": f"Bearer {key}"} if key else {}
        self.http = httpx.Client(timeout=httpx.Timeout(90.0, connect=10.0))

    def _wait(self, attempt: int, retry_after: str | None) -> float:
        if retry_after and retry_after.isdigit():
            return float(retry_after)
        return min(30.0, 0.5 * 2**attempt) * random.uniform(0.5, 1.0)

    def chat(self, messages: list[dict], temperature: float = 0.2, max_tokens: int = 800, max_attempts: int = 4) -> dict:
        payload = {"model": self.model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens}
        for attempt in range(1, max_attempts + 1):
            try:
                r = self.http.post(f"{self.base_url}/chat/completions", json=payload, headers=self.headers)
            except (httpx.ConnectError, httpx.ReadTimeout, httpx.RemoteProtocolError) as e:
                if attempt == max_attempts:
                    raise
                wait = self._wait(attempt, None)
                print(f"[llm] network error {e!r}, retrying in {wait:.1f}s", file=sys.stderr)
                time.sleep(wait)
                continue
            if r.status_code in RETRYABLE and attempt < max_attempts:
                wait = self._wait(attempt, r.headers.get("retry-after"))
                print(f"[llm] HTTP {r.status_code}, retrying in {wait:.1f}s", file=sys.stderr)
                time.sleep(wait)
                continue
            r.raise_for_status()
            data = r.json()
            if usage := data.get("usage"):
                print(f"[llm] tokens in={usage['prompt_tokens']} out={usage['completion_tokens']}", file=sys.stderr)
            return data
        raise RuntimeError("unreachable")

    def stream(self, messages: list[dict], temperature: float = 0.2) -> Iterator[str]:
        """Yield text fragments as they arrive (Server-Sent Events)."""
        payload = {"model": self.model, "messages": messages, "temperature": temperature, "stream": True}
        with self.http.stream("POST", f"{self.base_url}/chat/completions", json=payload, headers=self.headers) as r:
            r.raise_for_status()
            for line in r.iter_lines():
                if not line.startswith("data: "):
                    continue
                chunk = line.removeprefix("data: ")
                if chunk == "[DONE]":
                    break
                delta = json.loads(chunk)["choices"][0].get("delta", {}).get("content")
                if delta:
                    yield delta


if __name__ == "__main__":
    client = LlmClient()
    question = " ".join(sys.argv[1:]) or "Give 3 tips for handling a busy in-store pickup counter."
    for piece in client.stream([{"role": "user", "content": question}]):
        print(piece, end="", flush=True)
    print()

Streaming

With "stream": true, the server sends the answer as a sequence of Server-Sent Events while it is being generated. Users see text appear immediately instead of waiting 20 seconds for the full answer.

data: {"choices":[{"delta":{"role":"assistant","content":""}}]}
data: {"choices":[{"delta":{"content":"Check"}}]}
data: {"choices":[{"delta":{"content":" the back"}}]}
data: {"choices":[{"delta":{"content":" room first."}}]}
data: {"choices":[{"delta":{},"finish_reason":"stop"}]}
data: [DONE]

//> using scala 3
//> using dep io.circe::circe-parser:0.14.15
import io.circe.parser.parse
import java.net.URI
import java.net.http.{HttpClient, HttpRequest, HttpResponse}
import scala.jdk.CollectionConverters.*

@main def streamDemo(question: String): Unit =
  val base  = sys.env.getOrElse("LLM_BASE_URL", "http://localhost:11434/v1")
  val model = sys.env.getOrElse("LLM_MODEL", "llama3.2")
  val body  = io.circe.Json.obj(
    "model"    -> io.circe.Json.fromString(model),
    "stream"   -> io.circe.Json.True,
    "messages" -> io.circe.Json.arr(io.circe.Json.obj(
      "role" -> io.circe.Json.fromString("user"), "content" -> io.circe.Json.fromString(question)))
  ).noSpaces

  val builder = HttpRequest.newBuilder(URI.create(s"$base/chat/completions"))
    .header("Content-Type", "application/json")
    .POST(HttpRequest.BodyPublishers.ofString(body))
  sys.env.get("LLM_API_KEY").foreach(k => builder.header("Authorization", s"Bearer $k"))

  val lines = HttpClient.newHttpClient().send(builder.build(), HttpResponse.BodyHandlers.ofLines()).body()
  lines.iterator().asScala
    .collect { case l if l.startsWith("data: ") => l.stripPrefix("data: ") }
    .takeWhile(_ != "[DONE]")
    .flatMap(chunk => parse(chunk).toOption.flatMap(_.hcursor.downField("choices").downN(0).downField("delta").get[String]("content").toOption))
    .foreach(piece => print(piece))
  println()

Tokens, cost and latency

• Every response reports usage (input and output tokens). Log it per feature so you know what each feature costs.
• Output tokens usually cost more than input and take longer. Ask for concise answers and set max_tokens.
• Prompt caching: many providers discount or speed up repeated prefixes. Put stable content (system prompt, policies, tool definitions) first and variable content last.
• Latency has two parts: time-to-first-token (affected by prompt length and model) and tokens per second. Streaming hides the second.
• Batch APIs: for offline jobs (classify 50,000 tickets overnight), many providers offer discounted asynchronous batch processing.
• Right-size the model: small models for classification and extraction, big ones for reasoning. Measure with golden tests (page 01).

SDKs vs raw HTTP

Production checklist

• Config from environment / secret manager, with an exact model pinned.
• Connect and read timeouts. Overall deadline per user request.
• Retries only for retryable errors, with jittered back-off and Retry-After.
• Concurrency limit (a semaphore or pool) to stay under rate limits.
• Structured logging of model, latency, tokens and status, with no secrets and minimal customer data.
• Truncation (finish_reason) and refusal handling.
• Fallback behaviour: a smaller or backup model, a cached answer, or a graceful "AI unavailable" message.
• Golden tests run on model or prompt changes.