Handle Async Errors

What Changes with Async Errors

The async client raises the same exceptions as the sync client: RateLimitError, APIConnectionError, AuthenticationError, and the rest of the hierarchy (Lesson 9). For a single awaited call, handling is identical:

import anthropic

try:
    message = await client.messages.create(...)
except anthropic.RateLimitError:
    ...
except anthropic.APIError as error:
    ...

What changes is concurrency. When many coroutines run at once, you must decide:

• What happens to the other requests when one fails?
• Where do exceptions from background tasks go?
• How do you retry failed items without restarting everything?
• How do you report many different failures clearly?

This lesson covers error handling for each concurrency tool from Lesson 74.

Errors in gather()

Default Behavior

results = await asyncio.gather(ask("Q1"), ask("Q2"), ask("Q3"))

If ask("Q2") raises, gather() raises that exception immediately. The other coroutines are not cancelled: they keep running, but their results are lost to you. This is rarely what you want for independent requests.

return_exceptions=True

results = await asyncio.gather(ask("Q1"), ask("Q2"), ask("Q3"), return_exceptions=True)

for question, result in zip(["Q1", "Q2", "Q3"], results):
    if isinstance(result, Exception):
        print(f"{question} failed: {type(result).__name__}")
    else:
        print(f"{question}: {result}")

Each position in the result list holds either the result or the exception. Every request runs to completion, and you decide afterward what to do with each failure.

Note: return_exceptions=True also returns asyncio.CancelledError as a result if a task was cancelled. Check for BaseException instead of Exception if you need to detect cancellations.

Catching Inside Each Coroutine

An alternative is to handle errors inside each coroutine and return a result object (Lesson 75):

async def safe_ask(question: str) -> tuple[str | None, str | None]:
    try:
        return await ask(question), None
    except anthropic.APIError as error:
        return None, type(error).__name__

This keeps error handling next to the request, and makes results uniform.

Errors in TaskGroup (Python 3.11+)

A TaskGroup treats its tasks as a unit. If any task fails:

1. all other tasks in the group are cancelled,
2. the group waits for them to finish cancelling, and
3. it raises an ExceptionGroup containing every exception that occurred.

Handle exception groups with except*, which matches exceptions of a given type inside a group:

try:
    async with asyncio.TaskGroup() as group:
        outline = group.create_task(ask("Outline a Django tutorial."))
        glossary = group.create_task(ask("List 10 Django terms with definitions."))
except* anthropic.RateLimitError:
    print("Rate limited. Try again later.")
except* anthropic.APIError as group_error:
    print(f"{len(group_error.exceptions)} request(s) failed.")

Errors in Background Tasks

A task created with asyncio.create_task() runs independently. If it raises and is never awaited, the exception is not raised anywhere. Python only logs a warning, "Task exception was never retrieved", when the task is garbage-collected, which is easy to miss.

Rules for background tasks:

• Keep a reference to every task.
• Await it eventually, or check task.exception() after it is done.
• For "fire-and-forget" tasks, add a done callback that logs failures:

import logging

logger = logging.getLogger("tasks")


def log_failure(task: asyncio.Task) -> None:
    if not task.cancelled() and task.exception() is not None:
        logger.error("background task failed", exc_info=task.exception())


task = asyncio.create_task(update_memory(user_id, text))
task.add_done_callback(log_failure)

Cancellation Is Not an Error

asyncio.CancelledError means "stop now": a timeout, a user disconnecting, or a shutdown. It inherits from BaseException, not Exception, so except Exception: does not catch it, which is intentional.

try:
    message = await client.messages.create(...)
except anthropic.APIError:
    ...        # Handle API problems
# CancelledError is not caught here, so it propagates and the task stops correctly

If you must clean up on cancellation, use finally, or catch CancelledError, clean up, and re-raise it.

Errors in Async Streams

Errors in async streams behave like sync stream errors (Lesson 57): failures before the stream starts are raised at async with, and mid-stream failures are raised from the async for loop. Wrap the whole async with block in try, and remember that partial text may already have been shown to the user.

Classifying Errors for Retries

Not every error deserves a retry. Group errors by what retrying can achieve:

The SDK already retries temporary errors twice per request (Lesson 10). Failures that reach your code have survived those retries, so a second pass after a pause is a reasonable recovery strategy for large jobs.

Practical Example: A Resilient Concurrent Job

This job translates a list of short texts concurrently. It collects every outcome with return_exceptions=True, classifies failures, retries only temporary failures in a second pass after a pause, stops early on configuration errors, and prints an error report.

"""resilient_job.py: concurrent requests with error classification and a retry pass."""

import asyncio
from collections import Counter

import anthropic
from anthropic import AsyncAnthropic

TEMPORARY = (anthropic.RateLimitError, anthropic.InternalServerError, anthropic.APIConnectionError)
CONFIGURATION = (anthropic.AuthenticationError, anthropic.PermissionDeniedError)
CONCURRENCY = 5

TEXTS = [
    "Welcome to the Python course!",
    "Submit your project before Friday.",
    "Your certificate is ready to download.",
    "Live sessions start at 7 PM IST.",
    "Please update your profile picture.",
]


async def translate(client: AsyncAnthropic, semaphore: asyncio.Semaphore, text: str) -> str:
    async with semaphore:
        message = await client.messages.create(
            model="claude-haiku-5-5",
            max_tokens=1024,
            messages=[{"role": "user", "content":
                       f"Translate into Tamil. Reply with only the translation.\n<text>{text}</text>"}],
        )
    return "".join(b.text for b in message.content if b.type == "text").strip()


async def run_pass(client, semaphore, items: dict[int, str]) -> dict[int, str | BaseException]:
    indexes = list(items)
    outcomes = await asyncio.gather(
        *(translate(client, semaphore, items[i]) for i in indexes), return_exceptions=True
    )
    return dict(zip(indexes, outcomes))


async def main() -> None:
    semaphore = asyncio.Semaphore(CONCURRENCY)
    pending = dict(enumerate(TEXTS, start=1))
    done: dict[int, str] = {}
    failed: dict[int, str] = {}

    async with AsyncAnthropic() as client:
        for attempt in (1, 2):
            outcomes = await run_pass(client, semaphore, pending)
            retry_later: dict[int, str] = {}

            for index, outcome in outcomes.items():
                if isinstance(outcome, CONFIGURATION):
                    raise SystemExit(f"Configuration problem ({type(outcome).__name__}). Stopping the job.")
                if isinstance(outcome, BaseException):
                    if isinstance(outcome, TEMPORARY) and attempt == 1:
                        retry_later[index] = pending[index]
                    else:
                        failed[index] = type(outcome).__name__
                else:
                    done[index] = outcome

            if not retry_later:
                break
            print(f"Retrying {len(retry_later)} temporary failure(s) after a pause...")
            await asyncio.sleep(5)
            pending = retry_later

    for index in sorted(done):
        print(f"{index}. {TEXTS[index - 1]} -> {done[index]}")
    if failed:
        print("\nFailures by type:", dict(Counter(failed.values())))
        for index, name in sorted(failed.items()):
            print(f"  {index}. {TEXTS[index - 1]!r}: {name}")


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

Example output (with one temporary failure on the first pass):

Retrying 1 temporary failure(s) after a pause...
1. Welcome to the Python course! -> பைதான் பாடத்திட்டத்திற்கு வரவேற்கிறோம்!
2. Submit your project before Friday. -> வெள்ளிக்கிழமைக்கு முன் உங்கள் திட்டத்தைச் சமர்ப்பிக்கவும்.
3. Your certificate is ready to download. -> உங்கள் சான்றிதழ் பதிவிறக்கத் தயாராக உள்ளது.
4. Live sessions start at 7 PM IST. -> நேரலை அமர்வுகள் இரவு 7 மணிக்கு (IST) தொடங்கும்.
5. Please update your profile picture. -> உங்கள் சுயவிவரப் படத்தைப் புதுப்பிக்கவும்.

How It Works

1. TEMPORARY and CONFIGURATION are tuples of exception classes. isinstance(outcome, TEMPORARY) checks against all of them at once, and also matches subclasses, such as APITimeoutError, which is a subclass of APIConnectionError.
2. run_pass() runs one pass over the pending items with gather(..., return_exceptions=True). It returns a dictionary from each item's index to its outcome: a translation or an exception. dict(zip(indexes, outcomes)) relies on gather() preserving order.
3. Classification: configuration errors stop the whole job immediately, because every other request would fail the same way. Temporary errors on the first pass are queued for retry. Everything else is recorded as a failure.
4. The retry pass runs only the temporary failures, after a 5-second pause, which gives rate limits or overloaded servers time to recover. A second temporary failure is recorded as final, so the job always ends.
5. The report prints successes in their original order, then failures grouped with Counter, so recurring problem types stand out.
6. The semaphore is shared by both passes, so concurrency never exceeds the limit, even during retries.

Common Mistakes

Plain gather() for Independent Items

What happens: the first failure raises, and every other result is lost.

Fix: use return_exceptions=True, or catch errors inside each coroutine.

Ignoring Background Task Exceptions

What happens: failures disappear, except for a late warning in the logs.

Fix: await tasks, or attach a done callback that logs failures.

Catching BaseException or CancelledError and Continuing

What happens: tasks cannot be cancelled, timeouts do not work, and shutdowns hang.

Fix: catch Exception subclasses for errors, and let cancellation propagate.

Retrying Everything

What happens: invalid requests and invalid keys are retried pointlessly, which wastes time and hides the real problem.

Fix: retry only temporary errors, and stop on configuration errors.

Unbounded Retry Loops

What happens: a persistent problem keeps the job running forever.

Fix: limit the number of retry passes.

Best Practices

• Choose the failure model deliberately: independent items with return_exceptions=True, linked steps with TaskGroup.
• Classify errors into temporary, permanent, and configuration problems.
• Retry only temporary errors, after a pause, a limited number of times.
• Stop early on configuration errors.
• Never lose background task errors.
• Let cancellation propagate, and clean up with finally.
• Report failures grouped by type.

Exercise

1. Add an invalid model name for one item and confirm it is reported as a permanent failure, not retried.
2. Rewrite run_pass() to catch errors inside translate() and return (index, translation, error) tuples instead of using return_exceptions=True.
3. On Python 3.11+, run two related requests in a TaskGroup, make one fail, and handle the ExceptionGroup with except*.
4. Add a done callback to a background task that raises an exception, and confirm that the failure is logged.

Step 1: an invalid model returns 404 not_found_error, so the request raises anthropic.NotFoundError. It is neither in TEMPORARY nor in CONFIGURATION, so it goes to failed in the first pass. The report shows {'NotFoundError': 1}.

Step 4:

import asyncio
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("tasks")


def log_failure(task: asyncio.Task) -> None:
    if not task.cancelled() and task.exception() is not None:
        logger.error("background task failed", exc_info=task.exception())


async def failing_job() -> None:
    raise ValueError("something went wrong")


async def main() -> None:
    task = asyncio.create_task(failing_job())
    task.add_done_callback(log_failure)
    await asyncio.sleep(0.1)          # Give the task time to run


asyncio.run(main())

The callback runs when the task finishes, and logs the exception with its traceback. Calling task.exception() also marks the exception as retrieved, so Python does not print its "never retrieved" warning.