| name | python-anthropic-sdk |
| description | Anthropic Python SDK(`anthropic` 패키지) — Python 백엔드에서 Claude API를 호출하는 방법. 동기/비동기 클라이언트, Messages API, 스트리밍, 프롬프트 캐싱(5m/1h), 도구 사용, 비전 입력, 에러 핸들링·재시도, 토큰 카운팅, AWS Bedrock / GCP Vertex 변형까지 다룬다. <example>사용자: "FastAPI에서 Claude API를 비동기로 호출하면서 스트리밍하고 싶다"</example> <example>사용자: "긴 시스템 프롬프트를 캐시해서 비용을 줄이고 싶다"</example> <example>사용자: "Claude API 응답을 도구 호출로 받아 JSON 형태로 처리하고 싶다"</example>
|
Python Anthropic SDK — Claude API 통합 가이드 (Python 백엔드)
소스:
짝 스킬 (Companion Skills)
backend/python-fastapi — Python 백엔드 프레임워크backend/python-async-asyncio — 비동기 동작 (AsyncAnthropic 사용 시 필수 이해)frontend/claude-api-streaming-frontend — 프론트엔드에서 SSE 스트림 수신 측
이 스킬은 Python 백엔드 측 통합만 다룬다. 프론트엔드 측 SSE 처리는 짝 스킬을 참조한다.
1. 설치
pip install anthropic
uv add anthropic
pip install "anthropic[bedrock]"
pip install "anthropic[vertex]"
pip install "anthropic[aws]"
pip install "anthropic[aiohttp]"
Python 요구 버전: 3.9 이상.
설치 후 버전 확인:
import anthropic
print(anthropic.__version__)
2. 클라이언트 생성
2.1 동기 클라이언트 (Anthropic)
import os
from anthropic import Anthropic
client = Anthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
2.2 비동기 클라이언트 (AsyncAnthropic)
import os
import asyncio
from anthropic import AsyncAnthropic
client = AsyncAnthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
async def main() -> None:
message = await client.messages.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-5",
)
print(message.content)
asyncio.run(main())
2.3 환경변수 관리 (보안 필수)
- API 키를 코드에 하드코딩하지 않는다.
ANTHROPIC_API_KEY 환경변수에 둔다.
- 로컬 개발은
.env + python-dotenv 권장 (.env 는 .gitignore에 추가).
- 컨테이너/서버 배포는 시크릿 매니저(AWS Secrets Manager, GCP Secret Manager, Kubernetes Secrets) 사용.
- 로그에 API 키가 찍히지 않도록 검토.
Authorization 헤더는 SDK가 자동 처리하므로 직접 다루지 않는다.
from dotenv import load_dotenv
load_dotenv()
from anthropic import Anthropic
client = Anthropic()
3. Messages API — 기본 호출
message = client.messages.create(
model="claude-opus-5",
max_tokens=1024,
messages=[
{"role": "user", "content": "Hello, Claude"},
],
)
print(message.content)
print(message.usage)
print(message._request_id)
주요 파라미터:
| 파라미터 | 필수 | 설명 |
|---|
model | ✅ | 모델 ID (다음 섹션 참조) |
max_tokens | ✅ | 생성할 최대 출력 토큰 |
messages | ✅ | `{"role": "user" |
system | | 시스템 프롬프트(문자열 또는 텍스트 블록 배열) |
thinking | | {"type": "adaptive"} — Opus 5는 기본 ON(생략 시 adaptive) |
output_config | | {"effort": "low"|"medium"|"high"|"xhigh"|"max"} — 사고 깊이·토큰 사용량 조절 |
tools | | 도구 정의 배열 (섹션 6) |
stream | | True 시 SSE 스트림 반환 |
주의 — 5 계열(Opus 5·Sonnet 5·Fable 5)과 Opus 4.7/4.8에서 제거된 파라미터:
temperature / top_p / top_k → 400 에러. 제거하고 프롬프팅으로 출력 성향을 유도한다.thinking: {"type": "enabled", "budget_tokens": N} → 400 에러. {"type": "adaptive"} + output_config.effort로 대체한다.- 마지막 assistant 턴 prefill → 400 에러.
output_config.format(structured outputs) 또는 시스템 프롬프트로 대체한다.
Opus 5 고유 규약:
- 사고가 기본 ON이다(파라미터 생략 시 adaptive).
max_tokens는 사고 + 응답 텍스트 합산 상한이므로, 기존에 사고 없이 돌던 경로는 max_tokens를 늘리지 않으면 응답이 잘린다.
thinking: {"type": "disabled"}는 effort high 이하에서만 허용된다. xhigh/max와 함께 쓰면 400 에러.thinking.display 기본값은 "omitted"(사고 텍스트가 빈 문자열). 사용자에게 추론 요약을 보여주려면 {"type": "adaptive", "display": "summarized"}를 명시한다.
message = client.messages.create(
model="claude-opus-5",
max_tokens=16000,
thinking={"type": "adaptive", "display": "summarized"},
output_config={"effort": "high"},
messages=[{"role": "user", "content": "단계적으로 분석해줘"}],
)
4. 스트리밍
스트리밍은 두 가지 방식이 있다. 거의 대부분은 스트리밍 헬퍼(messages.stream()) 가 권장된다.
4.1 스트리밍 헬퍼 (권장) — sync
from anthropic import Anthropic
client = Anthropic()
with client.messages.stream(
model="claude-opus-5",
max_tokens=1024,
messages=[{"role": "user", "content": "Say hello there!"}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print()
final = stream.get_final_message()
print(final.to_json())
4.2 스트리밍 헬퍼 — async
import asyncio
from anthropic import AsyncAnthropic
client = AsyncAnthropic()
async def main() -> None:
async with client.messages.stream(
model="claude-opus-5",
max_tokens=1024,
messages=[{"role": "user", "content": "Say hello there!"}],
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True)
final = await stream.get_final_message()
print(final.to_json())
asyncio.run(main())
4.3 원시 이벤트 스트림 — stream=True
메모리를 더 적게 쓰고 누적 객체를 만들지 않는다. 직접 이벤트 타입을 처리해야 한다.
stream = client.messages.create(
model="claude-opus-5",
max_tokens=1024,
messages=[{"role": "user", "content": "Hello"}],
stream=True,
)
for event in stream:
print(event.type)
4.4 이벤트 종류 (Server-Sent Events)
| 이벤트 | 의미 |
|---|
message_start | 응답 시작, 빈 Message 객체 포함 |
content_block_start | 콘텐츠 블록 시작 (text / tool_use 등) |
content_block_delta | 블록 내 증분 (text_delta, input_json_delta) |
content_block_stop | 콘텐츠 블록 종료 |
message_delta | top-level 메시지 변경 (stop_reason, usage 등) |
message_stop | 응답 종료 |
ping | 연결 유지 신호 |
주의: 직접 stream=True를 처리할 때 content_block_delta의 delta.type이 text_delta인지 input_json_delta(tool use)인지 분기해야 한다. 헬퍼를 쓰면 SDK가 자동 처리한다.
4.5 FastAPI에서 스트리밍 응답 전달 (짝 스킬)
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from anthropic import AsyncAnthropic
app = FastAPI()
client = AsyncAnthropic()
@app.post("/chat")
async def chat(prompt: str):
async def event_generator():
async with client.messages.stream(
model="claude-opus-5",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
) as stream:
async for text in stream.text_stream:
yield f"data: {text}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(event_generator(), media_type="text/event-stream")
프론트엔드 측 EventSource/Fetch 스트림 처리는 frontend/claude-api-streaming-frontend 스킬을 참조한다.
5. 프롬프트 캐싱 (Prompt Caching)
긴 시스템 프롬프트, 문서, 도구 정의를 캐싱해 비용·지연을 절감한다.
5.1 기본 사용
response = client.messages.create(
model="claude-opus-5",
max_tokens=1024,
system=[
{
"type": "text",
"text": "당신은 법률 문서 분석 전문가입니다.",
},
{
"type": "text",
"text": "여기 50페이지 계약서 전문: ...",
"cache_control": {"type": "ephemeral"},
},
],
messages=[
{"role": "user", "content": "핵심 조항을 정리해줘"},
],
)
print(response.usage)
5.2 TTL — 5분(기본) vs 1시간
"cache_control": {"type": "ephemeral"}
"cache_control": {"type": "ephemeral", "ttl": "1h"}
| TTL | 캐시 쓰기 비용 | 사용 시점 |
|---|
5m (기본) | 베이스 입력의 +25% | 빈번한 재호출 (몇 분 내) |
1h | 베이스 입력의 ×2 | 5분 초과 ~ 1시간 이내 재호출, 지연 민감 워크로드 |
캐시 읽기는 두 TTL 모두 베이스 가격의 10%.
5.3 배치 위치 — tools → system → messages 순으로 무효화 전파
캐시 키는 cache_control 블록 이전의 모든 콘텐츠를 포함한다. 따라서:
- 변하지 않는 블록 뒤에 배치 (마지막 안정 블록)
- 변하는 값(타임스탬프, 사용자 ID 등) 뒤에는 캐시 브레이크 두지 않기
system=[
{"type": "text", "text": "정적 지시문..."},
{
"type": "text",
"text": f"현재 시각: {datetime.now()}",
"cache_control": {"type": "ephemeral"},
},
]
system=[
{"type": "text", "text": "정적 지시문...", "cache_control": {"type": "ephemeral"}},
{"type": "text", "text": f"현재 시각: {datetime.now()}"},
]
5.4 최소 캐시 토큰
주의: 최소 토큰 미만이면 캐시가 적용되지 않고 에러 없이 일반 요청으로 처리된다. cache_creation_input_tokens와 cache_read_input_tokens가 모두 0이면 캐싱이 일어나지 않은 것.
| 모델 | 최소 캐시 토큰 |
|---|
| Claude Opus 5, Fable 5 | 512 |
| Claude Opus 4.8, Sonnet 5, Sonnet 4.6 / 4.5 | 1,024 |
| Claude Opus 4.7 | 2,048 |
| Claude Opus 4.6 / 4.5 | 4,096 |
| Claude Haiku 4.5 | 4,096 |
주의: 최소 캐시 토큰은 세대 순으로 단조 감소하지 않는다. Opus 5는 512로 가장 낮지만
Opus 4.8은 1,024, Opus 4.7은 2,048, Opus 4.6은 4,096이다. 모델을 바꾸면 캐시 임계값도
다시 확인해야 한다. Opus 4.8 → Opus 5로 옮기면 임계값이 절반(1,024 → 512)이 되므로,
기존에 "너무 짧아서 캐시 안 된다"고 판단했던 프롬프트가 코드 변경 없이 캐시될 수 있다.
5.5 TTL 혼합 규칙
같은 요청에 5m와 1h를 섞을 수 있지만, 1h가 5m 앞에 와야 한다.
6. 도구 사용 (Tool Use)
6.1 명시적 도구 정의
tools = [
{
"name": "get_weather",
"description": "주어진 도시의 현재 날씨를 조회합니다.",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "도시명 (예: Seoul)",
},
},
"required": ["location"],
},
}
]
response = client.messages.create(
model="claude-opus-5",
max_tokens=1024,
tools=tools,
messages=[{"role": "user", "content": "서울 날씨 알려줘"}],
)
while response.stop_reason == "tool_use":
tool_use = next(b for b in response.content if b.type == "tool_use")
result = handle_tool(tool_use.name, tool_use.input)
response = client.messages.create(
model="claude-opus-5",
max_tokens=1024,
tools=tools,
messages=[
{"role": "user", "content": "서울 날씨 알려줘"},
{"role": "assistant", "content": response.content},
{
"role": "user",
"content": [
{
: ,
: tool_use.,
: result,
}
],
},
],
)
6.2 @beta_tool 데코레이터 (간편)
주의: 베타 기능. 시그니처가 변할 수 있으므로 SDK 버전 핀 권장.
import json
from anthropic import Anthropic, beta_tool
client = Anthropic()
@beta_tool
def get_weather(location: str) -> str:
"""도시의 현재 날씨를 반환합니다.
Args:
location: 도시명, 예: Seoul
Returns:
JSON 문자열 (location, temperature, condition).
"""
return json.dumps({"location": location, "temperature": "20°C", "condition": "Clear"})
runner = client.beta.messages.tool_runner(
model="claude-opus-5",
max_tokens=1024,
tools=[get_weather],
messages=[{"role": "user", "content": "서울 날씨?"}],
)
for message in runner:
print(message)
6.3 구조화된 JSON 출력 강제
도구를 단일로 지정하고 tool_choice로 강제 호출하면 JSON 응답을 강제할 수 있다.
tools = [{
"name": "extract_info",
"description": "텍스트에서 이름과 이메일을 추출",
"input_schema": {
"type": "object",
"properties": {
"name": {"type": "string"},
"email": {"type": "string"},
},
"required": ["name", "email"],
},
}]
response = client.messages.create(
model="claude-opus-5",
max_tokens=512,
tools=tools,
tool_choice={"type": "tool", "name": "extract_info"},
messages=[{"role": "user", "content": "홍길동, hong@example.com"}],
)
tool_use = next(b for b in response.content if b.type == "tool_use")
print(tool_use.input)
상세 레퍼런스 (예제·고급 패턴·흔한 실수) → references/REFERENCE.md
