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 约定，但某些不向后兼容的更改可能会作为次要版本发布：

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

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

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

要求

Python 3.7 或更高版本。