Choose Sync or Async

The Decision

The Anthropic SDK offers two clients with identical capabilities:

Neither is "better". The right choice depends on where your code runs and how many things it does at once. This lesson gives you a practical way to decide, explains how to get concurrency without async, and shows how to combine both styles safely.

Start with the Environment

The most important question is what kind of code surrounds your Claude calls:

If the surrounding code is already async, use the async client. Mixing the sync client into async code blocks the event loop and slows down everything.

Then Consider Concurrency Needs

Concurrency Without Async: Thread Pools

Synchronous code can run requests concurrently with threads. Python's concurrent.futures.ThreadPoolExecutor runs a function in several threads. While one thread waits for Claude, the others continue, which gives the same overlap of waiting time as async, without changing the rest of your code:

from concurrent.futures import ThreadPoolExecutor

from anthropic import Anthropic

client = Anthropic()   # One client, shared by all threads


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


with ThreadPoolExecutor(max_workers=5) as pool:
    answers = list(pool.map(ask, questions))   # Results in input order

• max_workers=5 limits concurrency to five threads, like a semaphore.
• pool.map() returns results in the same order as the inputs.
• A single sync client can be shared across threads.

Threads vs. Async

Combining Sync and Async Code

Real projects sometimes need both styles. Use these bridges:

Calling Async Code from Sync Code

Start an event loop with asyncio.run(), once, at the boundary:

def summarize_all(texts: list[str]) -> list[str]:
    return asyncio.run(summarize_concurrently(texts))   # Sync function, async inside

Do not call asyncio.run() inside code that is already running in an event loop (Lesson 71).

Calling Sync Code from Async Code

Run blocking code in a thread with asyncio.to_thread(), so the event loop stays free (Lesson 74):

history = await asyncio.to_thread(load_history_from_sqlite, session_id)

Django: Sync and Async Views

Django supports both:

In async Django views, database access must use Django's async ORM methods (such as aget() and acreate()) or sync_to_async(), so blocking calls do not freeze the event loop.

A Decision Checklist

1. Is the surrounding framework async? Use AsyncAnthropic.
2. Is the code synchronous, with one request at a time? Use Anthropic.
3. Sync code that needs a little parallelism? Anthropic with a ThreadPoolExecutor.
4. Hundreds of concurrent requests or streams? AsyncAnthropic, with a semaphore and retries (Lesson 75).
5. Thousands of items, results not needed right away? The Message Batches API.
6. Unsure? Start with sync. Moving to async later is a contained change, especially if your Claude logic lives in one module (Lesson 61).

Practical Example: The Same Job, Three Ways

This program answers the same six questions sequentially, with a thread pool, and with asyncio, using the same concurrency limit for the last two, and compares the times.

"""compare_approaches.py: sequential, thread pool, and asyncio versions of the same job."""

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

from anthropic import Anthropic, AsyncAnthropic

MODEL = "claude-haiku-5-5"
CONCURRENCY = 3
QUESTIONS = [
    "What is a Python wheel? One sentence.",
    "What is a lambda function? One sentence.",
    "What does 'if __name__ == \"__main__\"' do? One sentence.",
    "What is a list slice? One sentence.",
    "What is the GIL? One sentence.",
    "What is an f-string? One sentence.",
]


def text_of(message) -> str:
    return "".join(b.text for b in message.content if b.type == "text").strip()


def run_sequential() -> list[str]:
    client = Anthropic()
    return [text_of(client.messages.create(model=MODEL, max_tokens=1024,
                                           messages=[{"role": "user", "content": q}]))
            for q in QUESTIONS]


def run_threads() -> list[str]:
    client = Anthropic()

    def ask(question: str) -> str:
        return text_of(client.messages.create(model=MODEL, max_tokens=1024,
                                              messages=[{"role": "user", "content": question}]))

    with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool:
        return list(pool.map(ask, QUESTIONS))


async def run_async() -> list[str]:
    semaphore = asyncio.Semaphore(CONCURRENCY)
    async with AsyncAnthropic() as client:

        async def ask(question: str) -> str:
            async with semaphore:
                message = await client.messages.create(model=MODEL, max_tokens=1024,
                                                       messages=[{"role": "user", "content": question}])
            return text_of(message)

        return await asyncio.gather(*(ask(q) for q in QUESTIONS))


def timed(label: str, function) -> list[str]:
    start = time.perf_counter()
    answers = function()
    print(f"{label:<12} {time.perf_counter() - start:5.1f}s")
    return answers


if __name__ == "__main__":
    timed("sequential", run_sequential)
    timed("threads", run_threads)
    answers = timed("asyncio", lambda: asyncio.run(run_async()))
    print("\nFirst answer:", answers[0])

Example output (timings vary):

sequential    9.6s
threads       3.4s
asyncio       3.3s

First answer: A Python wheel is a built package format (.whl) that installs quickly without a build step.

How It Works

1. run_sequential() sends one request at a time, so its time is the sum of all six.
2. run_threads() shares one sync client across three worker threads. pool.map() runs ask for each question, at most three at a time, and returns answers in input order.
3. run_async() uses one async client and a semaphore of three, with asyncio.gather() (Lessons 74 and 75). The ask coroutine is defined inside the async with block so it can use the client.
4. timed() runs each approach and prints its duration. The async version is started with asyncio.run(), wrapped in a lambda so timed() can call it like the others.
5. Results: with the same concurrency limit, threads and asyncio take about the same time, roughly a third of the sequential time with three-way concurrency. The difference between them appears at higher concurrency, where async scales with less overhead, and in async environments, where only async fits.

Common Mistakes

Using the Sync Client in Async Frameworks

What happens: each request blocks the event loop, and the server handles one user at a time.

Fix: use AsyncAnthropic in async code.

Rewriting a Sync Application in Async Without Need

What happens: large changes and new bugs, with no real benefit for one-request-at-a-time code.

Fix: keep sync code sync; add a thread pool if you need some parallelism.

Unlimited Threads or Coroutines

What happens: rate-limit errors and wasted resources.

Fix: limit concurrency with max_workers or a semaphore.

Calling asyncio.run() from Inside an Event Loop

What happens: RuntimeError: asyncio.run() cannot be called from a running event loop.

Fix: use await inside async code, and asyncio.run() only at the sync boundary.

Blocking Calls Hidden Inside Async Code

What happens: a synchronous database or file call inside a coroutine stalls all concurrent work.

Fix: use async libraries, or asyncio.to_thread().

Best Practices

• Follow your environment: async frameworks need the async client; sync code is usually best with the sync client.
• Default to sync for scripts, tools, and notebooks.
• Use thread pools for modest parallelism in sync code.
• Use async for high concurrency and many simultaneous streams.
• Use the Batches API for large jobs that can wait.
• Bridge carefully: asyncio.run() at sync boundaries, asyncio.to_thread() for blocking calls in async code.
• Keep Claude logic in one module, so switching styles later is easy.

Exercise

1. Run compare_approaches.py with CONCURRENCY set to 1, 3, and 6. How do the threads and asyncio times change?
2. Use the decision checklist to choose a client for: a Django admin action that summarizes 20 records; a FastAPI chat endpoint; a nightly job that classifies 50,000 reviews; and a Jupyter notebook experiment.
3. Rewrite run_threads() with concurrent.futures.as_completed() to print each answer as it finishes.
4. Explain why max_workers in a ThreadPoolExecutor acts as a concurrency limit, similar to a semaphore.

Step 2:

• Django admin action, 20 records: sync code (Django view), modest parallelism, so use the sync client with a ThreadPoolExecutor, or the async client in an async view.
• FastAPI chat endpoint: an async framework, so use AsyncAnthropic (or the sync client in a def endpoint, Lesson 68).
• Nightly job, 50,000 reviews: a large, non-urgent job, so use the Message Batches API.
• Jupyter notebook experiment: the sync client for simplicity, unless you are experimenting with concurrency.

Step 4: a ThreadPoolExecutor creates at most max_workers threads. Each thread runs one task at a time, and additional tasks wait in the executor's internal queue until a thread becomes free. At most max_workers requests are therefore in flight at once, the same effect as a semaphore with that many slots.