AsyncAnthropic은 프로세스마다 하나를 오래 써

요청마다 클라이언트를 만들지 마

비동기 클라이언트는 연결 풀을 품고 있어. FastAPI 라우트가 호출될 때마다 AsyncAnthropic()을 새로 만들면 연결 준비 비용을 반복하고 풀의 이점을 없애 버려. 애플리케이션 시작 때 한 번 만들고 상태 객체나 모듈 수준에서 공유해.

서로 독립인 호출은 함께 기다려

병렬 요약이나 여러 번의 평가처럼 한 요청 안에 독립적인 Claude 호출이 있다면 asyncio.gather로 동시에 실행할 수 있어. 공유 클라이언트가 연결을 재사용하고 호출 한도 재시도를 처리하므로, 순서가 필요 없는 작업을 일부러 줄 세울 이유가 없어. 다만 동시성 상한은 조직의 RPM과 TPM에 맞춰 따로 둬.

취소와 시간 제한을 끝까지 전달해

브라우저가 닫히거나 FastAPI가 CancelledError를 내면 진행 중인 Anthropic 요청도 취소돼야 해. SDK는 asyncio 취소를 따르며, 전체 호출 제한에는 async with asyncio.timeout()을 쓸 수 있어. 스트림 반복자를 asyncio.wait_for로 감싸 생성기를 망가뜨리지는 마.

원칙: 비동기 클라이언트는 DB 연결 풀처럼 오래 살며 공유하는 자원이야. 요청 하나의 임시 객체로 만들지 마.

Code

공유 async 클라이언트 가진 FastAPI lifespan·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이 내부에서 httpx 사용; connection pool 우아하게 닫기.
    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}

asyncio.gather로 동시 fan-out·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())

요청마다 클라이언트를 만들지 마

서로 독립인 호출은 함께 기다려

취소와 시간 제한을 끝까지 전달해

Code

External links

Exercise

Progress

댓글 0