Set Async Timeouts

Two Kinds of Time Limits

Async programs can limit time in two different places, and they behave differently:

Both are useful, and they work best together: the SDK timeout catches individual requests that hang, and an asyncio deadline guarantees that a whole operation, which may include retries and several requests, finishes on time.

SDK Timeouts

Setting Them

from anthropic import AsyncAnthropic

client = AsyncAnthropic(timeout=60.0)                  # Every request from this client

fast = client.with_options(timeout=15.0)               # A copy with a different timeout
message = await fast.messages.create(...)

The default is 10 minutes, which is generous on purpose: long responses take time (Lesson 10).

Fine-Grained Timeouts

A single number sets the overall limit for the request's network operations. For more control, use an httpx2.Timeout object:

import httpx2

client = AsyncAnthropic(timeout=httpx2.Timeout(60.0, connect=5.0))

This allows 5 seconds to establish a connection, and 60 seconds for the other phases, such as waiting for response data.

What an SDK Timeout Really Measures

HTTP client timeouts limit how long the client waits for each network operation, such as connecting, or receiving the next piece of data. They are not a deadline for the total duration of a request:

• For a non-streaming request, the server sends nothing until the full response is ready, so the read timeout effectively limits the total wait for the response.
• For a streaming request, data keeps arriving, so a stream can run longer than the timeout value as long as chunks keep coming. The timeout catches a stream that stalls.

That is why streaming is the right approach for long responses (Lesson 59), and why a separate deadline is needed when the total time matters.

Timeouts and Retries Multiply

A timed-out request is retried automatically, up to max_retries times (default 2). In the worst case, the total time is roughly:

worst case ≈ timeout × (max_retries + 1) + backoff delays

With timeout=60 and the default 2 retries, a request that keeps timing out can take more than 3 minutes before an error reaches your code. For interactive features, reduce both values, or add an asyncio deadline around the whole call.

asyncio Timeouts

asyncio.timeout() (Python 3.11+)

import asyncio

try:
    async with asyncio.timeout(20):
        message = await client.messages.create(...)
except TimeoutError:
    print("No answer within 20 seconds.")

asyncio.wait_for() (All Supported Versions)

try:
    message = await asyncio.wait_for(client.messages.create(...), timeout=20)
except asyncio.TimeoutError:
    print("No answer within 20 seconds.")

When the deadline passes, the running coroutine is cancelled. Cancellation reaches the await that is in progress, and any async with blocks inside, such as an open stream, are closed properly (Lesson 60).

Deadlines for Streams

An asyncio deadline is the right tool to limit the total duration of a stream:

chunks = []
try:
    async with asyncio.timeout(30):
        async with client.messages.stream(model="claude-opus-5-5", max_tokens=4096, messages=messages) as stream:
            async for text in stream.text_stream:
                chunks.append(text)
except TimeoutError:
    chunks.append("\n[stopped: time limit reached]")

Deadline Budgets for Multi-Step Operations

Some operations make several requests in sequence: classify, then answer, or try a primary model, then fall back to a faster one. A single overall deadline then has to be shared across the steps. Track the remaining time with a monotonic clock, and give each step only what is left.

import time


class Deadline:
    def __init__(self, seconds: float):
        self.end = time.monotonic() + seconds

    def remaining(self) -> float:
        return max(0.0, self.end - time.monotonic())

Why time.monotonic(): it never goes backward, unlike wall-clock time (time.time()), which can jump when the system clock is adjusted. Monotonic time is the correct choice for measuring durations and deadlines.

Practical Example: Answering Within a Time Budget

This function must answer within a fixed time budget, such as 20 seconds for an interactive feature. It first tries a strong model with part of the budget. If that does not finish in time, it uses the remaining time for a faster model. If no time remains, it returns a friendly fallback message, so the user always gets a response before the deadline.

"""deadline_answer.py: answer within a fixed time budget using async timeouts."""

import asyncio
import time

import anthropic
from anthropic import AsyncAnthropic

PRIMARY_MODEL = "claude-opus-5-5"
FALLBACK_MODEL = "claude-haiku-5-5"
FALLBACK_TEXT = "Sorry, I couldn't answer in time. Please try again."


class Deadline:
    def __init__(self, seconds: float):
        self.end = time.monotonic() + seconds

    def remaining(self) -> float:
        return max(0.0, self.end - time.monotonic())


async def ask(client: AsyncAnthropic, model: str, question: str) -> str:
    message = await client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": question}],
    )
    return "".join(b.text for b in message.content if b.type == "text").strip()


async def answer_within(client: AsyncAnthropic, question: str, budget: float = 20.0) -> tuple[str, str]:
    deadline = Deadline(budget)

    primary_budget = deadline.remaining() * 0.7
    primary = client.with_options(timeout=primary_budget, max_retries=0)
    try:
        text = await asyncio.wait_for(ask(primary, PRIMARY_MODEL, question), timeout=primary_budget)
        return text, PRIMARY_MODEL
    except (asyncio.TimeoutError, anthropic.APIError):
        pass

    remaining = deadline.remaining()
    if remaining < 2:
        return FALLBACK_TEXT, "none"

    fallback = client.with_options(timeout=remaining, max_retries=0)
    try:
        text = await asyncio.wait_for(ask(fallback, FALLBACK_MODEL, question), timeout=remaining)
        return text, FALLBACK_MODEL
    except (asyncio.TimeoutError, anthropic.APIError):
        return FALLBACK_TEXT, "none"


async def main() -> None:
    async with AsyncAnthropic() as client:
        for budget in (20.0, 3.0):
            start = time.perf_counter()
            text, model = await answer_within(
                client, "Explain the difference between a process and a thread in two sentences.", budget
            )
            print(f"[budget {budget:>4.1f}s | used {time.perf_counter() - start:4.1f}s | model: {model}]")
            print(text, "\n")


if __name__ == "__main__":
    asyncio.run(main())

Example output (timings vary):

[budget 20.0s | used  3.8s | model: claude-opus-5-5]
A process is an independent program with its own memory space, while a thread is a unit of execution inside a process. Threads in the same process share memory, which makes them lighter but requires care with shared data.

[budget  3.0s | used  2.6s | model: claude-haiku-5-5]
A process is a running program with its own memory; a thread is a lighter unit of work that runs inside a process and shares its memory.

How It Works

1. Deadline records the end time using time.monotonic(), and remaining() reports how much time is left, never less than zero.
2. The primary attempt gets 70% of the budget. Two limits enforce it: the SDK timeout (with_options(timeout=...)) protects the HTTP request, and asyncio.wait_for() enforces the deadline on the whole call. max_retries=0 disables automatic retries, which would otherwise break the budget.
3. Any timeout or API error on the primary attempt moves on to the fallback. Both asyncio.TimeoutError (the deadline) and anthropic.APIError (which includes the SDK's APITimeoutError) are caught.
4. The fallback uses whatever time remains, with a faster model. If less than 2 seconds remain, there is no realistic chance of an answer, so the function returns the fallback text immediately.
5. The function always returns before the deadline with something useful: an answer from the strong model, an answer from the fast model, or a friendly message. The returned model name makes this visible for logging and monitoring.
6. In the sample run, the 20-second budget was enough for the primary model. With a 3-second budget, the primary attempt was cut off after 2.1 seconds (70%), and the fast model answered in the remaining time.

Common Mistakes

Relying on the SDK Timeout as a Total Deadline

What happens: with retries, or with streams that keep sending data, operations take far longer than the timeout value.

Fix: use an asyncio deadline for total duration, and the SDK timeout for individual requests.

Forgetting That Retries Multiply Timeouts

What happens: an "8-second timeout" results in a 30-second wait.

Fix: lower max_retries for time-sensitive calls, or wrap them in a deadline.

Using time.time() for Deadlines

What happens: clock adjustments make deadlines too short or too long.

Fix: use time.monotonic().

Catching the Wrong Timeout Exception

What happens: except TimeoutError does not catch anthropic.APITimeoutError, and except anthropic.APITimeoutError does not catch asyncio's TimeoutError.

Fix: catch both where both can occur.

Very Long Timeouts Instead of Streaming

What happens: long silent requests are vulnerable to dropped connections.

Fix: stream long responses (Lesson 59).

Best Practices

• Use SDK timeouts to catch hung requests, and asyncio deadlines for total duration.
• Set timeouts per feature: short for interactive features, longer for background work.
• Reduce or disable retries inside strict deadlines.
• Share a deadline budget across multi-step operations.
• Always return something useful before the deadline, such as a fallback answer or a friendly message.
• Measure durations with time.monotonic().
• Log which path was taken (primary, fallback, or timeout) to tune budgets over time.

Exercise

1. Run deadline_answer.py with budgets of 1, 3, 10, and 30 seconds. Which model answers in each case?
2. Calculate the worst-case wait for a request with timeout=20 and max_retries=3, ignoring backoff.
3. Add a streaming version of the primary attempt that returns the partial answer, plus a note, if the deadline is reached mid-stream.
4. Explain why answer_within() sets max_retries=0 on both attempts.

Step 2: 20 × (3 + 1) = 80 seconds, plus the backoff delays between attempts.

Step 4: automatic retries happen inside the SDK, invisibly to the deadline logic. If a request timed out and the SDK retried it, the retry would consume time that the function had planned for the fallback, or exceed the overall budget. With max_retries=0, each attempt is a single request, and the deadline code decides what happens next: trying the fallback model, which is a smarter "retry" for this situation.