Python SDK
Python SDK
Anthropic Python SDK는 Python 애플리케이션에서 Claude API에 편리하게 접근할 수 있게 해 줘요. 동기·비동기 연산, 스트리밍, 그리고 Amazon Bedrock, Claude Platform on AWS, Google Cloud, Microsoft Foundry와의 통합을 지원해요.
참고: 코드 예제가 있는 API 기능 문서는 API 레퍼런스를 참고해 주세요. 이 페이지는 Python 전용 SDK 기능과 구성을 다뤄요.
출처: 문서
본문
설치
pip install anthropic
플랫폼별 통합이나 비동기 성능 향상이 필요하면 extras로 설치할 수 있어요.
# Amazon Bedrock 지원
pip install "anthropic[bedrock]"
# Google Cloud 지원
pip install "anthropic[vertex]"
# Claude Platform on AWS 지원
pip install "anthropic[aws]"
# Microsoft Foundry 지원은 기본 패키지에 포함
# aiohttp로 비동기 성능 향상
pip install "anthropic[aiohttp]"
요구 사항
Python 3.10 이상이 필요해요. SDK의 0.x 릴리스에서 업그레이드하는 경우, 비호환 변경 목록은 v1 마이그레이션 가이드를 참고해 주세요.
사용법
import os
from anthropic import Anthropic
client = Anthropic(
# This is the default and can be omitted
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
message = client.messages.create(
max_tokens=1024,
messages=[
{
"role": "user",
"content": "Hello, Claude",
}
],
model="claude-opus-5-5",
)
for block in message.content:
if block.type == "text":
print(block.text)
팁: python-dotenv를 써서
.env파일에ANTHROPIC_API_KEY="my-anthropic-api-key"를 추가하면 API 키가 소스 제어에 저장되는 걸 막을 수 있어요.
Workload Identity Federation을 포함한 인증 옵션은 인증을 참고해 주세요. API 키가 여러 워크스페이스에 접근할 수 있는 개인 또는 서비스 계정 키라면 anthropic-workspace-id 요청 헤더에 워크스페이스 ID를 설정해 주세요. 워크스페이스 선택에서 이 SDK의 요청별 옵션을 보여줘요.
비동기 사용
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-5",
)
print(message.content)
asyncio.run(main())
aiohttp로 더 나은 동시성
비동기 성능을 높이려면 기본 httpx2 대신 aiohttp HTTP 백엔드를 쓸 수 있어요.
import os
import asyncio
from anthropic import AsyncAnthropic, DefaultAioHttpClient
async def main() -> None:
async with AsyncAnthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY"),
http_client=DefaultAioHttpClient(),
) as client:
message = await client.messages.create(
max_tokens=1024,
messages=[
{
"role": "user",
"content": "Hello, Claude",
}
],
model="claude-opus-5-5",
)
print(message.content)
asyncio.run(main())
스트리밍 응답
SDK는 Server-Sent Events(SSE)를 사용한 스트리밍 응답을 지원해요.
client = Anthropic()
stream = client.messages.create(
max_tokens=1024,
messages=[
{
"role": "user",
"content": "Hello, Claude",
}
],
model="claude-opus-5-5",
stream=True,
)
for event in stream:
print(event.type)
비동기 클라이언트는 정확히 같은 인터페이스를 사용해요.
client = AsyncAnthropic()
stream = await client.messages.create(
max_tokens=1024,
messages=[
{
"role": "user",
"content": "Hello, Claude",
}
],
model="claude-opus-5-5",
stream=True,
)
async for event in stream:
print(event.type)
스트리밍 헬퍼
SDK는 컨텍스트 매니저를 사용하고 누적된 텍스트와 최종 메시지에 접근할 수 있는 스트리밍 헬퍼도 제공해요.
async def main() -> None:
async with client.messages.stream(
max_tokens=1024,
messages=[
{
"role": "user",
"content": "Say hello there!",
}
],
model="claude-opus-5-5",
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True)
print()
message = await stream.get_final_message()
print(message.to_json())
asyncio.run(main())
client.messages.stream(...)으로 스트리밍하면 누적, SDK 전용 이벤트 등 다양한 헬퍼를 쓸 수 있어요.
대안으로 client.messages.create(..., stream=True)는 스트림의 이벤트 iterable만 반환하고 메모리를 덜 써요. 최종 메시지 객체를 만들어 주지는 않거든요.
토큰 카운팅
주어진 요청의 정확한 사용량은 usage 응답 속성에서 볼 수 있어요.
message = client.messages.create(...)
print(message.usage)
# Usage(input_tokens=25, output_tokens=13)
요청을 보내기 전에 토큰 수도 셀 수 있어요.
count = client.messages.count_tokens(
model="claude-opus-5-5", messages=[{"role": "user", "content": "Hello, world"}]
)
print(count.input_tokens) # 10
도구 사용
이 SDK는 함수 호출이라고도 하는 도구 사용을 지원해요. 자세한 내용은 Claude와 도구 사용을 참고해 주세요.
도구 헬퍼
SDK는 순수 Python 함수로 도구를 정의하고 실행하는 헬퍼를 제공해요. @beta_tool 데코레이터가 함수 시그니처와 docstring에서 도구 스키마를 생성해 줘요.
import json
from anthropic import Anthropic, beta_tool
client = Anthropic()
@beta_tool
def get_weather(location: str) -> str:
"""Get the weather for a given location.
Args:
location: The city and state, for example, San Francisco, CA
Returns:
A JSON-encoded string with the location, temperature, and weather condition.
"""
return json.dumps(
{
"location": location,
"temperature": "68°F",
"condition": "Sunny",
}
)
# Use the tool_runner to automatically handle tool calls
runner = client.beta.messages.tool_runner(
max_tokens=1024,
model="claude-opus-5-5",
tools=[get_weather],
messages=[
{"role": "user", "content": "What is the weather in SF?"},
],
)
for message in runner:
print(message)
매 반복마다 API 요청이 발생해요. 응답에 주어진 도구 중 하나에 대한 호출이 포함되면 도구가 자동으로 호출되고, 결과는 다음 반복에서 모델로 직접 반환돼요.
메시지 배치
이 SDK는 client.messages.batches 아래에서 배치 처리를 지원해요.
배치 만들기
Message Batches는 요청 배열을 받아요. 각 객체는 custom_id 식별자와 표준 Messages API와 같은 params를 가져요.
client.messages.batches.create(
requests=[
{
"custom_id": "my-first-request",
"params": {
"model": "claude-opus-5-5",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "Hello, world"}],
},
},
{
"custom_id": "my-second-request",
"params": {
"model": "claude-opus-5-5",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "Hi again, friend"}],
},
},
]
)
배치 결과 가져오기
Message Batch가 처리되면(.processing_status == 'ended'로 표시), .batches.results()로 결과에 접근할 수 있어요.
client = anthropic.Anthropic()
batch_id = "batch_abc123"
result_stream = client.messages.batches.results(batch_id)
for entry in result_stream:
if entry.result.type == "succeeded":
print(entry.result.message.content)
파일 업로드
파일 업로드에 해당하는 요청 파라미터는 다양한 형태로 전달할 수 있어요.
PathLike객체 (예:pathlib.Path)(filename, content, content_type)튜플BinaryIO파일류 객체
from pathlib import Path
from anthropic import Anthropic
client = Anthropic()
# 파일 경로로 업로드
client.files.upload(
file=Path("/path/to/file"),
)
# 바이트로 업로드
client.files.upload(
file=("file.txt", b"my bytes", "text/plain"),
)
비동기 클라이언트는 정확히 같은 인터페이스를 사용해요. PathLike 인스턴스를 넘기면 파일 내용이 자동으로 비동기로 읽혀요.
오류 처리
라이브러리가 API에 연결할 수 없거나 API가 성공이 아닌 상태 코드(즉 4xx 또는 5xx 응답)를 반환하면 APIError의 하위 클래스가 발생해요.
import anthropic
# ...
try:
message = client.messages.create(
max_tokens=1024,
messages=[
{
"role": "user",
"content": "Hello, Claude",
}
],
model="claude-opus-5-5",
)
except anthropic.APIConnectionError as e:
print("The server could not be reached")
print(e.__cause__) # an underlying Exception, likely raised within httpx2
except anthropic.RateLimitError as e:
print("A 429 status code was received; we should back off a bit.")
except anthropic.APIStatusError as e:
print("Another non-200-range status code was received")
print(e.status_code)
print(e.response)
오류 코드는 다음과 같아요.
| 상태 코드 | 오류 타입 |
|---|---|
| 400 | BadRequestError |
| 401 | AuthenticationError |
| 403 | PermissionDeniedError |
| 404 | NotFoundError |
| 409 | ConflictError |
| 422 | UnprocessableEntityError |
| 429 | RateLimitError |
| >=500 | InternalServerError |
| N/A | APIConnectionError |
요청 ID
요청 디버깅에 대한 자세한 내용은 Request ID를 참고해 주세요.
SDK의 모든 객체 응답은 request-id 응답 헤더에서 가져온 _request_id 속성을 제공해요. 실패한 요청을 빠르게 로깅하고 Anthropic에 보고할 수 있죠.
message = client.messages.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-5-5",
)
print(message._request_id) # e.g., req_018EeWyXxfu5pfWkrYcMdjWG
참고: 다른
_접두사 속성과 달리_request_id속성은 공개예요. 별도로 문서화되지 않는 한, 다른 모든_접두사 속성·메서드·모듈은 비공개예요.
재시도
특정 오류는 기본적으로 짧은 지수 백오프로 2번 자동 재시도해요. 연결 오류(예: 네트워크 연결 문제), 408 Request Timeout, 409 Conflict, 429 Rate Limit, >=500 Internal 오류가 기본으로 재시도돼요.
max_retries 옵션으로 구성하거나 비활성화할 수 있어요.
# 모든 요청의 기본값 구성:
client = Anthropic(
max_retries=0, # default is 2
)
# 또는 요청별로 구성:
client.with_options(max_retries=5).messages.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-5-5",
)
타임아웃
기본적으로 요청은 10분 후 타임아웃돼요. timeout 옵션으로 구성할 수 있고, float 또는 httpx2.Timeout 객체를 받아요.
import httpx2
from anthropic import Anthropic
# 모든 요청의 기본값 구성:
client = Anthropic(
timeout=20.0, # 20 seconds (default is 10 minutes)
)
# 더 세밀한 제어:
client = Anthropic(
timeout=httpx2.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)
# 요청별로 덮어쓰기:
client.with_options(timeout=5.0).messages.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-5-5",
)
타임아웃 시 SDK는 APITimeoutError를 던져요.
타임아웃된 요청은 기본적으로 2번 재시도된다는 점을 기억해 주세요.
긴 요청
경고: 오래 걸리는 요청은 스트리밍 Messages API를 사용하는 걸 고려해 보세요.
스트리밍 없이 큰 max_tokens 값을 설정하는 건 피하는 게 좋아요. 일부 네트워크는 일정 시간 후 유휴 연결을 끊을 수 있어서, 요청이 실패하거나 Anthropic 응답 없이 타임아웃될 수 있어요.
SDK는 비스트리밍 요청이 약 10분 이상 걸릴 것으로 예상되면 ValueError를 던져요. stream=True를 넘기거나 클라이언트·요청 수준에서 timeout 옵션을 덮어쓰면 이 오류가 비활성화돼요.
비스트리밍 요청에서 예상 요청 지연이 타임아웃보다 길면 클라이언트가 연결을 종료하고 응답을 받지 않고 재시도해요.
SDK는 일부 네트워크에서 유휴 연결 타임아웃의 영향을 줄이기 위해 TCP 소켓 keep-alive 옵션을 설정해요. 커스텀 http_client 옵션을 클라이언트에 넘기면 덮어쓸 수 있어요.
자동 페이지네이션
Claude API의 목록 메서드는 페이지네이션돼요. for 구문으로 모든 페이지의 항목을 반복할 수 있어요.
client = Anthropic()
all_batches = []
# Automatically fetches more pages as needed.
for batch in client.messages.batches.list(limit=20):
all_batches.append(batch)
print(all_batches)
비동기 반복은 다음과 같아요.
async def main() -> None:
all_batches = []
async for batch in client.messages.batches.list(limit=20):
all_batches.append(batch)
print(all_batches)
asyncio.run(main())
페이지를 더 세밀하게 제어하려면 .has_next_page(), .next_page_info(), .get_next_page() 메서드를 쓸 수도 있어요.
first_page = await client.messages.batches.list(limit=20)
if first_page.has_next_page():
print(f"will fetch next page using these details: {first_page.next_page_info()}")
next_page = await first_page.get_next_page()
print(f"number of items we just fetched: {len(next_page.data)}")
# Remove `await` for non-async usage.
반환된 데이터로 직접 작업할 수도 있어요.
first_page = await client.messages.batches.list(limit=20)
print(f"next page cursor: {first_page.last_id}")
for batch in first_page.data:
print(batch.id)
# Remove `await` for non-async usage.
기본 헤더
SDK는 2023-06-01로 설정된 anthropic-version 헤더를 자동으로 보내요.
필요하면 클라이언트 객체나 요청별로 기본 헤더를 설정해 덮어쓸 수 있어요.
경고: 기본 헤더를 덮어쓰면 SDK에서 잘못된 타입이나 기타 예상치 못한 또는 정의되지 않은 동작이 발생할 수 있어요.
# Set default headers for all requests on the client
client = Anthropic(
default_headers={"anthropic-version": "My-Custom-Value"},
)
# Or override per-request
client.messages.with_raw_response.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-5-5",
extra_headers={"anthropic-version": "My-Custom-Value"},
)
타입 시스템
요청 파라미터
중첩 요청 파라미터는 TypedDict예요. 응답은 JSON으로 다시 직렬화하는 헬퍼 메서드가 있는 Pydantic 모델(v1, v2)이에요.
타입 있는 요청과 응답은 에디터에서 자동 완성과 문서를 제공해요. VS Code에서 타입 오류를 보고 버그를 더 일찍 잡으려면 python.analysis.typeCheckingMode를 basic으로 설정하세요.
응답 모델
Pydantic 모델을 딕셔너리로 변환하려면 헬퍼 메서드를 사용해요.
message = client.messages.create(...)
# Convert to JSON string
json_str = message.to_json()
# Convert to dictionary
data = message.to_dict()
null vs 누락 필드 처리
응답에서 명시적으로 null인 필드와 반환되지 않은(누락된) 필드를 구분할 수 있어요.
response = client.messages.create(
model="claude-opus-5-5",
max_tokens=1024,
messages=[{"role": "user", "content": "Hello"}],
)
if response.my_field is None:
if "my_field" not in response.model_fields_set:
print("field was not in the response")
else:
print("field was null")
고급 사용법
원시 응답 데이터 접근 (예: 헤더)
httpx2가 반환한 "raw" Response는 클라이언트의 .with_raw_response 속성을 통해 접근할 수 있어요. 응답 헤더나 기타 메타데이터에 접근할 때 유용해요.
client = Anthropic()
response = client.messages.with_raw_response.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-5-5",
)
print(response.headers.get("request-id"))
message = (
response.parse()
) # get the object that `messages.create()` would have returned
print(message.content)
이 메서드들은 APIResponse 객체를 반환해요. 비동기 클라이언트에서는 AsyncAPIResponse를 반환하며, .parse(), .read(), .text(), .json()은 await해야 해요.
스트리밍 응답 본문
.with_raw_response 접근 방식은 요청할 때 전체 응답 본문을 적극적으로(즉시) 읽어요. 대신 응답 본문을 스트리밍하려면 .with_streaming_response를 사용하세요. 컨텍스트 매니저가 필요하고, .read(), .text(), .json(), .iter_bytes(), .iter_text(), .iter_lines(), .parse()를 호출할 때만 응답 본문을 읽어요. 비동기 클라이언트에서는 이들이 비동기 메서드예요.
with client.messages.with_streaming_response.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-5-5",
) as response:
print(response.headers.get("request-id"))
for line in response.iter_lines():
print(line)
응답이 확실히 닫히도록 컨텍스트 매니저가 필요해요.
로깅
SDK는 표준 라이브러리 logging 모듈을 사용해요.
ANTHROPIC_LOG 환경 변수를 debug 또는 info로 설정하면 로깅을 켤 수 있어요.
export ANTHROPIC_LOG=debug
커스텀/문서화되지 않은 요청 만들기
이 라이브러리는 문서화된 API에 편리하게 접근하도록 타입이 지정돼 있어요. 문서화되지 않은 엔드포인트, 파라미터, 응답 속성에 접근해야 한다면 그래도 쓸 수 있어요.
문서화되지 않은 엔드포인트
문서화되지 않은 엔드포인트에 요청하려면 client.get, client.post 등 HTTP 동사를 사용할 수 있어요. 재시도 같은 클라이언트 옵션은 이 요청들을 만들 때도 존중돼요.
import httpx2
response = client.post(
"/foo",
cast_to=httpx2.Response,
body={"my_param": True},
)
print(response.json())
문서화되지 않은 요청 파라미터
추가 파라미터를 명시적으로 보내려면 extra_query, extra_body, extra_headers 요청 옵션을 사용할 수 있어요.
경고:
extra_파라미터는 같은 이름의 문서화된 파라미터를 덮어써요. 보안상의 이유로 이 메서드는 신뢰할 수 있는 입력 데이터로만 사용해야 해요.
문서화되지 않은 응답 속성
문서화되지 않은 응답 속성에 접근하려면 response.unknown_prop처럼 추가 필드에 접근할 수 있어요. response.model_extra로 Pydantic 모델의 모든 추가 필드를 딕셔너리로 가져올 수도 있어요.
HTTP 클라이언트 구성
SDK는 httpx의 API 호환 포크인 httpx2로 요청을 보내요. 프록시와 전송을 포함해 HTTP 클라이언트를 커스터마이즈하려면 자체 httpx2 클라이언트를 http_client로 넘기세요.
import httpx2
from anthropic import Anthropic, DefaultHttpxClient
client = Anthropic(
# Or use the `ANTHROPIC_BASE_URL` env var
base_url="http://my.test.server.example.com:8083",
http_client=DefaultHttpxClient(
proxy="http://my.test.proxy.example.com",
transport=httpx2.HTTPTransport(local_address="0.0.0.0"),
),
)
with_options()로 요청별로 클라이언트를 커스터마이즈할 수도 있어요.
client.with_options(http_client=DefaultHttpxClient(...))
참고: SDK의 기본 구성(타임아웃, 연결 제한 등)이 보존되도록 raw
httpx2.Client와httpx2.AsyncClient대신DefaultHttpxClient와DefaultAsyncHttpxClient를 사용하세요.http_client인자는httpx2클라이언트여야 해요. 별도httpx패키지의 클라이언트를 넘기면TypeError가 발생해요.
OpenTelemetry의 HTTPXClientInstrumentor, Sentry의 httpx 통합, respx, pytest-httpx처럼 httpx 자체를 패치하는 추적·모킹 도구는 기본적으로 SDK 요청을 보지 못해요. 이 도구를 쓰려면 아무것도 httpx를 import하기 전에 시작 시 httpx2.alias_httpx()를 한 번 호출하세요. 그러면 프로세스 전체에서 import httpx가 httpx2로 해석돼요.
HTTP 리소스 관리
기본적으로 라이브러리는 클라이언트가 가비지 컬렉션될 때 기본 HTTP 연결을 닫아요. 원하면 .close() 메서드로 클라이언트를 수동으로 닫거나, 종료 시 닫히는 컨텍스트 매니저를 사용할 수 있어요.
with Anthropic() as client:
message = client.messages.create(...)
# HTTP client is automatically closed
베타 기능
베타 기능은 정식 릴리스 전에 조기 피드백과 새 기능 테스트를 위해 제공돼요. 모든 Claude 능력과 도구의 가용성은 build with Claude 개요에서 확인할 수 있어요.
대부분의 베타 API 기능은 클라이언트의 beta 속성으로 접근할 수 있어요. 특정 베타 기능을 켜려면 메시지를 만들 때 betas 필드에 적절한 베타 헤더를 추가해야 해요.
예를 들어 컨텍스트 편집을 켜려면 이렇게 해요.
client = Anthropic()
response = client.beta.messages.create(
model="claude-opus-5-5",
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
betas=["context-management-2025-06-27"],
)
플랫폼 통합
참고: 코드 예제가 있는 상세한 플랫폼 설정 가이드는 다음을 참고해 주세요.
다섯 개의 클라이언트 클래스 모두 기본 anthropic 패키지에 포함돼 있어요.
| 프로바이더 | 클라이언트 | 추가 의존성 |
|---|---|---|
| Agent Platform | from anthropic import AnthropicVertex |
pip install "anthropic[vertex]" |
| Bedrock | from anthropic import AnthropicBedrockMantle |
pip install "anthropic[bedrock]" |
Bedrock (bedrock-runtime 경로) |
from anthropic import AnthropicBedrock |
pip install "anthropic[bedrock]" |
| Claude Platform on AWS | from anthropic import AnthropicAWS |
pip install "anthropic[aws]" |
| Foundry | from anthropic import AnthropicFoundry |
없음 |
AnthropicAWS 클라이언트는 베타예요. 생성자에 workspace_id를 넘기거나 ANTHROPIC_AWS_WORKSPACE_ID 환경 변수를 설정하세요.
새 프로젝트에는 AnthropicBedrockMantle를 사용하고, Bedrock InvokeModel API를 쓰는 기존 애플리케이션에는 AnthropicBedrock을 남겨 두세요.
시맨틱 버저닝
이 패키지는 일반적으로 SemVer 규칙을 따르지만, 일부 이전 버전과 호환되지 않는 변경이 마이너 버전으로 릴리스될 수 있어요.
- 런타임 동작을 깨지 않고 정적 타입에만 영향을 주는 변경
- 기술적으로 공개지만 외부 사용 의도나 문서화가 없는 라이브러리 내부 변경
- 실제로 대다수 사용자에게 영향을 주지 않을 것으로 예상되는 변경
설치된 버전 확인
최신 버전으로 업그레이드했는데 기대한 새 기능이 안 보인다면, Python 환경이 여전히 이전 버전을 쓰고 있을 가능성이 높아요. 런타임에 사용 중인 버전을 확인할 수 있어요.
print(anthropic.__version__)