Claude Platform Docs
CLI、SDK 與函式庫用戶端 SDK

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) 形式的 tuple
  • 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",
    )
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)

錯誤代碼如下:

狀態碼錯誤類型
400BadRequestError
401AuthenticationError
403PermissionDeniedError
404NotFoundError
409ConflictError
422UnprocessableEntityError
429RateLimitError
>=500InternalServerError
N/AAPIConnectionError

請求 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-01anthropic-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 等操作(v1v2)。

具型別的請求與回應可在您的編輯器中提供自動完成與文件說明。如果您希望在 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 設為 debuginfo 來啟用日誌記錄:

export ANTHROPIC_LOG=debug

發出自訂/未記載於文件的請求

此函式庫具有型別定義,以便方便地存取已記載於文件的 API。如果您需要存取未記載於文件的端點、參數或回應屬性,仍然可以使用此函式庫。

未記載於文件的端點

若要向未記載於文件的端點發出請求,您可以使用 client.getclient.post 及其他 HTTP 動詞。發出這些請求時,會遵循用戶端上的選項(例如重試)。

import httpx2

response = client.post(
    "/foo",
    cast_to=httpx2.Response,
    body={"my_param": True},
)

print(response.json())

未記載於文件的請求參數

如果您想明確傳送額外的參數,可以使用 extra_queryextra_bodyextra_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 整合、respxpytest-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 Platformfrom anthropic import AnthropicVertexpip install "anthropic[vertex]"
Bedrockfrom anthropic import AnthropicBedrockMantlepip install "anthropic[bedrock]"
Bedrock(bedrock-runtime 路徑)from anthropic import AnthropicBedrockpip install "anthropic[bedrock]"
Claude Platform on AWSfrom anthropic import AnthropicAWSpip install "anthropic[aws]"
Foundryfrom anthropic import AnthropicFoundry

AnthropicAWS 用戶端目前為 beta 版。請將 workspace_id 傳入建構函式,或設定 ANTHROPIC_AWS_WORKSPACE_ID 環境變數。

新專案請使用 AnthropicBedrockMantleAnthropicBedrock 則保留給使用 Bedrock InvokeModel API 的既有應用程式。

語意化版本

此套件大致遵循 SemVer 慣例,但某些不向後相容的變更可能會以次要版本發布:

  1. 僅影響靜態型別、不破壞執行期行為的變更。
  2. 對函式庫內部的變更,這些內部在技術上是公開的,但並非預期或記載供外部使用。
  3. 實務上預期不會影響絕大多數使用者的變更。

確認已安裝的版本

如果您已升級至最新版本,卻沒有看到預期的新功能,您的 Python 環境很可能仍在使用較舊的版本。您可以透過以下方式確認執行期所使用的版本:

print(anthropic.__version__)

其他資源

Was this page helpful?