Anthropic Python SDK 提供從 Python 應用程式便捷存取 Anthropic REST API 的方式。它支援同步與非同步操作、串流,以及與 Amazon Bedrock、Vertex AI、Microsoft Foundry 和 Claude Platform on AWS 的整合。
如需包含程式碼範例的 API 功能文件,請參閱 API 參考。本頁面涵蓋 Python 專屬的 SDK 功能與設定。
pip install anthropic如需平台專屬的整合或提升非同步效能,請使用額外套件進行安裝:
# 用於支援 Amazon Bedrock
pip install "anthropic[bedrock]"
# 用於支援 Vertex AI
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-4-8",
)
print(message.content)建議使用 python-dotenv 將 ANTHROPIC_API_KEY="my-anthropic-api-key" 加入您的 .env 檔案,以避免將 API 金鑰儲存在原始碼版本控制中。
如需了解包含「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-4-8",
)
print(message.content)
asyncio.run(main())為了提升非同步效能,您可以使用 aiohttp HTTP 後端來取代預設的 httpx:
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-4-8",
)
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-4-8",
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-4-8",
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-4-8",
) 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)您也可以在發出請求之前計算 token 數量:
count = client.messages.count_tokens(
model="claude-opus-4-8", messages=[{"role": "user", "content": "Hello, world"}]
)
print(count.input_tokens) # 10此 SDK 支援「tool use」(工具使用),也稱為函式呼叫。如需更多詳細資訊,請參閱搭配 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",
}
)
# 使用 tool_runner 自動處理工具呼叫
runner = client.beta.messages.tool_runner(
max_tokens=1024,
model="claude-opus-4-8",
tools=[get_weather],
messages=[
{"role": "user", "content": "What is the weather in SF?"},
],
)
for message in runner:
print(message)在每次迭代中,都會發出一個 API 請求。如果 Claude 想要呼叫其中一個指定的工具,該工具會自動被呼叫,且結果會在下一次迭代中直接回傳給模型。
此 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-4-8",
"max_tokens": 1024,
"messages": [{"role": "user", "content": "Hello, world"}],
},
},
{
"custom_id": "my-second-request",
"params": {
"model": "claude-opus-4-8",
"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.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-4-8",
)
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-4-8",
)
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-4-8",
)預設情況下,請求會在 10 分鐘後逾時。您可以使用 timeout 選項進行設定,該選項接受浮點數或 httpx.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-4-8",
)逾時時會拋出 APITimeoutError。
請注意,逾時的請求預設會重試兩次。
對於執行時間較長的請求,建議使用串流 Messages API。
避免在不使用串流的情況下設定較大的 max_tokens 值。某些網路可能會在一段時間後中斷閒置連線,這可能導致請求失敗或逾時而未收到 Anthropic 的回應。
如果預期非串流請求的執行時間超過約 10 分鐘,SDK 會拋出 ValueError。傳遞 stream=True 或在客戶端或請求層級覆寫 timeout 選項可停用此錯誤。
對於非串流請求,如果預期的請求「latency」(延遲)超過逾時設定,客戶端會終止連線並重試,而不會收到回應。
SDK 設定了 TCP socket keep-alive 選項,以減少某些網路上閒置連線逾時的影響。這可以透過將自訂的 http_client 選項傳遞給客戶端來覆寫。
Claude API 中的列表方法是分頁的。您可以使用 for 語法來迭代所有頁面中的項目:
client = Anthropic()
all_batches = []
# 根據需要自動擷取更多頁面。
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)}")
# 若為非非同步用法,請移除 `await`。或直接處理回傳的資料:
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)
# 若為非非同步用法,請移除 `await`。SDK 會自動傳送設定為 2023-06-01 的 anthropic-version 標頭。
如有需要,您可以在客戶端物件上或針對個別請求設定預設標頭來覆寫它。
覆寫預設標頭可能會導致 SDK 中出現不正確的類型以及其他非預期或未定義的行為。
# 在用戶端上為所有請求設定預設標頭
client = Anthropic(
default_headers={"anthropic-version": "My-Custom-Value"},
)
# 或針對個別請求進行覆寫
client.messages.with_raw_response.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-4-8",
extra_headers={"anthropic-version": "My-Custom-Value"},
)巢狀請求參數是 TypedDicts。回應是 Pydantic 模型,它們也提供輔助方法,例如序列化回 JSON(v1、v2)。
具類型的請求和回應可在您的編輯器中提供自動完成和文件說明。如果您想在 VS Code 中看到類型錯誤以協助及早發現錯誤,請將 python.analysis.typeCheckingMode 設定為 basic。
若要將 Pydantic 模型轉換為字典,請使用輔助方法:
message = client.messages.create(...)
# 轉換為 JSON 字串
json_str = message.to_json()
# 轉換為字典
data = message.to_dict()在回應中,您可以區分明確為 null 的欄位與未回傳(缺少)的欄位:
response = client.messages.create(
model="claude-opus-4-8",
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")可以透過客戶端上的 .with_raw_response 屬性存取 httpx 回傳的「原始」Response。這對於存取回應標頭或其他中繼資料很有用:
client = Anthropic()
response = client.messages.with_raw_response.create(
max_tokens=1024,
messages=[{"role": "user", "content": "Hello, Claude"}],
model="claude-opus-4-8",
)
print(response.headers.get("request-id"))
message = (
response.parse()
) # get the object that `messages.create()` would have returned
print(message.content)這些方法會回傳 APIResponse 物件。
.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-4-8",
) 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 httpx
response = client.post(
"/foo",
cast_to=httpx.Response,
body={"my_param": True},
)
print(response.json())如果您想明確傳送額外的參數,可以使用 extra_query、extra_body 和 extra_headers 請求選項。
extra_ 參數會覆寫同名的已記錄參數。基於安全考量,請確保這些方法僅用於受信任的輸入資料。
若要存取未記錄的回應屬性,您可以像 response.unknown_prop 這樣存取額外欄位。您也可以使用 response.model_extra 以字典形式取得 Pydantic 模型上的所有額外欄位。
您可以直接覆寫 httpx 客戶端以針對您的使用案例進行自訂,包括支援代理伺服器和傳輸:
import httpx
from anthropic import Anthropic, DefaultHttpxClient
client = Anthropic(
# 或使用 `ANTHROPIC_BASE_URL` 環境變數
base_url="http://my.test.server.example.com:8083",
http_client=DefaultHttpxClient(
proxy="http://my.test.proxy.example.com",
transport=httpx.HTTPTransport(local_address="0.0.0.0"),
),
)您也可以使用 with_options() 針對個別請求自訂客戶端:
client.with_options(http_client=DefaultHttpxClient(...))請使用 DefaultHttpxClient 和 DefaultAsyncHttpxClient,而非原始的 httpx.Client 和 httpx.AsyncClient,以確保保留 SDK 的預設設定(逾時、連線限制等)。
預設情況下,當客戶端被垃圾回收時,程式庫會關閉底層的 HTTP 連線。如有需要,您可以使用 .close() 方法手動關閉客戶端,或使用在退出時關閉的上下文管理器。
with Anthropic() as client:
message = client.messages.create(...)
# HTTP 客戶端會自動關閉Beta 功能在正式發布之前提供,以便獲得早期回饋並測試新功能。您可以在使用 Claude 建構總覽中查看 Claude 所有功能和工具的可用性。
您可以透過客戶端的 beta 屬性存取大多數 beta API 功能。若要啟用特定的 beta 功能,您需要在建立訊息時將適當的 beta 標頭新增至 betas 欄位。
例如,若要使用 Files API:
client = Anthropic()
response = client.beta.messages.create(
model="claude-opus-4-8",
max_tokens=1024,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "Please summarize this document for me."},
{
"type": "document",
"source": {
"type": "file",
"file_id": "file_abc123",
},
},
],
},
],
betas=["files-api-2025-04-14"],
)如需包含程式碼範例的詳細平台設定指南,請參閱:
所有五個客戶端類別都包含在基礎 anthropic 套件中:
| 供應商 | 客戶端 | 額外相依套件 |
|---|---|---|
| Bedrock | from anthropic import AnthropicBedrockMantle | pip install "anthropic[bedrock]" |
Bedrock(bedrock-runtime 路徑) | from anthropic import AnthropicBedrock | pip install "anthropic[bedrock]" |
| Vertex AI | from anthropic import AnthropicVertex | pip install "anthropic[vertex]" |
| Foundry | from anthropic import AnthropicFoundry | 無 |
| Claude Platform on AWS | from anthropic import AnthropicAWS | pip install "anthropic[aws]" |
AnthropicAWS 客戶端目前為 beta 版。請將 workspace_id 傳遞給建構函式,或設定 ANTHROPIC_AWS_WORKSPACE_ID 環境變數。
新專案請使用 AnthropicBedrockMantle;AnthropicBedrock 保留給使用 Bedrock InvokeModel API 的現有應用程式。
此套件大致遵循 SemVer 慣例,但某些向後不相容的變更可能會以次要版本發布:
如果您已升級到最新版本,但沒有看到預期的新功能,您的 Python 環境可能仍在使用較舊的版本。您可以在執行時使用以下方式判斷正在使用的版本:
print(anthropic.__version__)Was this page helpful?