Python SDK
安裝並設定支援同步與非同步用戶端的 Anthropic Python SDK
Anthropic Python SDK 讓您能從 Python 應用程式便捷地存取 Claude API。它支援同步與非同步操作、「streaming」(串流),以及與 Amazon Bedrock、Claude Platform on AWS、Claude Models on Google Cloud (Anthropic managed)、Claude Models on Google Cloud (Google managed) 和 Microsoft Foundry 的整合。
安裝
pip install anthropic若需要特定平台的整合或更佳的非同步效能,請安裝額外套件:
# 用於 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]"若要使用 Claude Models on Google Cloud (Anthropic managed),請執行 pip install "anthropic[google_cloud]"。
需求
需要 Python 3.10 或更新版本。如果您是從 SDK 的 0.x 版本升級,請參閱 v1 遷移指南以了解重大變更清單。
使用方式
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-5",
)
for block in message.content:
if block.type == "text":
print(block.text)如需包括「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 以獲得更佳的並行效能
為了提升非同步效能,您可以使用 aiohttp HTTP 後端來取代預設的 httpx2:
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 也提供串流輔助工具,這些工具使用 context manager(情境管理器),並可存取累積的文字與最終訊息:
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),它只會回傳串流中事件的可迭代物件,並使用較少的記憶體(它不會為您建構最終的訊息物件)。
Token 計數
您可以透過 usage 回應屬性查看特定請求的確切用量:
message = client.messages.create(...)
print(message.usage)
# Usage(input_tokens=25, output_tokens=13)您也可以在發出請求之前計算 token 數量:
count = client.messages.count_tokens(
model="claude-opus-5-5",
messages=[{"role": "user", "content": "Hello, world"}],
)
print(count.input_tokens) # 10工具使用
此 SDK 支援「tool use」(工具使用),也稱為「function calling」(函式呼叫)。如需更多詳細資訊,請參閱搭配 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-5-5",
tools=[get_weather],
messages=[
{"role": "user", "content": "What is the weather in SF?"},
],
)
for message in runner:
print(message)每次迭代都會發出一個 API 請求。如果回應中包含對某個指定工具的呼叫,該工具會被自動呼叫,且結果會在下一次迭代中直接回傳給模型。
若要透過此 SDK 執行瀏覽器使用工具或電腦使用工具,請參閱使用 SDK 工具集進行瀏覽器與電腦使用。
訊息批次
此 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
如需更多關於偵錯請求的資訊,請參閱請求 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重試
某些錯誤預設會自動重試 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 選項進行設定,該選項接受浮點數或 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。
請注意,逾時的請求預設會重試兩次。
長時間請求
避免在未使用串流的情況下設定較大的 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)}")
# 若為非 async 用法,請移除 `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)
# 若為非 async 用法,請移除 `await`。預設標頭
SDK 會自動傳送 anthropic-version 標頭,其值設定為此 SDK 主要版本所對應的 API 版本(請參閱語意化版本控制)。
如有需要,您可以透過在用戶端物件上或針對個別請求設定預設標頭來覆寫它。
# 為用戶端上的所有請求設定預設標頭
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-5-5",
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 與缺少的欄位
在回應中,您可以區分明確為 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")進階用法
存取原始回應資料(例如標頭)
您可以透過用戶端上的 .with_raw_response 屬性存取 httpx2 回傳的「原始」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,它需要 context manager,且只有在您呼叫 .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)需要使用 context manager,以確保回應能可靠地被關閉。
日誌記錄
SDK 使用標準函式庫的 logging 模組。
您可以透過將環境變數 ANTHROPIC_LOG 設定為 debug 或 info 來啟用日誌記錄:
export ANTHROPIC_LOG=debugOpenTelemetry 追蹤
當您的應用程式註冊 OpenTelemetry tracer provider 時,SDK 會為其傳送的每個請求發出一個 span。若要將這些 span 傳送至您的可觀測性平台,請參閱搭配 Claude SDK 設定 OpenTelemetry。
發出自訂/未記載的請求
此程式庫具有型別定義,方便存取已記載的 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 請求選項。
未記載的回應屬性
若要存取未記載的回應屬性,您可以存取額外欄位,例如 response.unknown_prop。您也可以使用 response.model_extra 以字典形式取得 Pydantic 模型上的所有額外欄位。
設定 HTTP 用戶端
SDK 使用 httpx2 傳送請求,這是 httpx 的 API 相容分支。若要自訂 HTTP 用戶端(包括代理伺服器與傳輸層),請將您自己的 httpx2 用戶端作為 http_client 傳入:
import httpx2
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=httpx2.HTTPTransport(local_address="0.0.0.0"),
),
)您也可以使用 with_options() 針對個別請求自訂用戶端:
client.with_options(http_client=DefaultHttpxClient(...))會修補 httpx 本身的追蹤與模擬工具,例如 OpenTelemetry 的 HTTPXClientInstrumentor、Sentry 的 httpx 整合、respx 或 pytest-httpx,預設不會看到 SDK 的請求。若要使用這些工具,請在啟動時、任何程式匯入 httpx 之前呼叫一次 httpx2.alias_httpx()。這會讓整個程序中的 import httpx 解析為 httpx2。
管理 HTTP 資源
預設情況下,每當用戶端被垃圾回收時,程式庫就會關閉底層的 HTTP 連線。如有需要,您可以使用 .close() 方法手動關閉用戶端,或使用在離開時自動關閉的 context manager。
with Anthropic() as client:
message = client.messages.create(...)
# HTTP 用戶端會自動關閉Beta 功能
Beta 功能會在正式發布前提供,以取得早期回饋並測試新功能。您可以在功能概覽中查看 Claude 所有功能與工具的可用性。
您可以透過用戶端的 beta 屬性存取大多數 beta API 功能。若要啟用特定的 beta 功能,您需要在建立訊息時將適當的 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]" |
| Claude Models on Google Cloud (Anthropic managed) | from anthropic import AnthropicGoogleCloud | pip install "anthropic[google_cloud]" |
| Foundry | from anthropic import AnthropicFoundry | 無 |
AnthropicAWS 用戶端目前為 beta 版。請將 workspace_id 傳入建構函式,或設定 ANTHROPIC_AWS_WORKSPACE_ID 環境變數。AnthropicGoogleCloud 用戶端同樣為 beta 版。請將 workspace_id 傳入建構函式,或設定 ANTHROPIC_GOOGLE_CLOUD_WORKSPACE_ID 環境變數。如需了解用戶端如何解析工作區 ID,請參閱專案、位置與工作區解析。
新專案請使用 AnthropicBedrockMantle;AnthropicBedrock 則保留給使用 Bedrock InvokeModel API 的現有應用程式。
語意化版本控制
此 SDK 的每個主要版本預設會在 anthropic-version 標頭中傳送一個 API 版本,且每個新的 API 版本都會伴隨一個新的主要版本。第一個傳送 2026-10-21 的主要版本是 2.0.0;請參閱遷移至 API 版本 2026-10-21。
此套件大致遵循 SemVer 慣例,但某些不向下相容的變更可能會以次要版本發布:
- 僅影響靜態型別、不會破壞執行時期行為的變更。
- 對程式庫內部的變更,這些內部在技術上是公開的,但並非設計或記載供外部使用。
- 預期在實務上不會影響絕大多數使用者的變更。
確認已安裝的版本
如果您已升級到最新版本,但沒有看到預期的新功能,您的 Python 環境很可能仍在使用較舊的版本。您可以使用以下方式確認執行時期所使用的版本:
print(anthropic.__version__)其他資源
Was this page helpful?