Anthropic Python SDK는 Python 애플리케이션에서 Anthropic REST 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]"
# AWS의 Claude Platform 지원용
pip install "anthropic[aws]"
# Microsoft Foundry 지원은 기본 패키지에 포함되어 있습니다
# aiohttp를 통한 비동기 성능 향상용
pip install "anthropic[aiohttp]"Python 3.9 이상이 필요합니다.
import os
from anthropic import Anthropic
client = Anthropic(
# 이것은 기본값이며 생략할 수 있습니다
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",
)
for block in message.content:
if block.type == "text":
print(block.text)API 키가 소스 컨트롤에 저장되지 않도록 python-dotenv를 사용하여 .env 파일에 ANTHROPIC_API_KEY="my-anthropic-api-key"를 추가하는 것을 고려하세요.
Workload Identity Federation을 포함한 인증 옵션에 대해서는 인증을 참조하세요.
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())향상된 비동기 성능을 위해 기본값인 httpx 대신 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",
)
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",
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",
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",
) 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)를 사용할 수 있으며, 이는 스트림의 이벤트에 대한 이터러블만 반환하고 더 적은 메모리를 사용합니다(최종 메시지 객체를 만들어 주지 않습니다).
usage 응답 속성을 통해 특정 요청의 정확한 사용량을 확인할 수 있습니다:
message = client.messages.create(...)
print(message.usage)
# Usage(input_tokens=25, output_tokens=13)요청을 보내기 전에 토큰을 계산할 수도 있습니다:
count = client.messages.count_tokens(
model="claude-opus-5", messages=[{"role": "user", "content": "Hello, world"}]
)
print(count.input_tokens) # 10이 SDK는 함수 호출(function calling)이라고도 하는 도구 사용을 지원합니다. 자세한 내용은 Claude와 함께하는 도구 사용을 참조하세요.
SDK는 순수 Python 함수로 도구를 정의하고 실행하기 위한 헬퍼를 제공합니다. @beta_tool 데코레이터는 함수 시그니처와 독스트링으로부터 도구 스키마를 생성합니다:
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",
}
)
# tool_runner를 사용하여 도구 호출을 자동으로 처리합니다
runner = client.beta.messages.tool_runner(
max_tokens=1024,
model="claude-opus-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 API를 지원합니다.
Message Batches는 요청 배열을 받으며, 각 객체는 custom_id 식별자와 표준 Messages API와 동일한 요청 params를 가집니다:
client.messages.batches.create(
requests=[
{
"custom_id": "my-first-request",
"params": {
"model": "claude-opus-5",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "Hello, world"}],
},
},
{
"custom_id": "my-second-request",
"params": {
"model": "claude-opus-5",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "Hi again, friend"}],
},
},
]
).processing_status == 'ended'로 표시되는 Message Batch 처리가 완료되면, .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.beta.files.upload(
file=Path("/path/to/file"),
)
# 바이트를 사용하여 업로드
client.beta.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",
)
except anthropic.APIConnectionError as e:
print("The server could not be reached")
print(e.__cause__) # an underlying Exception, likely raised within httpx
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를 참조하세요.
SDK의 모든 객체 응답은 request-id 응답 헤더에서 추가된 _request_id 속성을 제공하므로, 실패한 요청을 빠르게 로깅하고 Anthropic에 보고할 수 있습니다.
message = client.messages.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-5",
)
print(message._request_id) # e.g., req_018EeWyXxfu5pfWkrYcMdjWG_ 접두사를 사용하는 다른 속성과 달리 _request_id 속성은 공개(public)입니다. 별도로 문서화되지 않은 한, 다른 모든 _ 접두사 속성, 메서드, 모듈은 비공개(private)입니다.
특정 오류는 기본적으로 짧은 지수 백오프와 함께 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",
)기본적으로 요청은 10분 후에 타임아웃됩니다. float 또는 httpx.Timeout 객체를 받는 timeout 옵션으로 이를 구성할 수 있습니다:
import httpx
from anthropic import Anthropic
# 모든 요청에 대한 기본값 구성:
client = Anthropic(
timeout=20.0, # 20 seconds (default is 10 minutes)
)
# 더 세밀한 제어:
client = Anthropic(
timeout=httpx.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",
)타임아웃 시 SDK는 APITimeoutError를 발생시킵니다.
타임아웃된 요청은 기본적으로 두 번 재시도됩니다.
오래 실행되는 요청에는 스트리밍 Messages API 사용을 고려하세요.
스트리밍을 사용하지 않고 큰 max_tokens 값을 설정하는 것은 피하세요. 일부 네트워크는 일정 시간이 지나면 유휴 연결을 끊을 수 있으며, 이로 인해 Anthropic으로부터 응답을 받지 못한 채 요청이 실패하거나 타임아웃될 수 있습니다.
비스트리밍 요청이 약 10분 이상 걸릴 것으로 예상되는 경우 SDK는 ValueError를 발생시킵니다. stream=True를 전달하거나 클라이언트 또는 요청 수준에서 timeout 옵션을 재정의하면 이 오류가 비활성화됩니다.
비스트리밍 요청에서 예상 요청 latency(지연 시간)가 타임아웃보다 길면 클라이언트가 응답을 받지 못한 채 연결을 종료하고 재시도하게 됩니다.
SDK는 일부 네트워크에서 유휴 연결 타임아웃의 영향을 줄이기 위해 TCP 소켓 keep-alive 옵션을 설정합니다. 이는 클라이언트에 사용자 정의 http_client 옵션을 전달하여 재정의할 수 있습니다.
Claude API의 목록 메서드는 페이지네이션됩니다. for 구문을 사용하여 모든 페이지의 항목을 순회할 수 있습니다:
client = Anthropic()
all_batches = [