CLI、SDK 和库客户端 SDK

Python SDK

安装和配置 Anthropic Python SDK，支持同步和异步客户端

Anthropic Python SDK 提供了从 Python 应用程序便捷访问 Anthropic REST API 的方式。它支持同步和异步操作、流式传输，以及与 Amazon Bedrock、Vertex AI、Microsoft Foundry 和 AWS 上的 Claude Platform 的集成。

有关包含代码示例的 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 获得更好的并发性能

为了提高异步性能，您可以使用 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)

您也可以在发出请求之前计算令牌数：

count = client.messages.count_tokens(
    model="claude-opus-4-8", messages=[{"role": "user", "content": "Hello, world"}]
)
print(count.input_tokens)  # 10

工具使用

此 SDK 支持工具使用，也称为函数调用。有关更多详细信息，请参阅 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-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

有关调试请求的更多信息，请参阅请求 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 请求超时、409 冲突、429 速率限制以及 >=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-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 选项可禁用此错误。

对于非流式传输请求，如果预期的请求延迟超过超时时间，客户端将终止连接并重试，而不会收到响应。

SDK 设置了 TCP 套接字保持活动选项，以减少某些网络上空闲连接超时的影响。可以通过向客户端传递自定义 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 与缺失字段

在响应中，您可以区分显式为 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 模型上的所有额外字段。

配置 HTTP 客户端

您可以直接覆盖 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 资源

默认情况下，每当客户端被垃圾回收时，库都会关闭底层 HTTP 连接。如果需要，您可以使用 .close() 方法手动关闭客户端，或使用在退出时关闭的上下文管理器。

with Anthropic() as client:
    message = client.messages.create(...)

# HTTP 客户端会自动关闭

Beta 功能

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`	无
AWS 上的 Claude Platform	`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?

CLI、SDK 和库客户端 SDK

Python SDK

安装和配置 Anthropic Python SDK，支持同步和异步客户端

有关包含代码示例的 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 获得更好的并发性能

为了提高异步性能，您可以使用 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)

您也可以在发出请求之前计算令牌数：

count = client.messages.count_tokens(
    model="claude-opus-4-8", messages=[{"role": "user", "content": "Hello, world"}]
)
print(count.input_tokens)  # 10

工具使用

此 SDK 支持工具使用，也称为函数调用。有关更多详细信息，请参阅 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-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

有关调试请求的更多信息，请参阅请求 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 属性是公开的。除非另有说明，所有其他带 _ 前缀的属性、方法和模块都是私有的。

重试

您可以使用 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。

如果非流式传输请求预计耗时超过约 10 分钟，SDK 将抛出 ValueError。传递 stream=True 或在客户端或请求级别覆盖 timeout 选项可禁用此错误。

对于非流式传输请求，如果预期的请求延迟超过超时时间，客户端将终止连接并重试，而不会收到响应。

SDK 设置了 TCP 套接字保持活动选项，以减少某些网络上空闲连接超时的影响。可以通过向客户端传递自定义 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）。

响应模型

要将 Pydantic 模型转换为字典，请使用辅助方法：

message = client.messages.create(...)

# 转换为 JSON 字符串
json_str = message.to_json()

# 转换为字典
data = message.to_dict()

处理 null 与缺失字段

在响应中，您可以区分显式为 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 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_ 参数会覆盖同名的已记录参数。出于安全原因，请确保这些方法仅用于受信任的输入数据。

未记录的响应属性

配置 HTTP 客户端

您可以直接覆盖 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 资源

with Anthropic() as client:
    message = client.messages.create(...)

# HTTP 客户端会自动关闭

Beta 功能

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`	无
AWS 上的 Claude Platform	`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?