Anthropic Python API 库

Anthropic Python 库为任何 Python 3.7+ 应用程序提供了便捷访问 Anthropic REST API 的方式。它包含所有请求参数和响应字段的类型定义，并提供由 httpx 驱动的同步和异步客户端。

文档

REST API 文档可在 docs.anthropic.com 上找到。本库的完整 API 可在 api.md 中找到。

安装

# 从 PyPI 安装
pip install anthropic

使用

本库的完整 API 可在 api.md 中找到。

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": "你好，Claude",
        }
    ],
    model="claude-3-opus-20240229",
)
print(message.content)

虽然你可以提供 api_key 关键字参数，但我们建议使用 python-dotenv 将 ANTHROPIC_API_KEY="my-anthropic-api-key" 添加到你的 .env 文件中，这样你的 API 密钥就不会存储在源代码管理中。

异步使用

只需导入 AsyncAnthropic 而不是 Anthropic，并在每个 API 调用时使用 await：

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": "你好，Claude",
            }
        ],
        model="claude-3-opus-20240229",
    )
    print(message.content)


asyncio.run(main())

同步和异步客户端的功能在其他方面是相同的。

流式响应

我们提供使用服务器端事件（SSE）的流式响应支持。

from anthropic import Anthropic

client = Anthropic()

stream = client.messages.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "你好，Claude",
        }
    ],
    model="claude-3-opus-20240229",
    stream=True,
)
for event in stream:
    print(event.type)

异步客户端使用完全相同的接口。

from anthropic import AsyncAnthropic

client = AsyncAnthropic()

stream = await client.messages.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "你好，Claude",
        }
    ],
    model="claude-3-opus-20240229",
    stream=True,
)
async for event in stream:
    print(event.type)

流式助手

本库为流式消息提供了几个便利功能，例如：

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def main() -> None:
    async with client.messages.stream(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "说你好！",
            }
        ],
        model="claude-3-opus-20240229",
    ) 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(...)
message.usage
# Usage(input_tokens=25, output_tokens=13)

工具使用

该 SDK 提供对工具使用（又称函数调用）的支持。更多详细信息可以在文档中找到。

AWS Bedrock

如果你使用 bedrock 额外选项安装此库，例如 pip install -U anthropic[bedrock]，该库还提供对 Anthropic Bedrock API 的支持。

然后你可以导入和实例化一个单独的 AnthropicBedrock 类，其余的 API 是相同的。

from anthropic import AnthropicBedrock

client = AnthropicBedrock()

message = client.messages.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "你好！",
        }
    ],
    model="anthropic.claude-3-sonnet-20240229-v1:0",
)
print(message)

有关更完整的示例，请参见 examples/bedrock.py。

Google Vertex

如果你使用 vertex 额外选项安装此库，例如 pip install -U anthropic[vertex]，该库还提供对 Anthropic Vertex API 的支持。

然后你可以导入和实例化一个单独的 AnthropicVertex/AsyncAnthropicVertex 类，它具有与基本 Anthropic/AsyncAnthropic 类相同的 API。

from anthropic import AnthropicVertex

client = AnthropicVertex()

message = client.messages.create(
    model="claude-3-sonnet@20240229",
    max_tokens=100,
    messages=[
        {
            "role": "user",
            "content": "你好！",
        }
    ],
)
print(message)

有关更完整的示例，请参见 examples/vertex.py。

使用类型

嵌套的请求参数是 TypedDicts。响应是 Pydantic 模型，它们还提供了一些辅助方法，例如：

序列化回 JSON，model.to_json()
转换为字典，model.to_dict()

类型化的请求和响应在你的编辑器中提供自动完成和文档。如果你想在 VS Code 中看到类型错误以帮助更早地捕获错误，请将 python.analysis.typeCheckingMode 设置为 basic。

处理错误

当库无法连接到 API（例如，由于网络连接问题或超时）时，会引发 anthropic.APIConnectionError 的子类。

当 API 返回非成功状态代码（即 4xx 或 5xx 响应）时，会引发 anthropic.APIStatusError 的子类，其中包含 status_code 和 response 属性。

所有错误都继承自 anthropic.APIError。

import anthropic
from anthropic import Anthropic

client = Anthropic()

try:
    client.messages.create(
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": "你好，Claude",
            }
        ],
        model="claude-3-opus-20240229",
    )
except anthropic.APIConnectionError as e:
    print("无法连接到服务器")
    print(e.__cause__)  # 一个底层异常，可能在 httpx 中引发。
except anthropic.RateLimitError as e:
    print("收到 429 状态码；我们应该稍微回退一下。")
except anthropic.APIStatusError as e:
    print("收到了另一个非 200 范围的状态码")
    print(e.status_code)
    print(e.response)

错误代码如下：

状态码	错误类型
400	`BadRequestError`
401	`AuthenticationError`
403	`PermissionDeniedError`
404	`NotFoundError`
422	`UnprocessableEntityError`
429	`RateLimitError`
>=500	`InternalServerError`
N/A	`APIConnectionError`

重试

某些错误默认会自动重试 2 次，并带有短暂的指数退避。连接错误（例如，由于网络连接问题）、408 请求超时、409 冲突、429 速率限制和 >=500 内部错误默认都会重试。

你可以使用 max_retries 选项来配置或禁用重试设置：

from anthropic import Anthropic

# 为所有请求配置默认值：
client = Anthropic(
    # 默认值为 2
    max_retries=0,
)

# 或者，按请求配置：
client.with_options(max_retries=5).messages.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "你好，Claude",
        }
    ],
    model="claude-3-opus-20240229",
)

超时

默认情况下，请求在 10 分钟后超时。你可以使用 timeout 选项配置这个值，它接受一个浮点数或一个 httpx.Timeout 对象：

from anthropic import Anthropic

# 为所有请求配置默认值：
client = Anthropic(
    # 20 秒（默认为 10 分钟）
    timeout=20.0,
)

# 更细粒度的控制：
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": "你好，Claude",
        }
    ],
    model="claude-3-opus-20240229",
)

超时时会抛出 APITimeoutError。

注意，超时的请求默认会重试两次。

默认头部

我们自动发送 anthropic-version 头部，设置为 2023-06-01。

如果需要，你可以通过在每个请求或客户端对象上设置默认头部来覆盖它。

请注意，这样做可能会导致 SDK 中的类型不正确以及其他意外或未定义的行为。

from anthropic import Anthropic

client = Anthropic(
    default_headers={"anthropic-version": "My-Custom-Value"},
)

高级

日志记录

我们使用标准库的 logging 模块。

你可以通过设置环境变量 ANTHROPIC_LOG 为 debug 来启用日志记录。

$ export ANTHROPIC_LOG=debug

如何判断 `None` 是表示 `null` 还是缺失

在 API 响应中，一个字段可能明确为 null，或者完全缺失；在这两种情况下，它在这个库中的值都是 None。你可以使用 .model_fields_set 来区分这两种情况：

if response.my_field is None:
  if 'my_field' not in response.model_fields_set:
    print('收到的 json 类似 {}，完全没有 "my_field" 键。')
  else:
    print('收到的 json 类似 {"my_field": null}。')

访问原始响应数据（例如头部）

可以通过在任何 HTTP 方法调用前加上 .with_raw_response. 前缀来访问"原始"响应对象，例如：

from anthropic import Anthropic

client = Anthropic()
response = client.messages.with_raw_response.create(
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": "你好，Claude",
    }],
    model="claude-3-opus-20240229",
)
print(response.headers.get('X-My-Header'))

message = response.parse() # 获取 messages.create() 会返回的对象 print(message.content)


这些方法返回一个 [`LegacyAPIResponse`](https://github.com/anthropics/anthropic-sdk-python/tree/main/src/anthropic/_legacy_response.py) 对象。这是一个遗留类，因为我们在下一个主要版本中会对其进行略微修改。

对于同步客户端，除了 `content` 和 `text` 将变为方法而非属性外，大部分内容将保持不变。在异步客户端中，所有方法都将是异步的。

我们将提供迁移脚本，整体迁移过程应该会很顺利。

#### `.with_streaming_response`

上述接口在发出请求时会立即读取完整的响应体，这可能并不总是你想要的。

要流式传输响应体，请使用 `.with_streaming_response`，它需要一个上下文管理器，并且只有在你调用 `.read()`、`.text()`、`.json()`、`.iter_bytes()`、`.iter_text()`、`.iter_lines()` 或 `.parse()` 时才会读取响应体。在异步客户端中，这些都是异步方法。

因此，`.with_streaming_response` 方法返回一个不同的 [`APIResponse`](https://github.com/anthropics/anthropic-sdk-python/tree/main/src/anthropic/_response.py) 对象，而异步客户端返回一个 [`AsyncAPIResponse`](https://github.com/anthropics/anthropic-sdk-python/tree/main/src/anthropic/_response.py) 对象。

```python
with client.messages.with_streaming_response.create(
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": "你好，Claude",
        }
    ],
    model="claude-3-opus-20240229",
) as response:
    print(response.headers.get("X-My-Header"))

    for line in response.iter_lines():
        print(line)

需要使用上下文管理器以确保响应能够可靠地关闭。

发送自定义/未记录的请求

该库已针对便捷访问已记录的 API 进行了类型设置。

如果你需要访问未记录的端点、参数或响应属性，仍然可以使用该库。

未记录的端点

要向未记录的端点发送请求，你可以使用 client.get、client.post 和其他 HTTP 动词方法。发送请求时，客户端的选项（如重试）将被遵守。

import httpx

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

print(response.headers.get("x-foo"))

未记录的请求参数

如果你想明确发送额外参数，可以使用 extra_query、extra_body 和 extra_headers 请求选项。

未记录的响应属性

要访问未记录的响应属性，你可以像 response.unknown_prop 这样访问额外字段。你还可以使用 response.model_extra 以字典形式获取 Pydantic 模型上的所有额外字段。

配置 HTTP 客户端

你可以直接覆盖 httpx 客户端以根据你的用例进行自定义，包括：

支持代理
自定义传输
其他高级功能

from anthropic import Anthropic, DefaultHttpxClient

client = Anthropic(
    # 或使用 `ANTHROPIC_BASE_URL` 环境变量
    base_url="http://my.test.server.example.com:8083",
    http_client=DefaultHttpxClient(
        proxies="http://my.test.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

你还可以使用 with_options() 在每个请求的基础上自定义客户端：

client.with_options(http_client=DefaultHttpxClient(...))

管理 HTTP 资源

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

版本控制

本包通常遵循 SemVer 约定，但某些不向后兼容的更改可能会作为次要版本发布：

只影响静态类型而不破坏运行时行为的更改。
对库内部的更改，这些内部技术上是公开的，但并非旨在或记录供外部使用。（如果你依赖于此类内部功能，请在 GitHub 上开一个 issue 告诉我们）。
我们预计不会影响绝大多数用户的更改。

我们非常重视向后兼容性，并努力确保你可以依赖平稳的升级体验。

我们热切期待你的反馈；请就问题、错误或建议在 issue 中提出。

要求

Python 3.7 或更高版本。

anthropic-sdk-python