AsyncAnthropic in FastAPI and Long-Running Workers

One client per process, not per request

The async client maintains a connection pool. Constructing a new AsyncAnthropic() inside every FastAPI route burns connection setup overhead and prevents the pool from being effective. Build the client once at startup, store it on the app state or a module-level singleton, and reuse it everywhere.

Concurrent calls with asyncio.gather

When a single request needs multiple Claude calls (parallel summarization, multi-shot evaluation, fan-out), asyncio.gather runs them concurrently. The SDK's async client handles connection reuse and rate-limit backoff cleanly under gather — you are not penalized for fanning out.

Cancellation and timeouts done right

If the upstream HTTP request from your client is canceled (browser closes, FastAPI raises CancelledError), you want the in-flight Anthropic call canceled too. The SDK respects asyncio cancellation. Use async with asyncio.timeout() for whole-call timeouts; do not wrap stream iteration in asyncio.wait_for (it kills the generator, see Messages track).

Principle: Async clients want to be long-lived and shared. Treat the client like a database connection pool, not a request-scoped object.

Code

FastAPI lifespan with a shared async client·python

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from anthropic import AsyncAnthropic

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.claude = AsyncAnthropic()
    yield
    # AsyncAnthropic uses httpx under the hood; close the connection pool gracefully.
    await app.state.claude.close()

app = FastAPI(lifespan=lifespan)

@app.post("/summarize")
async def summarize(request: Request, body: dict):
    client: AsyncAnthropic = request.app.state.claude
    resp = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        messages=[{"role": "user", "content": body["text"]}],
    )
    return {"summary": resp.content[0].text}

Concurrent fan-out with asyncio.gather·python

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def classify(text: str) -> str:
    r = await client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=32,
        system='Reply with one word: positive, negative, or neutral.',
        messages=[{"role": "user", "content": text}],
    )
    return r.content[0].text.strip()

async def main():
    async with asyncio.timeout(30):
        labels = await asyncio.gather(
            classify("I love this"),
            classify("meh"),
            classify("this is awful"),
        )
    print(labels)

asyncio.run(main())

AsyncAnthropic in FastAPI and Long-Running Workers

One client per process, not per request

Concurrent calls with asyncio.gather

Cancellation and timeouts done right

Code

External links

Exercise

Progress

Comments 0