Python SDK

Anthropic Python SDK обеспечивает удобный доступ к Claude API из приложений на Python. Он поддерживает как синхронные, так и асинхронные операции, «streaming» (потоковую передачу), а также интеграции с Amazon Bedrock, Claude Platform на AWS, Google Cloud и Microsoft Foundry.

Установка

pip install anthropic

Для интеграций с конкретными платформами или повышения производительности асинхронных операций установите пакет с дополнительными зависимостями:

# Для поддержки Amazon Bedrock
pip install "anthropic[bedrock]"

# Для поддержки Google Cloud
pip install "anthropic[vertex]"

# Для поддержки Claude Platform на 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; в разделе Выбор рабочего пространства показан вариант настройки для отдельного запроса в этом 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 для лучшей конкурентности

Для повышения производительности асинхронных операций вы можете использовать HTTP-бэкенд aiohttp вместо используемого по умолчанию 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 также предоставляет вспомогательные средства потоковой передачи, которые используют контекстные менеджеры и обеспечивают доступ к накопленному тексту и итоговому сообщению:

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), который возвращает только итерируемый объект событий потока и использует меньше памяти (он не формирует для вас итоговый объект сообщения).

Подсчёт токенов

Вы можете увидеть точное использование для конкретного запроса через свойство ответа usage:

message = client.messages.create(...)
print(message.usage)
# Usage(input_tokens=25, output_tokens=13)

Вы также можете подсчитать токены перед выполнением запроса:

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 генерирует схему инструмента на основе сигнатуры функции и строки документации:

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 и те же параметры запроса params, что и стандартный Messages API:

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"}],
            },
        },
    ]
)

Получение результатов пакета

После того как пакет сообщений обработан, на что указывает .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.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)

Коды ошибок следующие:

Идентификаторы запросов

Подробнее об отладке запросов см. в разделе Идентификатор запроса.

Все объекты ответов в 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.

SDK выбросит ValueError, если ожидается, что непотоковый запрос займёт более приблизительно 10 минут. Передача stream=True или переопределение параметра timeout на уровне клиента или запроса отключает эту ошибку.

Если ожидаемая задержка непотокового запроса превышает тайм-аут, клиент разорвёт соединение и повторит попытку, не получив ответа.

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 автоматически отправляет заголовок anthropic-version со значением 2023-06-01.

При необходимости вы можете переопределить его, установив заголовки по умолчанию для объекта клиента или для отдельного запроса.

# Задайте заголовки по умолчанию для всех запросов клиента
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")

Расширенное использование

Доступ к необработанным данным ответа (например, заголовкам)

К «необработанному» объекту Response, возвращаемому httpx2, можно получить доступ через свойство .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, который требует контекстного менеджера и считывает тело ответа только после вызова .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)

Контекстный менеджер необходим для того, чтобы ответ был надёжно закрыт.

Логирование

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. Вы также можете получить все дополнительные поля модели Pydantic в виде словаря с помощью response.model_extra.

Настройка HTTP-клиента

SDK отправляет запросы с помощью httpx2, API-совместимого форка httpx. Чтобы настроить 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, такие как HTTPXClientInstrumentor из OpenTelemetry, интеграция httpx в Sentry, respx или pytest-httpx, по умолчанию не видят запросы SDK. Чтобы использовать их, вызовите httpx2.alias_httpx() один раз при запуске, до того как что-либо импортирует httpx. В результате import httpx будет разрешаться в httpx2 для всего процесса.

Управление HTTP-ресурсами

По умолчанию библиотека закрывает базовые HTTP-соединения, когда клиент удаляется сборщиком мусора. При желании вы можете вручную закрыть клиент с помощью метода .close() или использовать контекстный менеджер, который закрывает его при выходе.

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

# HTTP-клиент закрывается автоматически

Бета-функции

Бета-функции доступны до общего выпуска для получения ранней обратной связи и тестирования новой функциональности. Вы можете проверить доступность всех возможностей и инструментов Claude в обзоре разработки с Claude.

Вы можете получить доступ к большинству бета-функций API через свойство 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:

Клиент AnthropicAWS находится в бета-версии. Передайте workspace_id в конструктор или установите переменную окружения ANTHROPIC_AWS_WORKSPACE_ID.

Используйте AnthropicBedrockMantle для новых проектов; AnthropicBedrock остаётся для существующих приложений, использующих API Bedrock InvokeModel.

Семантическое версионирование

Этот пакет в целом следует соглашениям SemVer, хотя некоторые обратно несовместимые изменения могут выпускаться как минорные версии:

1. Изменения, которые затрагивают только статические типы, не нарушая поведение во время выполнения.
2. Изменения внутренних компонентов библиотеки, которые технически являются публичными, но не предназначены и не документированы для внешнего использования.
3. Изменения, которые, как ожидается, на практике не затронут подавляющее большинство пользователей.

Определение установленной версии

Если вы обновились до последней версии, но не видите ожидаемых новых функций, вероятно, ваша среда Python всё ещё использует более старую версию. Вы можете определить используемую во время выполнения версию с помощью:

print(anthropic.__version__)

Дополнительные ресурсы

• Репозиторий GitHub
• Справочник API
• Потоковая передача сообщений
• Использование инструментов с Claude