Python SDK
安裝並設定 Anthropic Python SDK,支援同步與非同步用戶端
Anthropic Python SDK 讓您可以從 Python 應用程式方便地存取 Claude API。它同時支援同步與非同步操作、「streaming」(串流),以及與 Amazon Bedrock、Claude Platform on AWS、Google Cloud 和 Microsoft Foundry 的整合。
安裝
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(
# 此為預設值,可省略
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)如需了解包括 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",
)
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",
)
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 也提供使用 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",
) 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", 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-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",
"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"}],
},
},
]
)從批次取得結果
一旦 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)形式的 tupleBinaryIO類檔案物件
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",
)
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",
)
print(message._request_id) # e.g., req_018EeWyXxfu5pfWkrYcMdjWG重試
某些錯誤預設會自動重試 2 次,並採用短暫的指數退避。連線錯誤(例如因網路連線問題所致)、408 Request Timeout、409 Conflict、429 Rate Limit 以及 >=500 的內部錯誤預設都會重試。
您可以使用 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 分鐘後逾時。您可以使用 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",
)逾時時,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)}")
# 非同步以外的用法請移除 `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 標頭。
如有需要,您可以在用戶端物件上或針對個別請求設定預設標頭來覆寫它。
# 為用戶端上的所有請求設定預設標頭
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",
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",
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 回傳的「原始」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",
)
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",
) 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=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 請求選項。
未記載於文件的回應屬性
若要存取未記載於文件的回應屬性,您可以像 response.unknown_prop 這樣存取額外欄位。您也可以透過 response.model_extra 以 dict 形式取得 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 建構概覽中查看 Claude 所有功能與工具的可用性。
您可以透過用戶端的 beta 屬性存取大多數 beta API 功能。若要啟用特定的 beta 功能,您需要在建立訊息時將適當的 beta 標頭加入 betas 欄位。
例如,若要啟用上下文編輯:
client = Anthropic()
response = client.beta.messages.create(
model="claude-opus-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 用戶端目前為 beta 版。請將 workspace_id 傳入建構函式,或設定 ANTHROPIC_AWS_WORKSPACE_ID 環境變數。
新專案請使用 AnthropicBedrockMantle;AnthropicBedrock 則保留給使用 Bedrock InvokeModel API 的既有應用程式。
語意化版本
此套件大致遵循 SemVer 慣例,但某些不向後相容的變更可能會以次要版本發布:
- 僅影響靜態型別、不破壞執行期行為的變更。
- 對函式庫內部的變更,這些內部在技術上是公開的,但並非預期或記載供外部使用。
- 實務上預期不會影響絕大多數使用者的變更。
確認已安裝的版本
如果您已升級至最新版本,卻沒有看到預期的新功能,您的 Python 環境很可能仍在使用較舊的版本。您可以透過以下方式確認執行期所使用的版本:
print(anthropic.__version__)其他資源
Was this page helpful?