FastAPI와 long-running worker에서 AsyncAnthropic

요청당 X, 프로세스당 클라이언트 하나

Async 클라이언트는 connection pool 유지. 매 FastAPI 라우트에서 새 AsyncAnthropic() 만들면 connection setup 오버헤드 태우고 pool 효과 막아. 시작 시 한 번 빌드, app state나 모듈 레벨 싱글톤에 두고, 어디서든 재사용.

asyncio.gather로 동시 호출

한 요청이 여러 Claude 호출 필요할 때(병렬 요약, multi-shot 평가, fan-out), asyncio.gather가 동시 실행. SDK async 클라이언트가 connection 재사용·rate-limit backoff을 gather 아래서 깔끔히 처리 — fan-out에 페널티 없어.

Cancellation·timeout 제대로

클라이언트의 upstream HTTP 요청이 canceled되면(브라우저 닫힘, FastAPI CancelledError raise), in-flight Anthropic 호출도 cancel되길 원하지. SDK가 asyncio cancellation 존중. 전체 호출 timeout엔 async with asyncio.timeout(); stream iteration을 asyncio.wait_for로 감싸지 마(generator 죽임 — Messages 트랙 참고).

원칙: Async 클라이언트는 오래 살고 공유되길 원해. 클라이언트를 DB connection pool처럼 다뤄, request-scoped 객체 X.

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())

FastAPI와 long-running worker에서 AsyncAnthropic

요청당 X, 프로세스당 클라이언트 하나

asyncio.gather로 동시 호출

Cancellation·timeout 제대로

Code

External links

Exercise

Progress

댓글 0